redis 事务
在 redis 中，事务并不具有原子性和隔离性
redis单条命令是保证原子性的
但是 redis 事务时不保证原子性的。
redis 事务的本质，是一组命令的集合
如果你开启事务，你接下来执行的命令会
按照顺序一个个进入队列，然后一次性全部执行，
如果中间有命令发生错误，并不影响其他命令执行
所以 redis 事务具有一次性，顺序性，和排他性
但不具备 隔离级别的概念，所以他没有 幻读，脏读 和不可重复读的概念。
如果创建事务，命令会在队列中，这个时候命令并不会被执行，只有在提交事务后(exec) 才会一起执行。
例子:
- multi 开启事务
- 命令入队
- exec 提交事务
127.0.0.1:6379> multi # 开启事务OK127.0.0.1:6379> set k1 v1QUEUED # 命令入队127.0.0.1:6379> set k2 v2QUEUED127.0.0.1:6379> get k2QUEUED127.0.0.1:6379> set k3 v3QUEUED127.0.0.1:6379> exec # 提交事务1) OK2) OK3) "v2"4) OK
放弃事务
- 一旦放弃事务，队列中的命令都不会被执行
DISCARD
127.0.0.1:6379> multiOK127.0.0.1:6379> set k4 v4QUEUED127.0.0.1:6379> set k5 v5QUEUED127.0.0.1:6379> DISCARDOK127.0.0.1:6379> get k4(nil)
redis 事务中出现错误怎么办？
这种错误分为 编译错误 和 运行错误
如果是编译错误的话，那么事务中所有的命令都不会被执行
如果是运行错误，该条命令执行失败，队列中的其他命令被正常执行
编译错误：我们随便写一条不存在的命令
127.0.0.1:6379> multiOK127.0.0.1:6379> set k6 v6QUEUED127.0.0.1:6379> setabcsdwdwdw k7 v7(error) ERR unknown command 'setabcsdwdwdw' # 可以看到，编译时命令就报错了127.0.0.1:6379> set k8 v8QUEUED127.0.0.1:6379> exec(error) EXECABORT Transaction discarded because of previous errors. # 最后提交事务，会发现，因为出现编译错误，事务提交失败127.0.0.1:6379> get k6 # 事务提交失败，所以获取不到值(nil)
运行时错误：我们将一个字符串类型的value 自增
毫无疑问他会报错。
127.0.0.1:6379> multiOK127.0.0.1:6379> set views v1QUEUED127.0.0.1:6379> incr views # 这里一定会报运行时错误，因为它语法没错QUEUED127.0.0.1:6379> set users 100QUEUED127.0.0.1:6379> exec1) OK2) (error) ERR value is not an integer or out of range # 只在该条命令报错3) OK127.0.0.1:6379> get users # 并没有影响该条命令提交"100"
redis 中的乐观锁
同样的，我们在多线程的情况下，也需要给redis 加锁，防止数据的不一致性。
在redis 我们使用的是乐观锁。
首先要使用 watch 来监控一个 key 或多个key，然后在更新数据的时候，它会去判断是否有人修改过这个值，如果有人修改过，那么事务提交失败，只能再次监控和创建事务。
不加锁的情况
线程一：消费40元，对 money 和 out 进行修改
127.0.0.1:6379> set money 100OK127.0.0.1:6379> set out 0OK127.0.0.1:6379> multiOK127.0.0.1:6379> decrby money 40QUEUED127.0.0.1:6379> incrby out 40QUEUED127.0.0.1:6379> exec # 此时第二个线程以及 把 money值修改成了 1000. 1000 - 40 = 9601) (integer) 9602) (integer) 40127.0.0.1:6379>
线程二：在 线程一 还没有提交事务的时候对 money 进行修改
127.0.0.1:6379> set money 1000OK127.0.0.1:6379> get money"1000"
这样不加锁的时候，会发现最后输出的 money 和 out 变成了 960 和 40，并不是我们期望的 60 和 40
所以，为了避免这种情况的发生我们需要对，money字段 加上乐观锁
127.0.0.1:6379> set money 100OK127.0.0.1:6379> set out 0OK127.0.0.1:6379> watch moneyOK127.0.0.1:6379> multiOK127.0.0.1:6379> decrby money 40QUEUED127.0.0.1:6379> incrby out 40QUEUED127.0.0.1:6379> exec(nil) # 会发现事务提交的结果为 nil，这代表监控的值发生了变化127.0.0.1:6379> get money # 第二个线程修改后的结果。"1000"127.0.0.1:6379> unwatch # 记得最后要关闭监控OK
线程二：
127.0.0.1:6379> set money 1000OK
通过 jedis 操作redis
通过 jedis 操作redis 没什么复杂的，
所有的api 命令 与redis 中的命令都一致。
public static void main(String[] args) {//连接 redisJedis jedis = new Jedis("127.0.0.1", 6379);//测试连接System.out.println(jedis.ping());//添加 keyString set = jedis.set("username", "hello");System.out.println(set);//获取所有的keySet<String> keys = jedis.keys("*");keys.forEach(System.out::println);}
jedis 中的事务
public static void main(String[] args) {
//连接 redis
Jedis jedis = new Jedis("127.0.0.1", 6379);
Jedis jedis2 = new Jedis("127.0.0.1", 6379);
//测试连接
System.out.println(jedis.ping());
jedis.set("money","100");
jedis.set("out","0");
new Thread( () -> {
Transaction multi = jedis.multi();
try {
multi.decrBy("money",20);
multi.incrBy("out",40);
List<Object> exec = multi.exec();
exec.forEach(System.out::println);
} catch (Exception e) {
multi.discard()
System.out.println(e);
}
}).start();
new Thread(() -> {
jedis2.set("money", "1000");
String result = jedis2.get("money");
System.out.println("jedis2 ->" + result);
}).start();
}
jedis 中的乐观锁
public static void main(String[] args) {
//连接 redis
Jedis jedis = new Jedis("127.0.0.1", 6379);
Jedis jedis2 = new Jedis("127.0.0.1", 6379);
//测试连接
System.out.println(jedis.ping());
jedis.set("money","100");
jedis.set("out","0");
jedis.watch("money");
new Thread( () -> {
Transaction multi = jedis.multi();
try {
multi.decrBy("money",20);
multi.incrBy("out",40);
// int a = 100 / 0;
List<Object> exec = multi.exec();
if (exec != null){
System.out.println(exec);
exec.forEach(System.out::println);
} else {
System.out.println("监控的 key 值发生改变，请重新提交事务");
}
} catch (Exception e) {
multi.discard();
e.printStackTrace();
} finally {
jedis.close();
}
}).start();
new Thread(() -> {
jedis2.set("money", "1000");
String result = jedis2.get("money");
System.out.println("jedis2 ->" + result);
jedis2.close();
}).start();
}
springboot 中操作redis
springboot 中使用 spring-data-redis 操作redis
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>
在 springboot 2.0之后，原来使用的 jedis 变成了 lettcue
jedis：采用直连，如果多个线程操作的话，需要创建 jedis pool连接池，更像 bio 模式
lettcue：底层采用 netty，实例可以再多个线程中共享，存在线程不安全的问题，更像nio 模式
@Autowired
RedisTemplate<String, String> redisTemplate;
@Test
void contextLoads() {
redisTemplate.opsForValue().set("user","哈喽");
// redisTemplate.opsForValue(); //操作字符串
// redisTemplate.opsForList(); //操作 list
// redisTemplate.opsForSet(); //操作 set
// redisTemplate.opsForHash();//操作 hash
String user = redisTemplate.opsForValue().get("user");
System.out.println(user);
}
如果要将对象数据放入到 redis 中的时候，记得将对象序列化，spring-data-redis自动采用的是jdk的序列化。
但是现在我们一般都会使用json来序列化
我们可以在进行一些相关配置
RedisConfig
@Configuration
public class RedisConfig {
@Bean
@SuppressWarnings("all")
public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory factory) {
// 我们为了自己开发方便，一般直接使用 <String, Object>
RedisTemplate<String, Object> template = new RedisTemplate<String, Object>();
template.setConnectionFactory(factory);
// Json序列化配置
Jackson2JsonRedisSerializer jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer(Object.class);
ObjectMapper om = new ObjectMapper();
om.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
om.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);
jackson2JsonRedisSerializer.setObjectMapper(om);
// String 的序列化
StringRedisSerializer stringRedisSerializer = new StringRedisSerializer();
// key采用String的序列化方式
template.setKeySerializer(stringRedisSerializer);
// hash的key也采用String的序列化方式
template.setHashKeySerializer(stringRedisSerializer);
// value序列化方式采用jackson
template.setValueSerializer(jackson2JsonRedisSerializer);
// hash的value序列化方式采用jackson
template.setHashValueSerializer(jackson2JsonRedisSerializer);
template.afterPropertiesSet();
return template;
}
}
RedisUtil
当然，在开发时，我们不可能去使用redisTemplate封住的api，毕竟有些麻烦。
所以我们要自己进行再封装
@Component
public final class RedisUtil {
@Autowired
private RedisTemplate<String, Object> redisTemplate;
// =============================common============================
/**
* 指定缓存失效时间
* @param key 键
* @param time 时间(秒)
*/
public boolean expire(String key, long time) {
try {
if (time > 0) {
redisTemplate.expire(key, time, TimeUnit.SECONDS);
}
return true;
} catch (Exception e) {
e.printStackTrace();
return false;
}
}
/**
* 根据key 获取过期时间
* @param key 键 不能为null
* @return 时间(秒) 返回0代表为永久有效
*/
public long getExpire(String key) {
return redisTemplate.getExpire(key, TimeUnit.SECONDS);
}
/**
* 判断key是否存在
* @param key 键
* @return true 存在 false不存在
*/
public boolean hasKey(String key) {
try {
return redisTemplate.hasKey(key);
} catch (Exception e) {
e.printStackTrace();
return false;
}
}
/**
* 删除缓存
* @param key 可以传一个值 或多个
*/
@SuppressWarnings("unchecked")
public void del(String... key) {
if (key != null && key.length > 0) {
if (key.length == 1) {
redisTemplate.delete(key[0]);
} else {
redisTemplate.delete(CollectionUtils.arrayToList(key));
}
}
}
// ============================String=============================
/**
* 普通缓存获取
* @param key 键
* @return 值
*/
public Object get(String key) {
return key == null ? null : redisTemplate.opsForValue().get(key);
}
/**
* 普通缓存放入
* @param key 键
* @param value 值
* @return true成功 false失败
*/
public boolean set(String key, Object value) {
try {
redisTemplate.opsForValue().set(key, value);
return true;
} catch (Exception e) {
e.printStackTrace();
return false;
}
}
/**
* 普通缓存放入并设置时间
* @param key 键
* @param value 值
* @param time 时间(秒) time要大于0 如果time小于等于0 将设置无限期
* @return true成功 false 失败
*/
public boolean set(String key, Object value, long time) {
try {
if (time > 0) {
redisTemplate.opsForValue().set(key, value, time, TimeUnit.SECONDS);
} else {
set(key, value);
}
return true;
} catch (Exception e) {
e.printStackTrace();
return false;
}
}
/**
* 递增
* @param key 键
* @param delta 要增加几(大于0)
*/
public long incr(String key, long delta) {
if (delta < 0) {
throw new RuntimeException("递增因子必须大于0");
}
return redisTemplate.opsForValue().increment(key, delta);
}
/**
* 递减
* @param key 键
* @param delta 要减少几(小于0)
*/
public long decr(String key, long delta) {
if (delta < 0) {
throw new RuntimeException("递减因子必须大于0");
}
return redisTemplate.opsForValue().increment(key, -delta);
}
// ================================Map=================================
/**
* HashGet
* @param key 键 不能为null
* @param item 项 不能为null
*/
public Object hget(String key, String item) {
return redisTemplate.opsForHash().get(key, item);
}
/**
* 获取hashKey对应的所有键值
* @param key 键
* @return 对应的多个键值
*/
public Map<Object, Object> hmget(String key) {
return redisTemplate.opsForHash().entries(key);
}
/**
* HashSet
* @param key 键
* @param map 对应多个键值
*/
public boolean hmset(String key, Map<String, Object> map) {
try {
redisTemplate.opsForHash().putAll(key, map);
return true;
} catch (Exception e) {
e.printStackTrace();
return false;
}
}
/**
* HashSet 并设置时间
* @param key 键
* @param map 对应多个键值
* @param time 时间(秒)
* @return true成功 false失败
*/
public boolean hmset(String key, Map<String, Object> map, long time) {
try {
redisTemplate.opsForHash().putAll(key, map);
if (time > 0) {
expire(key, time);
}
return true;
} catch (Exception e) {
e.printStackTrace();
return false;
}
}
/**
* 向一张hash表中放入数据,如果不存在将创建
*
* @param key 键
* @param item 项
* @param value 值
* @return true 成功 false失败
*/
public boolean hset(String key, String item, Object value) {
try {
redisTemplate.opsForHash().put(key, item, value);
return true;
} catch (Exception e) {
e.printStackTrace();
return false;
}
}
/**
* 向一张hash表中放入数据,如果不存在将创建
*
* @param key 键
* @param item 项
* @param value 值
* @param time 时间(秒) 注意:如果已存在的hash表有时间,这里将会替换原有的时间
* @return true 成功 false失败
*/
public boolean hset(String key, String item, Object value, long time) {
try {
redisTemplate.opsForHash().put(key, item, value);
if (time > 0) {
expire(key, time);
}
return true;
} catch (Exception e) {
e.printStackTrace();
return false;
}
}
/**
* 删除hash表中的值
*
* @param key 键 不能为null
* @param item 项 可以使多个 不能为null
*/
public void hdel(String key, Object... item) {
redisTemplate.opsForHash().delete(key, item);
}
/**
* 判断hash表中是否有该项的值
*
* @param key 键 不能为null
* @param item 项 不能为null
* @return true 存在 false不存在
*/
public boolean hHasKey(String key, String item) {
return redisTemplate.opsForHash().hasKey(key, item);
}
/**
* hash递增 如果不存在,就会创建一个 并把新增后的值返回
*
* @param key 键
* @param item 项
* @param by 要增加几(大于0)
*/
public double hincr(String key, String item, double by) {
return redisTemplate.opsForHash().increment(key, item, by);
}
/**
* hash递减
*
* @param key 键
* @param item 项
* @param by 要减少记(小于0)
*/
public double hdecr(String key, String item, double by) {
return redisTemplate.opsForHash().increment(key, item, -by);
}
// ============================set=============================
/**
* 根据key获取Set中的所有值
* @param key 键
*/
public Set<Object> sGet(String key) {
try {
return redisTemplate.opsForSet().members(key);
} catch (Exception e) {
e.printStackTrace();
return null;
}
}
/**
* 根据value从一个set中查询,是否存在
*
* @param key 键
* @param value 值
* @return true 存在 false不存在
*/
public boolean sHasKey(String key, Object value) {
try {
return redisTemplate.opsForSet().isMember(key, value);
} catch (Exception e) {
e.printStackTrace();
return false;
}
}
/**
* 将数据放入set缓存
*
* @param key 键
* @param values 值 可以是多个
* @return 成功个数
*/
public long sSet(String key, Object... values) {
try {
return redisTemplate.opsForSet().add(key, values);
} catch (Exception e) {
e.printStackTrace();
return 0;
}
}
/**
* 将set数据放入缓存
*
* @param key 键
* @param time 时间(秒)
* @param values 值 可以是多个
* @return 成功个数
*/
public long sSetAndTime(String key, long time, Object... values) {
try {
Long count = redisTemplate.opsForSet().add(key, values);
if (time > 0)
expire(key, time);
return count;
} catch (Exception e) {
e.printStackTrace();
return 0;
}
}
/**
* 获取set缓存的长度
*
* @param key 键
*/
public long sGetSetSize(String key) {
try {
return redisTemplate.opsForSet().size(key);
} catch (Exception e) {
e.printStackTrace();
return 0;
}
}
/**
* 移除值为value的
*
* @param key 键
* @param values 值 可以是多个
* @return 移除的个数
*/
public long setRemove(String key, Object... values) {
try {
Long count = redisTemplate.opsForSet().remove(key, values);
return count;
} catch (Exception e) {
e.printStackTrace();
return 0;
}
}
// ===============================list=================================
/**
* 获取list缓存的内容
*
* @param key 键
* @param start 开始
* @param end 结束 0 到 -1代表所有值
*/
public List<Object> lGet(String key, long start, long end) {
try {
return redisTemplate.opsForList().range(key, start, end);
} catch (Exception e) {
e.printStackTrace();
return null;
}
}
/**
* 获取list缓存的长度
*
* @param key 键
*/
public long lGetListSize(String key) {
try {
return redisTemplate.opsForList().size(key);
} catch (Exception e) {
e.printStackTrace();
return 0;
}
}
/**
* 通过索引 获取list中的值
*
* @param key 键
* @param index 索引 index>=0时， 0 表头，1 第二个元素，依次类推；index<0时，-1，表尾，-2倒数第二个元素，依次类推
*/
public Object lGetIndex(String key, long index) {
try {
return redisTemplate.opsForList().index(key, index);
} catch (Exception e) {
e.printStackTrace();
return null;
}
}
/**
* 将list放入缓存
*
* @param key 键
* @param value 值
*/
public boolean lSet(String key, Object value) {
try {
redisTemplate.opsForList().rightPush(key, value);
return true;
} catch (Exception e) {
e.printStackTrace();
return false;
}
}
/**
* 将list放入缓存
* @param key 键
* @param value 值
* @param time 时间(秒)
*/
public boolean lSet(String key, Object value, long time) {
try {
redisTemplate.opsForList().rightPush(key, value);
if (time > 0)
expire(key, time);
return true;
} catch (Exception e) {
e.printStackTrace();
return false;
}
}
/**
* 将list放入缓存
*
* @param key 键
* @param value 值
* @return
*/
public boolean lSet(String key, List<Object> value) {
try {
redisTemplate.opsForList().rightPushAll(key, value);
return true;
} catch (Exception e) {
e.printStackTrace();
return false;
}
}
/**
* 将list放入缓存
*
* @param key 键
* @param value 值
* @param time 时间(秒)
* @return
*/
public boolean lSet(String key, List<Object> value, long time) {
try {
redisTemplate.opsForList().rightPushAll(key, value);
if (time > 0)
expire(key, time);
return true;
} catch (Exception e) {
e.printStackTrace();
return false;
}
}
/**
* 根据索引修改list中的某条数据
*
* @param key 键
* @param index 索引
* @param value 值
* @return
*/
public boolean lUpdateIndex(String key, long index, Object value) {
try {
redisTemplate.opsForList().set(key, index, value);
return true;
} catch (Exception e) {
e.printStackTrace();
return false;
}
}
/**
* 移除N个值为value
*
* @param key 键
* @param count 移除多少个
* @param value 值
* @return 移除的个数
*/
public long lRemove(String key, long count, Object value) {
try {
Long remove = redisTemplate.opsForList().remove(key, count, value);
return remove;
} catch (Exception e) {
e.printStackTrace();
return 0;
}
}
}
若有收获，就点个赞吧
0 人点赞
