Redis分布式锁

　　分布式应用在逻辑 处理中经常会遇到并发问题。如一个操作要修改用户的状态，需要先读出用户的状态，再在内存中进行修改，改完了再还回去。但是如果有多个这样的操作同时进行，就会出现并发问题，，因为读取和修改这两个操作不是原子操作（原子操作是指不会被线程调度机制打断的操作，原子操作一旦开始，就会一直运行结束，中间不会有任何线程切换。）

 分布式锁的原理

　　分布式锁本质上就是在Redis里面占一个坑，当别的线程也要来占坑时，发现已经被占了，只好放弃或者稍后再试。占坑一般使用setnx（set if not exists）指令，只允许一个客户端占坑，先来先占，完成操作再调用del命令释放坑。

　　需要注意：

　　　　1）、一定要用 SET key value [EX seconds|PX milliseconds|KEEPTTL] [NX|XX] [GET]   执行，如SET key value EX 60 NX来保证setnx和expire指令原子执行

　　　　2）、value要具有唯一性。这个是为了在解锁的时候，需要验证value是和加锁的一致才删除key

死锁问题：

　　如果逻辑执行中出现异常，del指令没有被调用，导致锁不能释放，就会造成死锁问题，锁永远得不到释放。

　　因此需在拿到锁时设置过期时间，这样即使出现异常也能保证在到期之后释放锁。

　　redis1.x版本需要用两个指令来获取锁和设置过期时间，分别是setnx和expire，但是setnx和expire是两条指令而不是原子指令，如果在setnx和expire两个指令之间

服务器挂掉了也会导致expire得不到执行，也会造成死锁。解决这个问题需要使用lua脚本来使这两个指令变成一个原子操作。

　　Redis2.8版本中加入了set指令的拓展参数，可以使得setnx和expire指令可以原子执行。如：SET key value EX 60 NX

　　

 超时问题：

　　如果在加锁后的逻辑处理执行时间太长，以至于超过了锁的超时机制，就会出现问题，因为这个时候，A线程持有的锁过期了，但A线程的逻辑还未处理

　　完，这时候B线程获得了锁，仍然存在并发问题。如果这时A线程执行完成了任务，然后去释放锁，这时释放的就是B线程创建和持有的锁。

　　为了避免这个问题：

　　1、Redis分布式锁不要用来执行较长时间的任务

　　2、加锁的value是个特殊值（如uuid），只有持有锁的线程知道，释放锁前先对比value是否相同，相同的话再释放锁。

　　　为了防止对比时，释放锁前当前锁超时，其他线程再创建新的锁，需要使获取锁value和释放锁是一个原子操作，用lua脚本来解决。

分布式锁之Redlock算法

　　集群环境下分布式锁的问题：

　　在Sentinel集群中，当主节点挂掉时，从节点会取而代之，但客户端上并没有明显感知。比如第一个客户端在主节点上申请成功了一把锁，但是

这把锁还没有来得及同步到从节点，主节点突然挂掉了，然后从节点变成了主节点，这个新的主节点内部没有这个锁，所以当另一个客户端过来请

求加锁时，立即就批准了。这样导致系统中同样一把锁被两个客户端同时持有，不安全性由此产生。

　　这种不安全仅在主从发生failover（失效接管）的情况下才会产生，持续的时间极短，业务系统多数情况下可以容忍。

　　Redlock的出现就是为了解决这个问题。

　　要使用Redlock，需要提供多个Redis实例，这些实例之前相互独立，没有主从关系。同很多分布式算法一样，Redlock也使用  “大多数机制“；

　　加锁时，它会向过半节点发送  set（key，value，nx=True，ex=xxx）指令，只要过半节点set成功，就认为加锁成功。释放锁时，需要向所有节点发送del指令。

不过Redlock算法还需要考虑出错重试、时钟漂移（时钟抖动频率在10hz一下）等很多细节问题。同时因为Redlock需要向多个节点进行读写，意味着其相比单实例Redis的性能会下降一些

　　Redlock使用场景：非常看重高可用性，即使Redis挂了一台也完全不受影响就使用Redlock。代价是需要更多的Redis实例，性能也会下降，需要引入额外的library，运维上也需要区别对待。

