Redis分布式锁的使用场景
分布式锁（set nx）
分布式锁redlock
基于zookeeper的分布式锁更安全吗？
总结
今天我们来聊一聊redis分布式锁。
首先大家可以先思考一个简单的问题，为什么要使用分布式锁？普通的jvm锁为什么不可以？
这个时候，大家肯定会吧啦吧啦想到一堆，例如java应用属于进程级，不同的ecs中部署相同的应用，他们之间相互独立。
所以，在分布式系统中，当有多个客户端需要获取锁时，我们需要分布式锁。此时，锁是保存在一个共享存储系统中的，可以被多个客户端共享访问和获取。
分布式锁（set nx）
知道了分布式锁的使用场景，我们来自己简单的实现下分布式锁：
public class indexcontroller { public string deductstock() { string lockkey = lock:product_101; //setnx 获取分布式锁 string clientid = uuid.randomuuid().tostring(); boolean result = stringredistemplate.opsforvalue().setifabsent(lockkey, clientid, 30, timeunit.seconds); //jedis.setnx(k,v) if (!result) { return error_code; } try { int stock = integer.parseint(stringredistemplate.opsforvalue().get(stock)); // jedis.get(stock) if (stock > 0) { int realstock = stock - 1; stringredistemplate.opsforvalue().set(stock, realstock + ); // jedis.set(key,value) system.out.println(扣减成功，剩余库存: + realstock); } else { system.out.println(扣减失败，库存不足); } } finally { //解锁 if (clientid.equals(stringredistemplate.opsforvalue().get(lockkey))) { stringredistemplate.delete(lockkey); } }}
以上代码简单的实现了一个扣减库存的业务逻辑，我们拆开来说下都做了什么事情：
1、首先声明了lockkey，表示我们需要set的keyname
2、其次uuid.randomuuid().tostring();生成该次请求的requestid，为什么需要生成这个唯一的uuid，后面在解锁的时候会说到
3、获取分布式锁，通过stringredistemplate.opsforvalue().setifabsent来实现，该语句的意思是如果存在该key则返回false，若不存在则进行key的设置，设置成功后返回true，将当前线程获取的uuid设置成value，给定一个锁的过期时间，防止该线程无限制持久锁导致死锁，也为了防止该服务器突然宕机，导致其他机器的应用无法获取该锁，这个是必须要做的设置，至于过期的时间，可以根据内层业务逻辑的执行时间来决定
4、执行内层的业务逻辑，进行扣库存的操作
5、业务逻辑执行完成后，走到finally的解锁操作，进行解锁操作时，首先我们来判断当前锁的值是否为该线程持有的，防止当前线程执行较慢，导致锁过期，从而删除了其他线程持有的分布式锁，对于该操作，我来举个例子：
时刻1：线程a获取分布式锁，开始执行业务逻辑
时刻2：线程b等待分布式锁释放
时刻3：线程a所在机器io处理缓慢、gc pause等问题导致处理缓慢
时刻4：线程a依旧处于block状态，锁过期
时刻5：线程b获取分布式锁，开始执行业务逻辑，此时线程a结束block，开始释放锁
时刻6：线程b处理业务逻辑缓慢，线程a释放分布式锁，但是此时释放的是线程b的锁，导致其他线程可以开始获取锁
看到这里，为什么每个请求需要requestid，并且在释放锁的情况下判断是否是当前的requestid是有必要的。
以上，就是一个简单的分布式锁的实现过程。但是你觉得上述实现还存在问题吗？
答案是肯定的。若是在判断完分布式锁的value与requestid之后，锁过期了，依然会存在以上问题。
那么有没有什么办法可以规避以上问题，让我们不需要去完成这些实现，只需要专注于业务逻辑呢？
我们可以使用redisson，并且redisson有中文文档，方便英文不好的同学查看（开发团队中有中国的jackygurui）。
接下来我们再把上述代码简单的改造下就可以规避这些问题：
public class indexcontroller { public string deductstock() { string lockkey = lock:product_101; //setnx 获取分布式锁 //string clientid = uuid.randomuuid().tostring(); //boolean result = stringredistemplate.opsforvalue().setifabsent(lockkey, clientid, 30, timeunit.seconds); //jedis.setnx(k,v) //获取锁对象 rlock redissonlock = redisson.getlock(lockkey); //加分布式锁 redissonlock.lock(); try { int stock = integer.parseint(stringredistemplate.opsforvalue().get(stock)); // jedis.get(stock) if (stock > 0) { int realstock = stock - 1; stringredistemplate.opsforvalue().set(stock, realstock + ); // jedis.set(key,value) system.out.println(扣减成功，剩余库存: + realstock); } else { system.out.println(扣减失败，库存不足); } } finally { //解锁 //if (clientid.equals(stringredistemplate.opsforvalue().get(lockkey))) { // stringredistemplate.delete(lockkey); //} //redisson分布式锁解锁 redissonlock.unlock(); }}
可以看到，使用redisson分布式锁会简单很多，我们通过redissonlock.lock()和redissonlock.unlock()解决了这个问题，看到这里，是不是有同学会问，如果服务器宕机了，分布式锁会一直存在吗，也没有去指定过期时间？
redisson分布式锁中有一个watchdog机制，即会给一个leasetime，默认为30s，到期后锁自动释放，如果一直没有解锁，watchdog机制会一直重新设定锁的过期时间，通过设置timetask，延迟10s再次执行锁续命，将锁的过期时间重置为30s。下面就从redisson.lock()的源码来看下：
lock的最终加锁方法：
rfuture trylockinnerasync(long leasetime, timeunit unit, long threadid, redisstrictcommand command) { internallockleasetime = unit.tomillis(leasetime); return commandexecutor.evalwriteasync(getname(), longcodec.instance, command, if (redis.call('exists', keys[1]) == 0) then + redis.call('hset', keys[1], argv[2], 1); + redis.call('pexpire', keys[1], argv[1]); + return nil; + end; + if (redis.call('hexists', keys[1], argv[2]) == 1) then + redis.call('hincrby', keys[1], argv[2], 1); + redis.call('pexpire', keys[1], argv[1]); + return nil; + end; + return redis.call('pttl', keys[1]);, collections.singletonlist(getname()), internallockleasetime, getlockname(threadid)); }
可以看到lua脚本中redis.call('pexpire', keys[1], argv[1]);对key进行设置，并给定了一个internallockleasetime，给定的internallockleasetime就是默认的加锁时间，为30s。
接下来我们在看下锁续命的源码：
private void scheduleexpirationrenewal(final long threadid) { if (!expirationrenewalmap.containskey(this.getentryname())) { timeout task = this.commandexecutor.getconnectionmanager().newtimeout(new timertask() { public void run(timeout timeout) throws exception { //重新设置锁过期时间 rfuture future = redissonlock.this.commandexecutor.evalwriteasync(redissonlock.this.getname(), longcodec.instance, rediscommands.eval_boolean, if (redis.call('hexists', keys[1], argv[2]) == 1) then redis.call('pexpire', keys[1], argv[1]); return 1; end; return 0;, collections.singletonlist(redissonlock.this.getname()), new object[]{redissonlock.this.internallockleasetime, redissonlock.this.getlockname(threadid)}); future.addlistener(new futurelistener() { public void operationcomplete(future future) throws exception { redissonlock.expirationrenewalmap.remove(redissonlock.this.getentryname()); if (!future.issuccess()) { redissonlock.log.error(can't update lock + redissonlock.this.getname() + expiration, future.cause()); } else { //获取方法调用的结果 if ((boolean)future.getnow()) { //进行递归调用 redissonlock.this.scheduleexpirationrenewal(threadid); } } } }); } //延迟 this.internallockleasetime / 3l 再执行run方法 }, this.internallockleasetime / 3l, timeunit.milliseconds); if (expirationrenewalmap.putifabsent(this.getentryname(), task) != null) { task.cancel(); } } }
从源码层可以看到，加锁成功后，会延迟10s执行task中的run方法，然后在run方法里面执行锁过期时间的重置，如果时间重置成功，则继续递归调用该方法，延迟10s后进行锁续命，若重置锁时间失败，则可能表示锁已释放，退出该方法。
以上，就是关于一个redis分布式锁的说明，看到这里，大家应该对分布式锁有一个大致的了解了。
但是尽管使用了redisson完成分布式锁的实现，对于分布式锁是否还存在问题，分布式锁真的安全吗？
一般的，线上的环境肯定使用redis cluster，如果数据量不大，也会使用的redis sentinal。那么就存在主从复制的问题，那么是否会存在这种情况，在主库设置了分布式锁，但是可能由于网络或其他原因导致数据还没有同步到从库，此时主库宕机，选择从库作为主库，新主库中并没有该锁的信息，其他线程又可以进行锁申请，造成了发生线程安全问题的可能。
为了解决这个问题，redis的作者实现了redlock，基于redlock的实现有很大的争论，并且现在已经弃用了，但是我们还是需要了解下原理，以及之后基于这些问题的解决方案。
基于 spring boot + mybatis plus + vue & element 实现的后台管理系统 + 用户小程序，支持 rbac 动态权限、多租户、数据权限、工作流、三方登录、支付、短信、商城等功能
项目地址：https://gitee.com/zhijiantianya/ruoyi-vue-pro
视频教程：https://doc.iocoder.cn/video/
分布式锁redlock
redlock是基于单redis节点的分布式锁在failover的时候会产生解决不了的安全性问题而产生的，基于n个完全独立的redis节点。
下面我来看下redlock获取锁的过程：
运行redlock算法的客户端依次执行下面各个步骤，来完成获取锁 的操作：
获取当前时间（毫秒数）。
按顺序依次向n个redis节点执行获取锁 的操作。这个获取操作跟前面基于单redis节点的获取锁 的过程相同，包含随机字符串my_random_value，也包含过期时间(比如px 30000，即锁的有效时间)。为了保证在某个redis节点不可用的时候算法能够继续运行，这个获取锁 的操作还有一个超时时间(time out)，它要远小于锁的有效时间（几十毫秒量级）。客户端在向某个redis节点获取锁失败以后，应该立即尝试下一个redis节点。这里的失败，应该包含任何类型的失败，比如该redis节点不可用，或者该redis节点上的锁已经被其它客户端持有
计算整个获取锁的过程总共消耗了多长时间，计算方法是用当前时间减去第1步记录的时间。如果客户端从大多数redis节点（>= n/2+1）成功获取到了锁，并且获取锁总共消耗的时间没有超过锁的有效时间(lock validity time)，那么这时客户端才认为最终获取锁成功；否则，认为最终获取锁失败。
如果最终获取锁成功了，那么这个锁的有效时间应该重新计算，它等于最初的锁的有效时间减去第3步计算出来的获取锁消耗的时间。
如果最终获取锁失败了（可能由于获取到锁的redis节点个数少于n/2+1，或者整个获取锁的过程消耗的时间超过了锁的最初有效时间），那么客户端应该立即向所有redis节点发起释放锁 的操作。
好了，了解了redlock获取锁的机制之后，我们再来讨论下redlock会有哪些问题：
问题一：
假设一共有5个redis节点：a, b, c, d, e。设想发生了如下的事件序列：
客户端1成功锁住了a, b, c，获取锁 成功（但d和e没有锁住）。
节点c崩溃重启了，但客户端1在c上加的锁没有持久化下来，丢失了。
节点c重启后，客户端2锁住了c, d, e，获取锁 成功。
这样，客户端1和客户端2同时获得了锁（针对同一资源）。
在默认情况下，redis的aof持久化方式是每秒写一次磁盘（即执行fsync），因此最坏情况下可能丢失1秒的数据。为了尽可能不丢数据，redis允许设置成每次修改数据都进行fsync，但这会降低性能。当然，即使执行了fsync也仍然有可能丢失数据（这取决于系统而不是redis的实现）。所以，上面分析的由于节点重启引发的锁失效问题，总是有可能出现的。为了应对这一问题，redis作者antirez又提出了延迟重启 (delayed restarts)的概念。也就是说，一个节点崩溃后，先不立即重启它，而是等待一段时间再重启，这段时间应该大于锁的有效时间(lock validity time)。这样的话，这个节点在重启前所参与的锁都会过期，它在重启后就不会对现有的锁造成影响。
关于redlock还有一点细节值得拿出来分析一下：在最后释放锁 的时候，antirez在算法描述中特别强调，客户端应该向所有redis节点发起释放锁 的操作。也就是说，即使当时向某个节点获取锁没有成功，在释放锁的时候也不应该漏掉这个节点。这是为什么呢？设想这样一种情况，客户端发给某个redis节点的获取锁 的请求成功到达了该redis节点，这个节点也成功执行了set操作，但是它返回给客户端的响应包却丢失了。这在客户端看来，获取锁的请求由于超时而失败了，但在redis这边看来，加锁已经成功了。因此，释放锁的时候，客户端也应该对当时获取锁失败的那些redis节点同样发起请求。实际上，这种情况在异步通信模型中是有可能发生的：客户端向服务器通信是正常的，但反方向却是有问题的。
所以，如果不进行延迟重启，或者对于同一个主节点进行多个从节点的备份，并要求从节点的同步必须实时跟住主节点，也就是说需要配置redis从库的同步策略，将延迟设置为最小（主从同步是异步进行的），通过min-replicas-max-lag(旧版本的redis使用min-slaves-max-lag)来设置主从库间进行数据复制时，从库给主库发送 ack 消息的最大延迟（以秒为单位），也就是说，这个值需要设置为0，否则都有可能出现延迟，但是这个实际上在redis中是不存在的，min-replicas-max-lag设置为0，就代表着这个配置不生效。redis本身是为了高效而存在的，如果因为需要保证业务的准确性而使用，大大降低了redis的性能，建议使用的别的方式。
问题二：
如果客户端长期阻塞导致锁过期，那么它接下来访问共享资源就不安全了（没有了锁的保护）。在redlock中还是存在该问题的。
虽然在获取锁之后redlock会去判断锁的有效性，如果锁过期了，则会再去重新拿锁。但是如果发生在获取锁之后，那么该有效性都得不到保障了。
在上面的时序图中，假设锁服务本身是没有问题的，它总是能保证任一时刻最多只有一个客户端获得锁。上图中出现的lease这个词可以暂且认为就等同于一个带有自动过期功能的锁。客户端1在获得锁之后发生了很长时间的gc pause，在此期间，它获得的锁过期了，而客户端2获得了锁。当客户端1从gc pause中恢复过来的时候，它不知道自己持有的锁已经过期了，它依然向共享资源（上图中是一个存储服务）发起了写数据请求，而这时锁实际上被客户端2持有，因此两个客户端的写请求就有可能冲突（锁的互斥作用失效了）。
初看上去，有人可能会说，既然客户端1从gc pause中恢复过来以后不知道自己持有的锁已经过期了，那么它可以在访问共享资源之前先判断一下锁是否过期。但仔细想想，这丝毫也没有帮助。因为gc pause可能发生在任意时刻，也许恰好在判断完之后。
也有人会说，如果客户端使用没有gc的语言来实现，是不是就没有这个问题呢？质疑者martin指出，系统环境太复杂，仍然有很多原因导致进程的pause，比如虚存造成的缺页故障(page fault)，再比如cpu资源的竞争。即使不考虑进程pause的情况，网络延迟也仍然会造成类似的结果。
总结起来就是说，即使锁服务本身是没有问题的，而仅仅是客户端有长时间的pause或网络延迟，仍然会造成两个客户端同时访问共享资源的冲突情况发生。
那怎么解决这个问题呢？martin给出了一种方法，称为fencing token。fencing token是一个单调递增的数字，当客户端成功获取锁的时候它随同锁一起返回给客户端。而客户端访问共享资源的时候带着这个fencing token，这样提供共享资源的服务就能根据它进行检查，拒绝掉延迟到来的访问请求（避免了冲突）。如下图：
在上图中，客户端1先获取到的锁，因此有一个较小的fencing token，等于33，而客户端2后获取到的锁，有一个较大的fencing token，等于34。客户端1从gc pause中恢复过来之后，依然是向存储服务发送访问请求，但是带了fencing token = 33。存储服务发现它之前已经处理过34的请求，所以会拒绝掉这次33的请求。这样就避免了冲突。
但是，对于客户端和资源服务器之间的延迟（即发生在算法第3步之后的延迟），antirez是承认所有的分布式锁的实现，包括redlock，是没有什么好办法来应对的。包括在我们到生产环境中，无法避免分布式锁超时。
在讨论中，有人提出客户端1和客户端2都发生了gc pause，两个fencing token都延迟了，它们几乎同时到达了文件服务器，而且保持了顺序。那么，我们新加入的判断逻辑，即判断fencing token的合理性，应该对两个请求都会放过，而放过之后它们几乎同时在操作文件，还是冲突了。既然martin宣称fencing token能保证分布式锁的正确性，那么上面这种可能的猜测也许是我们理解错了。但是martin并没有在后面做出解释。
问题三：
redlock对系统记时(timing)的过分依赖，下面给出一个例子（还是假设有5个redis节点a, b, c, d, e）：
客户端1从redis节点a, b, c成功获取了锁（多数节点）。由于网络问题，与d和e通信失败。
节点c上的时钟发生了向前跳跃，导致它上面维护的锁快速过期。
客户端2从redis节点c, d, e成功获取了同一个资源的锁（多数节点）。
客户端1和客户端2现在都认为自己持有了锁。
上面这种情况之所以有可能发生，本质上是因为redlock的安全性(safety property)对系统的时钟有比较强的依赖，一旦系统的时钟变得不准确，算法的安全性也就保证不了了。
但是作者反驳到，通过恰当的运维，完全可以避免时钟发生大的跳动，而redlock对于时钟的要求在现实系统中是完全可以满足的。哪怕是手动修改时钟这种人为原因，不要那么做就是了。否则的话，都会出现问题。
说了这么多关于redlock的问题，到底有没有什么分布式锁能保证安全性呢？我们接下来再来看看zookeeper分布式锁。
基于 spring cloud alibaba + gateway + nacos + rocketmq + vue & element 实现的后台管理系统 + 用户小程序，支持 rbac 动态权限、多租户、数据权限、工作流、三方登录、支付、短信、商城等功能
项目地址：https://gitee.com/zhijiantianya/yudao-cloud
视频教程：https://doc.iocoder.cn/video/
基于zookeeper的分布式锁更安全吗？
很多人（也包括martin在内）都认为，如果你想构建一个更安全的分布式锁，那么应该使用zookeeper，而不是redis。那么，为了对比的目的，让我们先暂时脱离开本文的题目，讨论一下基于zookeeper的分布式锁能提供绝对的安全吗？它需要fencing token机制的保护吗？
flavio junqueira是zookeeper的作者之一，他的这篇blog就写在martin和antirez发生争论的那几天。他在文中给出了一个基于zookeeper构建分布式锁的描述（当然这不是唯一的方式）：
客户端尝试创建一个znode节点，比如/lock。那么第一个客户端就创建成功了，相当于拿到了锁；而其它的客户端会创建失败（znode已存在），获取锁失败。
持有锁的客户端访问共享资源完成后，将znode删掉，这样其它客户端接下来就能来获取锁了。
znode应该被创建成ephemeral的。这是znode的一个特性，它保证如果创建znode的那个客户端崩溃了，那么相应的znode会被自动删除。这保证了锁一定会被释放。
看起来这个锁相当完美，没有redlock过期时间的问题，而且能在需要的时候让锁自动释放。但仔细考察的话，并不尽然。
zookeeper是怎么检测出某个客户端已经崩溃了呢？实际上，每个客户端都与zookeeper的某台服务器维护着一个session，这个session依赖定期的心跳(heartbeat)来维持。如果zookeeper长时间收不到客户端的心跳（这个时间称为sesion的过期时间），那么它就认为session过期了，通过这个session所创建的所有的ephemeral类型的znode节点都会被自动删除。
设想如下的执行序列：
客户端1创建了znode节点/lock，获得了锁。
客户端1进入了长时间的gc pause。
客户端1连接到zookeeper的session过期了。znode节点/lock被自动删除。
客户端2创建了znode节点/lock，从而获得了锁。
客户端1从gc pause中恢复过来，它仍然认为自己持有锁。
最后，客户端1和客户端2都认为自己持有了锁，冲突了。这与之前martin在文章中描述的由于gc pause导致的分布式锁失效的情况类似。
看起来，用zookeeper实现的分布式锁也不一定就是安全的。该有的问题它还是有。但是，zookeeper作为一个专门为分布式应用提供方案的框架，它提供了一些非常好的特性，是redis之类的方案所没有的。像前面提到的ephemeral类型的znode自动删除的功能就是一个例子。
还有一个很有用的特性是zookeeper的watch机制。这个机制可以这样来使用，比如当客户端试图创建/lock的时候，发现它已经存在了，这时候创建失败，但客户端不一定就此对外宣告获取锁失败。客户端可以进入一种等待状态，等待当/lock节点被删除的时候，zookeeper通过watch机制通知它，这样它就可以继续完成创建操作（获取锁）。这可以让分布式锁在客户端用起来就像一个本地的锁一样：加锁失败就阻塞住，直到获取到锁为止。这样的特性redlock就无法实现。
小结一下，基于zookeeper的锁和基于redis的锁相比在实现特性上有两个不同：
在正常情况下，客户端可以持有锁任意长的时间，这可以确保它做完所有需要的资源访问操作之后再释放锁。这避免了基于redis的锁对于有效时间(lock validity time)到底设置多长的两难问题。实际上，基于zookeeper的锁是依靠session（心跳）来维持锁的持有状态的，而redis不支持session。
基于zookeeper的锁支持在获取锁失败之后等待锁重新释放的事件。这让客户端对锁的使用更加灵活。
总结
综上所述，我们可以得出两种结论：
如果仅是为了效率(efficiency)，那么你可以自己选择你喜欢的一种分布式锁的实现。当然，你需要清楚地知道它在安全性上有哪些不足，以及它会带来什么后果，这也是为什么我们需要了解实现原理的原因，大多数情况下不会出问题，但是就万一的情况，处理起来可能需要大量的时间定位问题。
如果你是为了正确性(correctness)，那么请慎之又慎。就目前来说zookeeper的分布锁相对于redlock更加合理。
最后，由于redlock的出现其实是为了保证分布式锁的可靠性，但是由于实现的种种问题其可靠性并没有zookeeper分布式锁来的高，对于可容错的希望效率的场景下，redis分布式锁又可以完全满足，这也是导致了redlock被弃用的原因。
恒讯科技分析：本地数据中心与云数据中心哪个适合我们的业务？
魅蓝Max价格下调到1388元，较之小米Max2性价比不弱下风
博通收购赛门铁克失败：每股报价距成交只差75美分
植物照明市场是崧盛股份今年重点发展的领域之一
VR、AR如今的发展状况如何了
Redis分布式锁的使用场景
华为推出全球首款支持LTE Cat 6应用处理器
河套IT WALK(总第70期)：量子计算崛起、人机融合探索和智能网络革新
应用电子技术专业未来就业方向解读
安居侠智能环境检测面板PMT200评测，室内环境监测全能王
荣耀8怎么样？荣耀8评测：华为荣耀8四大隐藏功能汇总，不买荣耀8你买谁？
选择无线传感器需要注意什么？
Codasip向客户提供Tessent Enhanced Trace Encoder增强型追踪编码器解决方案
Xilinx发布单芯片SoC功能安全性解决方案 可缩短认证与开发时间 24个月
调查：高质体验、易用性和低成本推动视频对话/会议逐渐普及
汽车线束的一般设计流程和设计原则
AVENTK带你了解摄像模组AA制程胶作用及性能参数
蓝牙模块有哪些种类？BLE低功耗蓝牙模块有什么特点？
江波龙2款DDR3产品开始量产
集成电路产业在两会上备受“大佬”们的关注
分布式锁redlock
基于zookeeper的分布式锁更安全吗？
总结
今天我们来聊一聊redis分布式锁。
首先大家可以先思考一个简单的问题，为什么要使用分布式锁？普通的jvm锁为什么不可以？
这个时候，大家肯定会吧啦吧啦想到一堆，例如java应用属于进程级，不同的ecs中部署相同的应用，他们之间相互独立。
所以，在分布式系统中，当有多个客户端需要获取锁时，我们需要分布式锁。此时，锁是保存在一个共享存储系统中的，可以被多个客户端共享访问和获取。
分布式锁（set nx）
知道了分布式锁的使用场景，我们来自己简单的实现下分布式锁：
public class indexcontroller { public string deductstock() { string lockkey = lock:product_101; //setnx 获取分布式锁 string clientid = uuid.randomuuid().tostring(); boolean result = stringredistemplate.opsforvalue().setifabsent(lockkey, clientid, 30, timeunit.seconds); //jedis.setnx(k,v) if (!result) { return error_code; } try { int stock = integer.parseint(stringredistemplate.opsforvalue().get(stock)); // jedis.get(stock) if (stock > 0) { int realstock = stock - 1; stringredistemplate.opsforvalue().set(stock, realstock + ); // jedis.set(key,value) system.out.println(扣减成功，剩余库存: + realstock); } else { system.out.println(扣减失败，库存不足); } } finally { //解锁 if (clientid.equals(stringredistemplate.opsforvalue().get(lockkey))) { stringredistemplate.delete(lockkey); } }}
以上代码简单的实现了一个扣减库存的业务逻辑，我们拆开来说下都做了什么事情：
1、首先声明了lockkey，表示我们需要set的keyname
2、其次uuid.randomuuid().tostring();生成该次请求的requestid，为什么需要生成这个唯一的uuid，后面在解锁的时候会说到
3、获取分布式锁，通过stringredistemplate.opsforvalue().setifabsent来实现，该语句的意思是如果存在该key则返回false，若不存在则进行key的设置，设置成功后返回true，将当前线程获取的uuid设置成value，给定一个锁的过期时间，防止该线程无限制持久锁导致死锁，也为了防止该服务器突然宕机，导致其他机器的应用无法获取该锁，这个是必须要做的设置，至于过期的时间，可以根据内层业务逻辑的执行时间来决定
4、执行内层的业务逻辑，进行扣库存的操作
5、业务逻辑执行完成后，走到finally的解锁操作，进行解锁操作时，首先我们来判断当前锁的值是否为该线程持有的，防止当前线程执行较慢，导致锁过期，从而删除了其他线程持有的分布式锁，对于该操作，我来举个例子：
时刻1：线程a获取分布式锁，开始执行业务逻辑
时刻2：线程b等待分布式锁释放
时刻3：线程a所在机器io处理缓慢、gc pause等问题导致处理缓慢
时刻4：线程a依旧处于block状态，锁过期
时刻5：线程b获取分布式锁，开始执行业务逻辑，此时线程a结束block，开始释放锁
时刻6：线程b处理业务逻辑缓慢，线程a释放分布式锁，但是此时释放的是线程b的锁，导致其他线程可以开始获取锁
看到这里，为什么每个请求需要requestid，并且在释放锁的情况下判断是否是当前的requestid是有必要的。
以上，就是一个简单的分布式锁的实现过程。但是你觉得上述实现还存在问题吗？
答案是肯定的。若是在判断完分布式锁的value与requestid之后，锁过期了，依然会存在以上问题。
那么有没有什么办法可以规避以上问题，让我们不需要去完成这些实现，只需要专注于业务逻辑呢？
我们可以使用redisson，并且redisson有中文文档，方便英文不好的同学查看（开发团队中有中国的jackygurui）。
接下来我们再把上述代码简单的改造下就可以规避这些问题：
public class indexcontroller { public string deductstock() { string lockkey = lock:product_101; //setnx 获取分布式锁 //string clientid = uuid.randomuuid().tostring(); //boolean result = stringredistemplate.opsforvalue().setifabsent(lockkey, clientid, 30, timeunit.seconds); //jedis.setnx(k,v) //获取锁对象 rlock redissonlock = redisson.getlock(lockkey); //加分布式锁 redissonlock.lock(); try { int stock = integer.parseint(stringredistemplate.opsforvalue().get(stock)); // jedis.get(stock) if (stock > 0) { int realstock = stock - 1; stringredistemplate.opsforvalue().set(stock, realstock + ); // jedis.set(key,value) system.out.println(扣减成功，剩余库存: + realstock); } else { system.out.println(扣减失败，库存不足); } } finally { //解锁 //if (clientid.equals(stringredistemplate.opsforvalue().get(lockkey))) { // stringredistemplate.delete(lockkey); //} //redisson分布式锁解锁 redissonlock.unlock(); }}
可以看到，使用redisson分布式锁会简单很多，我们通过redissonlock.lock()和redissonlock.unlock()解决了这个问题，看到这里，是不是有同学会问，如果服务器宕机了，分布式锁会一直存在吗，也没有去指定过期时间？
redisson分布式锁中有一个watchdog机制，即会给一个leasetime，默认为30s，到期后锁自动释放，如果一直没有解锁，watchdog机制会一直重新设定锁的过期时间，通过设置timetask，延迟10s再次执行锁续命，将锁的过期时间重置为30s。下面就从redisson.lock()的源码来看下：
lock的最终加锁方法：
rfuture trylockinnerasync(long leasetime, timeunit unit, long threadid, redisstrictcommand command) { internallockleasetime = unit.tomillis(leasetime); return commandexecutor.evalwriteasync(getname(), longcodec.instance, command, if (redis.call('exists', keys[1]) == 0) then + redis.call('hset', keys[1], argv[2], 1); + redis.call('pexpire', keys[1], argv[1]); + return nil; + end; + if (redis.call('hexists', keys[1], argv[2]) == 1) then + redis.call('hincrby', keys[1], argv[2], 1); + redis.call('pexpire', keys[1], argv[1]); + return nil; + end; + return redis.call('pttl', keys[1]);, collections.singletonlist(getname()), internallockleasetime, getlockname(threadid)); }
可以看到lua脚本中redis.call('pexpire', keys[1], argv[1]);对key进行设置，并给定了一个internallockleasetime，给定的internallockleasetime就是默认的加锁时间，为30s。
接下来我们在看下锁续命的源码：
private void scheduleexpirationrenewal(final long threadid) { if (!expirationrenewalmap.containskey(this.getentryname())) { timeout task = this.commandexecutor.getconnectionmanager().newtimeout(new timertask() { public void run(timeout timeout) throws exception { //重新设置锁过期时间 rfuture future = redissonlock.this.commandexecutor.evalwriteasync(redissonlock.this.getname(), longcodec.instance, rediscommands.eval_boolean, if (redis.call('hexists', keys[1], argv[2]) == 1) then redis.call('pexpire', keys[1], argv[1]); return 1; end; return 0;, collections.singletonlist(redissonlock.this.getname()), new object[]{redissonlock.this.internallockleasetime, redissonlock.this.getlockname(threadid)}); future.addlistener(new futurelistener() { public void operationcomplete(future future) throws exception { redissonlock.expirationrenewalmap.remove(redissonlock.this.getentryname()); if (!future.issuccess()) { redissonlock.log.error(can't update lock + redissonlock.this.getname() + expiration, future.cause()); } else { //获取方法调用的结果 if ((boolean)future.getnow()) { //进行递归调用 redissonlock.this.scheduleexpirationrenewal(threadid); } } } }); } //延迟 this.internallockleasetime / 3l 再执行run方法 }, this.internallockleasetime / 3l, timeunit.milliseconds); if (expirationrenewalmap.putifabsent(this.getentryname(), task) != null) { task.cancel(); } } }
从源码层可以看到，加锁成功后，会延迟10s执行task中的run方法，然后在run方法里面执行锁过期时间的重置，如果时间重置成功，则继续递归调用该方法，延迟10s后进行锁续命，若重置锁时间失败，则可能表示锁已释放，退出该方法。
以上，就是关于一个redis分布式锁的说明，看到这里，大家应该对分布式锁有一个大致的了解了。
但是尽管使用了redisson完成分布式锁的实现，对于分布式锁是否还存在问题，分布式锁真的安全吗？
一般的，线上的环境肯定使用redis cluster，如果数据量不大，也会使用的redis sentinal。那么就存在主从复制的问题，那么是否会存在这种情况，在主库设置了分布式锁，但是可能由于网络或其他原因导致数据还没有同步到从库，此时主库宕机，选择从库作为主库，新主库中并没有该锁的信息，其他线程又可以进行锁申请，造成了发生线程安全问题的可能。
为了解决这个问题，redis的作者实现了redlock，基于redlock的实现有很大的争论，并且现在已经弃用了，但是我们还是需要了解下原理，以及之后基于这些问题的解决方案。
基于 spring boot + mybatis plus + vue & element 实现的后台管理系统 + 用户小程序，支持 rbac 动态权限、多租户、数据权限、工作流、三方登录、支付、短信、商城等功能
项目地址：https://gitee.com/zhijiantianya/ruoyi-vue-pro
视频教程：https://doc.iocoder.cn/video/
分布式锁redlock
redlock是基于单redis节点的分布式锁在failover的时候会产生解决不了的安全性问题而产生的，基于n个完全独立的redis节点。
下面我来看下redlock获取锁的过程：
运行redlock算法的客户端依次执行下面各个步骤，来完成获取锁 的操作：
获取当前时间（毫秒数）。
按顺序依次向n个redis节点执行获取锁 的操作。这个获取操作跟前面基于单redis节点的获取锁 的过程相同，包含随机字符串my_random_value，也包含过期时间(比如px 30000，即锁的有效时间)。为了保证在某个redis节点不可用的时候算法能够继续运行，这个获取锁 的操作还有一个超时时间(time out)，它要远小于锁的有效时间（几十毫秒量级）。客户端在向某个redis节点获取锁失败以后，应该立即尝试下一个redis节点。这里的失败，应该包含任何类型的失败，比如该redis节点不可用，或者该redis节点上的锁已经被其它客户端持有
计算整个获取锁的过程总共消耗了多长时间，计算方法是用当前时间减去第1步记录的时间。如果客户端从大多数redis节点（>= n/2+1）成功获取到了锁，并且获取锁总共消耗的时间没有超过锁的有效时间(lock validity time)，那么这时客户端才认为最终获取锁成功；否则，认为最终获取锁失败。
如果最终获取锁成功了，那么这个锁的有效时间应该重新计算，它等于最初的锁的有效时间减去第3步计算出来的获取锁消耗的时间。
如果最终获取锁失败了（可能由于获取到锁的redis节点个数少于n/2+1，或者整个获取锁的过程消耗的时间超过了锁的最初有效时间），那么客户端应该立即向所有redis节点发起释放锁 的操作。
好了，了解了redlock获取锁的机制之后，我们再来讨论下redlock会有哪些问题：
问题一：
假设一共有5个redis节点：a, b, c, d, e。设想发生了如下的事件序列：
客户端1成功锁住了a, b, c，获取锁 成功（但d和e没有锁住）。
节点c崩溃重启了，但客户端1在c上加的锁没有持久化下来，丢失了。
节点c重启后，客户端2锁住了c, d, e，获取锁 成功。
这样，客户端1和客户端2同时获得了锁（针对同一资源）。
在默认情况下，redis的aof持久化方式是每秒写一次磁盘（即执行fsync），因此最坏情况下可能丢失1秒的数据。为了尽可能不丢数据，redis允许设置成每次修改数据都进行fsync，但这会降低性能。当然，即使执行了fsync也仍然有可能丢失数据（这取决于系统而不是redis的实现）。所以，上面分析的由于节点重启引发的锁失效问题，总是有可能出现的。为了应对这一问题，redis作者antirez又提出了延迟重启 (delayed restarts)的概念。也就是说，一个节点崩溃后，先不立即重启它，而是等待一段时间再重启，这段时间应该大于锁的有效时间(lock validity time)。这样的话，这个节点在重启前所参与的锁都会过期，它在重启后就不会对现有的锁造成影响。
关于redlock还有一点细节值得拿出来分析一下：在最后释放锁 的时候，antirez在算法描述中特别强调，客户端应该向所有redis节点发起释放锁 的操作。也就是说，即使当时向某个节点获取锁没有成功，在释放锁的时候也不应该漏掉这个节点。这是为什么呢？设想这样一种情况，客户端发给某个redis节点的获取锁 的请求成功到达了该redis节点，这个节点也成功执行了set操作，但是它返回给客户端的响应包却丢失了。这在客户端看来，获取锁的请求由于超时而失败了，但在redis这边看来，加锁已经成功了。因此，释放锁的时候，客户端也应该对当时获取锁失败的那些redis节点同样发起请求。实际上，这种情况在异步通信模型中是有可能发生的：客户端向服务器通信是正常的，但反方向却是有问题的。
所以，如果不进行延迟重启，或者对于同一个主节点进行多个从节点的备份，并要求从节点的同步必须实时跟住主节点，也就是说需要配置redis从库的同步策略，将延迟设置为最小（主从同步是异步进行的），通过min-replicas-max-lag(旧版本的redis使用min-slaves-max-lag)来设置主从库间进行数据复制时，从库给主库发送 ack 消息的最大延迟（以秒为单位），也就是说，这个值需要设置为0，否则都有可能出现延迟，但是这个实际上在redis中是不存在的，min-replicas-max-lag设置为0，就代表着这个配置不生效。redis本身是为了高效而存在的，如果因为需要保证业务的准确性而使用，大大降低了redis的性能，建议使用的别的方式。
问题二：
如果客户端长期阻塞导致锁过期，那么它接下来访问共享资源就不安全了（没有了锁的保护）。在redlock中还是存在该问题的。
虽然在获取锁之后redlock会去判断锁的有效性，如果锁过期了，则会再去重新拿锁。但是如果发生在获取锁之后，那么该有效性都得不到保障了。
在上面的时序图中，假设锁服务本身是没有问题的，它总是能保证任一时刻最多只有一个客户端获得锁。上图中出现的lease这个词可以暂且认为就等同于一个带有自动过期功能的锁。客户端1在获得锁之后发生了很长时间的gc pause，在此期间，它获得的锁过期了，而客户端2获得了锁。当客户端1从gc pause中恢复过来的时候，它不知道自己持有的锁已经过期了，它依然向共享资源（上图中是一个存储服务）发起了写数据请求，而这时锁实际上被客户端2持有，因此两个客户端的写请求就有可能冲突（锁的互斥作用失效了）。
初看上去，有人可能会说，既然客户端1从gc pause中恢复过来以后不知道自己持有的锁已经过期了，那么它可以在访问共享资源之前先判断一下锁是否过期。但仔细想想，这丝毫也没有帮助。因为gc pause可能发生在任意时刻，也许恰好在判断完之后。
也有人会说，如果客户端使用没有gc的语言来实现，是不是就没有这个问题呢？质疑者martin指出，系统环境太复杂，仍然有很多原因导致进程的pause，比如虚存造成的缺页故障(page fault)，再比如cpu资源的竞争。即使不考虑进程pause的情况，网络延迟也仍然会造成类似的结果。
总结起来就是说，即使锁服务本身是没有问题的，而仅仅是客户端有长时间的pause或网络延迟，仍然会造成两个客户端同时访问共享资源的冲突情况发生。
那怎么解决这个问题呢？martin给出了一种方法，称为fencing token。fencing token是一个单调递增的数字，当客户端成功获取锁的时候它随同锁一起返回给客户端。而客户端访问共享资源的时候带着这个fencing token，这样提供共享资源的服务就能根据它进行检查，拒绝掉延迟到来的访问请求（避免了冲突）。如下图：
在上图中，客户端1先获取到的锁，因此有一个较小的fencing token，等于33，而客户端2后获取到的锁，有一个较大的fencing token，等于34。客户端1从gc pause中恢复过来之后，依然是向存储服务发送访问请求，但是带了fencing token = 33。存储服务发现它之前已经处理过34的请求，所以会拒绝掉这次33的请求。这样就避免了冲突。
但是，对于客户端和资源服务器之间的延迟（即发生在算法第3步之后的延迟），antirez是承认所有的分布式锁的实现，包括redlock，是没有什么好办法来应对的。包括在我们到生产环境中，无法避免分布式锁超时。
在讨论中，有人提出客户端1和客户端2都发生了gc pause，两个fencing token都延迟了，它们几乎同时到达了文件服务器，而且保持了顺序。那么，我们新加入的判断逻辑，即判断fencing token的合理性，应该对两个请求都会放过，而放过之后它们几乎同时在操作文件，还是冲突了。既然martin宣称fencing token能保证分布式锁的正确性，那么上面这种可能的猜测也许是我们理解错了。但是martin并没有在后面做出解释。
问题三：
redlock对系统记时(timing)的过分依赖，下面给出一个例子（还是假设有5个redis节点a, b, c, d, e）：
客户端1从redis节点a, b, c成功获取了锁（多数节点）。由于网络问题，与d和e通信失败。
节点c上的时钟发生了向前跳跃，导致它上面维护的锁快速过期。
客户端2从redis节点c, d, e成功获取了同一个资源的锁（多数节点）。
客户端1和客户端2现在都认为自己持有了锁。
上面这种情况之所以有可能发生，本质上是因为redlock的安全性(safety property)对系统的时钟有比较强的依赖，一旦系统的时钟变得不准确，算法的安全性也就保证不了了。
但是作者反驳到，通过恰当的运维，完全可以避免时钟发生大的跳动，而redlock对于时钟的要求在现实系统中是完全可以满足的。哪怕是手动修改时钟这种人为原因，不要那么做就是了。否则的话，都会出现问题。
说了这么多关于redlock的问题，到底有没有什么分布式锁能保证安全性呢？我们接下来再来看看zookeeper分布式锁。
基于 spring cloud alibaba + gateway + nacos + rocketmq + vue & element 实现的后台管理系统 + 用户小程序，支持 rbac 动态权限、多租户、数据权限、工作流、三方登录、支付、短信、商城等功能
项目地址：https://gitee.com/zhijiantianya/yudao-cloud
视频教程：https://doc.iocoder.cn/video/
基于zookeeper的分布式锁更安全吗？
很多人（也包括martin在内）都认为，如果你想构建一个更安全的分布式锁，那么应该使用zookeeper，而不是redis。那么，为了对比的目的，让我们先暂时脱离开本文的题目，讨论一下基于zookeeper的分布式锁能提供绝对的安全吗？它需要fencing token机制的保护吗？
flavio junqueira是zookeeper的作者之一，他的这篇blog就写在martin和antirez发生争论的那几天。他在文中给出了一个基于zookeeper构建分布式锁的描述（当然这不是唯一的方式）：
客户端尝试创建一个znode节点，比如/lock。那么第一个客户端就创建成功了，相当于拿到了锁；而其它的客户端会创建失败（znode已存在），获取锁失败。
持有锁的客户端访问共享资源完成后，将znode删掉，这样其它客户端接下来就能来获取锁了。
znode应该被创建成ephemeral的。这是znode的一个特性，它保证如果创建znode的那个客户端崩溃了，那么相应的znode会被自动删除。这保证了锁一定会被释放。
看起来这个锁相当完美，没有redlock过期时间的问题，而且能在需要的时候让锁自动释放。但仔细考察的话，并不尽然。
zookeeper是怎么检测出某个客户端已经崩溃了呢？实际上，每个客户端都与zookeeper的某台服务器维护着一个session，这个session依赖定期的心跳(heartbeat)来维持。如果zookeeper长时间收不到客户端的心跳（这个时间称为sesion的过期时间），那么它就认为session过期了，通过这个session所创建的所有的ephemeral类型的znode节点都会被自动删除。
设想如下的执行序列：
客户端1创建了znode节点/lock，获得了锁。
客户端1进入了长时间的gc pause。
客户端1连接到zookeeper的session过期了。znode节点/lock被自动删除。
客户端2创建了znode节点/lock，从而获得了锁。
客户端1从gc pause中恢复过来，它仍然认为自己持有锁。
最后，客户端1和客户端2都认为自己持有了锁，冲突了。这与之前martin在文章中描述的由于gc pause导致的分布式锁失效的情况类似。
看起来，用zookeeper实现的分布式锁也不一定就是安全的。该有的问题它还是有。但是，zookeeper作为一个专门为分布式应用提供方案的框架，它提供了一些非常好的特性，是redis之类的方案所没有的。像前面提到的ephemeral类型的znode自动删除的功能就是一个例子。
还有一个很有用的特性是zookeeper的watch机制。这个机制可以这样来使用，比如当客户端试图创建/lock的时候，发现它已经存在了，这时候创建失败，但客户端不一定就此对外宣告获取锁失败。客户端可以进入一种等待状态，等待当/lock节点被删除的时候，zookeeper通过watch机制通知它，这样它就可以继续完成创建操作（获取锁）。这可以让分布式锁在客户端用起来就像一个本地的锁一样：加锁失败就阻塞住，直到获取到锁为止。这样的特性redlock就无法实现。
小结一下，基于zookeeper的锁和基于redis的锁相比在实现特性上有两个不同：
在正常情况下，客户端可以持有锁任意长的时间，这可以确保它做完所有需要的资源访问操作之后再释放锁。这避免了基于redis的锁对于有效时间(lock validity time)到底设置多长的两难问题。实际上，基于zookeeper的锁是依靠session（心跳）来维持锁的持有状态的，而redis不支持session。
基于zookeeper的锁支持在获取锁失败之后等待锁重新释放的事件。这让客户端对锁的使用更加灵活。
总结
综上所述，我们可以得出两种结论：
如果仅是为了效率(efficiency)，那么你可以自己选择你喜欢的一种分布式锁的实现。当然，你需要清楚地知道它在安全性上有哪些不足，以及它会带来什么后果，这也是为什么我们需要了解实现原理的原因，大多数情况下不会出问题，但是就万一的情况，处理起来可能需要大量的时间定位问题。
如果你是为了正确性(correctness)，那么请慎之又慎。就目前来说zookeeper的分布锁相对于redlock更加合理。
最后，由于redlock的出现其实是为了保证分布式锁的可靠性，但是由于实现的种种问题其可靠性并没有zookeeper分布式锁来的高，对于可容错的希望效率的场景下，redis分布式锁又可以完全满足，这也是导致了redlock被弃用的原因。
恒讯科技分析：本地数据中心与云数据中心哪个适合我们的业务？
魅蓝Max价格下调到1388元，较之小米Max2性价比不弱下风
博通收购赛门铁克失败：每股报价距成交只差75美分
植物照明市场是崧盛股份今年重点发展的领域之一
VR、AR如今的发展状况如何了
Redis分布式锁的使用场景
华为推出全球首款支持LTE Cat 6应用处理器
河套IT WALK(总第70期)：量子计算崛起、人机融合探索和智能网络革新
应用电子技术专业未来就业方向解读
安居侠智能环境检测面板PMT200评测，室内环境监测全能王
荣耀8怎么样？荣耀8评测：华为荣耀8四大隐藏功能汇总，不买荣耀8你买谁？
选择无线传感器需要注意什么？
Codasip向客户提供Tessent Enhanced Trace Encoder增强型追踪编码器解决方案
Xilinx发布单芯片SoC功能安全性解决方案 可缩短认证与开发时间 24个月
调查：高质体验、易用性和低成本推动视频对话/会议逐渐普及
汽车线束的一般设计流程和设计原则
AVENTK带你了解摄像模组AA制程胶作用及性能参数
蓝牙模块有哪些种类？BLE低功耗蓝牙模块有什么特点？
江波龙2款DDR3产品开始量产
集成电路产业在两会上备受“大佬”们的关注