redis之分布式锁(四)

一.为什么需要分布式锁

在分布式的情况下，比如：会员服务，可能会有多个，然而单个服务的加锁行为，只能锁住一个服务，这样就会出现问题。

二.分布式锁的基本原理

三.分布式锁的实现形式

分布式锁重点：保证 加锁 和 解锁 的原子性。

1.使用 redis的set命令

set key value [EX seconds][PX milliseconds][NX|XX]
// EX：过期时间单位是秒。
// PX：过期时间单位是毫秒。
// NX：key不存在，才会放。
// XX：key存在，才会放。

实现逻辑1
使用 
1）.获取锁（set lock ikun nx）
2）.设置过期时间（Expire lock 60）
3）.处理业务
4）.释放锁
问题：如果在获取锁之后，如果断网了，那么就不会设置过期时间，就会造成死锁。

实现逻辑2
解决实现逻辑1的问题
1）.获取锁和设置过期时间同时（原子性）set lock ikun ex 20 nx
2）.处理业务
3）.释放锁
问题：如果业务执行的时间大于设置的过期时间，然后锁已经释放，可能被别的线程获取到了，然后咱们再去执行删锁的时候，就会删除别人的锁。

实现逻辑3
解决实现逻辑2的问题
1）.获取锁和设置过期时间同时（原子性）set lock uuid ex 20 nx
2）.处理业务
3）.判断是否是自己的锁（使用UUID）
4）.释放锁
缺点：判断是自己的锁，然后准备去释放锁，然后锁过期自动释放，在3）完成，4）还没执行的时候，有线程又重新设置值，然后我们再去删锁，那我们就删了别人的锁。

实现逻辑4（使用redis和Lua脚本实现）
解决实现逻辑3的问题
1）.获取锁和设置过期时间同时（原子性）set lock uuid ex 20 nx
2）.处理业务
3）.Lua脚本释放锁（原子性）

## java代码中使用Lua脚本
	//使用get命令取到key然后和传进来的key作对比，相等就删掉返回1，否则就返回0。（要么全成功，要么全失败）
	String lua = "if redis.call("get",KEYS[1]) == ARGV[1] then return redis.call("del",KEYS[1]) else return 0 end";
	//执行脚本  Arrays.asList("key")相当于KEYS[1],uuid相当于ARGV[1]
	Integer result = redisTemplate.execute(new DefaultRedisScript<Integer>(lua,Integer.class),Arrays.asList("key"),uuid);

四.使用Redisson

1.添加pom文件

<dependency>
    <groupId>org.redisson</groupId>
    <artifactId>redisson</artifactId>
    <version>3.25.2</version>
</dependency>

2.配置redisson

@Configuration
public class RedissonConfig {
	@Bean(destroyMethod="shutdown")
	public RedissonClient redisson() throws IOException {
		//1.创建配置
		Config config = new Config();
		//2.配置单机的redisson连接  redis://必须加，如果启用SSL(安全连接)，必须加rediss://
		config.useSingleServer()
				.setAddress("redis://redis地址:端口号")
				.setPassword("redis密码")
                .setDatabase(0);//使用redis的第几个库
       //3.返回RedissonClient对象
       RedissonClient redissonClient = Redisson.create(config);
       return redissonClient; 
	}
}

五.Redisson-lock（重入锁）

1.lock的两大特点

//Lock锁实例
RLock lock = redisson.getLock("ikunLock");
//最常见用法（会自动续期）
lock.lock();
或
//手动设置过期时间（不会自动续期）
lock.lock(22,TimeUnit.SECONDS)

//尝试加锁，最多等待100S,上锁以后10S自动解锁
boolean flag = lock.tryLock(100,10,TimeUnit.SECONDS)

（1）看门狗-锁在运行期间，自动进行续期。
（2）为了防止死锁，默认是30S的过期时间。

2.看门狗的原理

看门狗的原理： 只要占锁成功，会开启一个定时任务，锁超时就会重新给默认的过期时间30S，续期时间 = 默认的过期时间/3 = 10S。默认过期时间也就是看门狗时间。

六.Redisson-lock（读写锁）

分布式可重入读写锁允许有多个读锁和一个写锁：写锁没释放，读就必须等待。可以包装拿到的数据一定是最新的值。
RReadWriteLock rwLock = redisson.getReadWriteLock("ikunLock");
//读锁
rwLock.readLock().lock();
//写锁
rwLock.writeLock().lock();
//也可以设置自动时间和最大尝试时间

1、读 + 读： 相当于无所。
2、写 + 读： 等待写锁释放。
3、读 + 写 ： 写需要等待。
4、写 + 写 ： 等待上一个写锁释放。
只要有写锁，都需要等待。

七.闭锁

举一个例子：坤哥开演唱会，只有听演唱会的小黑子走了，保安才能关门。
public String guanmen(){
	RCountDownLatch latch = redisson.getCountDownLatch("ikun");
	//一共两个人
	latch.trySetCount(2);
	//等待
	latch.await();
	return "小黑子走了，开心";
}

public String sanhui(){
	RCountDownLatch latch = redisson.getCountDownLatch("ikun");
	//走出演唱会
	latch.countDown(1);
	return "见不到哥哥了，难受";
}

执行guanmen()方法，然后必须执行两遍sanhui()方法，guanmen()方法才会执行完毕，不然会一直等待。

八.redisson信号量

举一个例子：去看坤哥开演唱会，找地方停车，停车场有2.5*2个位置。（可以进行限流）
public String stop(){
	RSemaphore stop = redisson.getSemaphore("ikun");
	//尝试获取车位
	boolean flag = stop.tryAcquire();
	if(flag){
	//有车位，停车
	}else{
		return "没有车位";
	}
	return "停车成功，开心";
}

public String out(){
	RSemaphore out = redisson.getSemaphore("ikun");
	//开出车库，释放车位
	out.release();
	return "开出停车场";
}

九.保证数据一致性

1.先写缓存，在写数据库
缺点： 如果写缓存成功，数据库失败。那么数据就是脏数据。

2.先写数据库，在写缓存
缺点： 数据库成功，缓存失败。缓存数据就是旧数据。

3.先删缓存，再写数据库
缺点： 在删除缓存后，网络卡顿还没来的急去写数据库，那么再次取缓存，那还是旧数据。
解决： 缓存双删，网络卡顿之后，写入数据库，然后间隔一段时间再去删除缓存。

4.先写数据库，再删缓存
缺点： 数据库写入之后，还没来的急删除缓存，就被读取到旧值。（但是正常情况下读操作也比写操作更快，所以比较推荐此操作，相对其他方案，问题出现的概率要小）

删除缓存失败：
①使用定时任务，存放进数据库，进行重试。重试到一定次数记录失败，等待后续处理，任意重试成功就返回成功。实时性没那么高。

②使用mq，存放到mq消息里面，进行处理，重试到一定次数，加入到死信队列。实时性比较高。

③使用canal（阿里开源的中间件，可以当做是数据的从服务器），数据库改变，canal就会记录什么数据改变，然后再去更新缓存。还可以解决数据异构问题：就是淘宝每个人推荐的东西都不一样，因为canal记录了你的浏览记录然后结合商品表去做计算，然后给你推送你需要的东西。

更多配置请看redisson文档