分布式锁之过期时间到了锁失效但任务还未执行完毕

 　　某个线程在申请分布式锁的时候，为了应对极端情况，比如机器宕机，那么这个锁就一直不能被释放。一个比较好的解决方案是，申请锁的时候，预估一个程序的执行时间，然后给锁设置一个超时时间，这样，即使机器宕机，锁也能自动释放。

　　但是这也带来了一个问题，就是在有时候负载很高，任务执行的很慢，锁超时自动释放了任务还未执行完毕，这时候其他线程获得了锁，导致程序执行的并发问题。

　　对这种情况的解决方案是：在获得锁之后，就开启一个守护线程，定时去查询Redis分布式锁的到期时间，如果发现将要过期了，就进行续期。

Redission

　　git官方地址：https://github.com/redisson/redisson

　　Redisson是一个企业级的开源Redis Client，也提供了分布式锁的支持

　　上面说了为了避免死锁问题，需要加锁的同时设置有效期。但是又存在超时问题，如果超时锁失效了，任务还未执行完毕，其他线程可能获得锁，又会造成安全问题。

　　Redisson分布式锁的实现：

Config config = new Config();
config.useClusterServers()
.addNodeAddress("redis://ip:port")
.addNodeAddress("redis://ip:port")
...;

RedissonClient redisson = Redisson.create(config);

RLock lock = redisson.getLock("key");
lock.lock(); // 获得锁
lock.unlock(); // 释放锁

只需要通过它的api中的lock和unlock即可完成分布式锁，具体细节交给Redisson去实现：

　　1）redisson所有指令都通过lua脚本执行，redis支持lua脚本原子性执行

　　2）redisson设置一个key的默认过期时间为30s,如果某个客户端持有一个锁超过了30s怎么办？

　　　　redisson中有一个watchdog的概念，翻译过来就是看门狗，它会在你获取锁之后，每隔10秒检查一下锁是否释放，如果没有释放，则帮你把key的超时时间重新设为30s这样的话，就算一直持有锁也不会出现key过期了，其他线程获取到锁的问题了。

　　　　redisson的“看门狗”逻辑保证了没有死锁发生。

　　　　(如果机器宕机了，看门狗也就没了。此时就不会延长key的过期时间，到了30s之后就会自动过期了，其他线程可以获取到锁)

　　Redission实践

　　引入依赖

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

　　1、配置连接Redis

// 1. Create config object
Config config = new Config();
config.useClusterServers()
       // use "rediss://" for SSL connection
      .addNodeAddress("redis://127.0.0.1:7181");

// or read config from file
config = Config.fromYAML(new File("config-file.yaml")); 

　　2、创建Redisson实例 

// 2. Create Redisson instance

// Sync and Async API
RedissonClient redisson = Redisson.create(config);

// Reactive API
RedissonReactiveClient redissonReactive = Redisson.createReactive(config);

// RxJava2 API
RedissonRxClient redissonRx = Redisson.createRx(config);

　　3、获取map缓存，通过Redisson封装的ConcurrentMap的实现

// 3. Get Redis based implementation of java.util.concurrent.ConcurrentMap
RMap<MyKey, MyValue> map = redisson.getMap("myMap");

RMapReactive<MyKey, MyValue> mapReactive = redissonReactive.getMap("myMap");

RMapRx<MyKey, MyValue> mapRx = redissonRx.getMap("myMap");

　　4、获取分布式锁，通过Redisson封装的Lock的实现

// 4. Get Redis based implementation of java.util.concurrent.locks.Lock
RLock lock = redisson.getLock("myLock");

RLockReactive lockReactive = redissonReactive.getLock("myLock");

RLockRx lockRx = redissonRx.getLock("myLock");

　　5、获取

// 5. Get Redis based implementation of java.util.concurrent.ExecutorService
RExecutorService executor = redisson.getExecutorService("myExecutorService");

// over 50 Redis based Java objects and services ...

附录：

SETNX：SET if Not eXists。当key已经存在时，什么都不做。

SET key value [EX seconds|PX milliseconds|KEEPTTL] [NX|XX] [GET]

　　Options

　　　　EX seconds -- Set the specified expire time, in seconds.　　　　

　　　　PX milliseconds -- Set the specified expire time, in milliseconds.

　　　　NX -- Only set the key if it does not already exist.

　　　　XX -- Only set the key if it already exist.

　　　　KEEPTTL -- Retain the time to live associated with the key.

　　　　GET -- Return the old value stored at key, or nil when key did not exist.

官方文档：https://redis.io/commands/setnx