为什么要用分布式锁？在讨论这个问题之前，我们先来看一个业务场景。图片来自Pexels为什么要使用分布式锁？系统A是电子商务系统，目前部署在一台机器上。系统中有用户下单的接口，但用户在下单前必须先查看库存，确保库存充足后才能为用户下单。由于系统具有一定的并发性，商品的库存会提前保存在Redis中，用户下单时会更新Redis的库存。此时的系统架构如下：但这会造成一个问题：如果在某个时刻，Redis中某商品的库存为1，此时有两个请求同时过来，其中一个是requests执行到上图中的第3步，更新后的数据库库存为0，但第4步还没有执行。而另一个请求执行到第2步，发现库存还是1，继续执行第3步，结果卖了2件，但实际上库存只有1件。显然错了！这是一个典型的库存超卖问题。至此，我们很容易想到一个解决方案：将步骤2、3、4用一把锁锁住，这样它们执行完之后，另一个线程就可以进来执行步骤2了。根据上图，在执行步骤时2、使用Java提供的Synchronized或者ReentrantLock进行加锁，执行完第4步后释放锁。这样，2、3、4这三个步骤就被“锁定”了，多个线程只能串行执行。但是好景不长，整个系统的并发量飙升，一台机器都搞不定了。现在添加一台机器，如下图：添加机器后，系统变成如上图，天哪！假设两个用户的请求同时到达，但是落在了不同的机器上，这两个请求是不是可以同时执行，或者会不会出现库存超卖的问题。为什么？因为上图中的两个A系统运行在两个不同的JVM中，所以它们加的锁只对自己JVM中的线程有效，对其他JVM中的线程无效。所以，这里的问题是：Java提供的原生锁机制在多机部署场景下失效，因为两台机器加的锁不是同一个锁（两个锁在不同的JVM）。那么，只要保证两台机器加的锁是同一把锁，问题不就解决了吗？至此，分布式锁就该隆重登场了。分布式锁的思想是：在整个系统中提供一个全局唯一的“东西”去获取锁，然后各个系统在需要加锁的时候会去向这个“东西”去获取锁，这样不同systems得到的可以认为是同一个锁。至于这个“东西”，可以是Redis，也可以是Zookeeper，也可以是数据库。文字描述不是很直观，我们看下图：通过上面的分析，我们知道在分布式部署系统的情况下使用Java原生的锁机制，在超卖库存场景下是无法保证线程安全的，所以我们需要使用分布式锁方案。那么，如何实现分布式锁呢？然后往下看！基于Redis实现分布式锁上面分析了为什么要使用分布式锁。这里我们就具体看看分布式锁落地的时候应该怎么处理。①一种常见的解决方案是使用Redis作为分布式锁。使用Redis作为分布式锁的思路大致是这样的：在Redis中设置一个值表示加锁，然后在释放锁的时候删除Key。具体代码如下：//获取锁//NX表示key不存在则成功，key存在则返回false。说明，这两条指令不是原子的//需要使用redis的lua脚本支持特性，redis执行lua脚本是原子的ifredis.call("get",KEYS[1])==ARGV[1]thenreturnredis.call("del",KEYS[1])elsereturn0end这个方法有几个关键点：一定要使用SET键值NXPX毫秒命令。如果没有，先设置值，再设置过期时间。这不是一个原子操作，在设置过期时间之前可能会崩溃，从而造成死锁（Key***存在）。值必须是唯一的。这是因为在解锁的时候，在删除Key之前需要验证Value和被锁的一致。这时候就避免了一种情况：假设A获得了锁，过期时间为30s。35s后，锁已经自动释放，A去释放锁，但是此时B可能获取到锁。客户端A无法删除B的锁。除了考虑客户端如何实现分布式锁，还需要考虑Redis的部署。Redis有三种部署方式：单机模式Master-Slave+Sentinel选举模式RedisCluster模式使用Redis作为分布式锁。缺点是如果采用单机部署方式，会出现单点问题，只要Redis失效。锁定将不起作用。使用Master-Slave模式，加锁时只有一个节点被加锁。即使通过Sentinel实现高可用，如果Master节点出现故障，发生主从切换，此时也可能会出现锁丢失的问题。基于以上考虑，Redis的作者也考虑到了这个问题，他提出了RedLock算法。这个算法的意思大概是这样的：假设Redis的部署方式是RedisCluster，一共有5个Master节点。通过以下步骤获取锁：获取当前时间戳，单位为毫秒。尽量在每个Master节点上依次创建锁，并且过期时间设置得短一些，一般为几十毫秒。尝试在大多数节点上建立锁，比如5个节点需要3个节点(n/2+1)。客户端计算建立锁的时间。如果建立锁的时间小于超时时间，则建立成功。如果加锁失败，则依次删除加锁。只要别人创建了分布式锁，你就得不停地轮询来尝试获取锁。不过这样的算法还是比较有争议的，可能还会有很多问题，并且不能保证加锁过程一定是正确的。②另一种方式：Redisson另外要实现Redis的分布式锁，除了基于RedisClient原生API实现外，还可以使用开源框架：Redission。Redisson是一个企业级开源的RedisClient，同样提供分布式锁的支持。我也非常推荐大家使用，为什么呢？回想一下我上面说的，如果我写代码通过Redis设置一个值，是通过下面的命令来设置的：SETanyLockunique_valueNXPX30000这里设置的超时时间是30s，如果超过30s还没有完成业务逻辑，Key就会过期，其他线程可能会获取锁。这样一来，第一个线程还没有执行完业务逻辑，第二个线程进来的时候也会出现线程安全问题。所以还是要维护这个过期时间，太麻烦了~来看看Redisson是怎么实现的吧？先感受下使用Redission的爽快：Configconfig=newConfig();config.useClusterServers().addNodeAddress("redis://192.168.31.101:7001").addNodeAddress("redis://192.168.31.101:7002")。addNodeAddress("redis://192.168.31.101:7003").addNodeAddress("redis://192.168.31.102:7001").addNodeAddress("redis://192.168.31.102:7002").addNodeAddress("redis://192.168.31.102:7003");RedissonClientredisson=Redisson.create(config);RLocklock=redisson.getLock("anyLock");lock.lock();lock.unlock();就这么简单，我们只需要通过其API中的Lock和Unlock可以完成分布式锁，帮助我们考虑了很多细节：Redisson的所有指令都是通过Lua脚本执行的，Redis支持Lua脚本的原子执行。Redisson设置Key的默认过期时间为30s。如果客户端持有锁超过30s怎么办？Redisson中有一个Watchdog的概念，翻译过来就是看门狗，它会在你获取到锁后每隔10s帮你设置Key超时时间为30s。这样即使一直持有锁，也不会出现Key过期其他线程获取锁的问题。Redisson的“看门狗”逻辑确保不会发生死锁。（如果机器宕机，看门狗也会宕机。此时Key的过期时间不会延长，30s后会自动过期，其他线程可以获取锁）下面是它的一点实现代码：//加锁逻辑privateRFuturetryAcquireAsync(longleaseTime,TimeUnitunit,finallongthreadId){if(leaseTime!=-1){returntryLockInnerAsync(leaseTime,unit,threadId,RedisCommands.EVAL_LONG);}//调用lua脚本，设置一些key，过期时间RFuturettlRemainingFuture=tryLockInnerAsync(commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout(),TimeUnit.MILLISECONDS,threadId,RedisCommands.EVAL_LONG);ttlRemainingFuture.addListener(newFutureListener(){@OverridepublicvoidoperationComplete(未来未来)throwsException{if(!future.isSuccess()){return;}LongttlRemaining=future.getNow();//lockacquiredif(ttlRemaining==null){//看门狗逻辑scheduleExpirationRenewal(threadId);}}});returnttlRemainingFuture;}RFuturetryLockInnerAsync(longleaseTime,TimeUnitunit,longthreadId,RedisStrictCommand命令){internalLockLeaseTime=unit.toMillis(leaseTime);returncommandExecutor.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]);"+"returnnil;"+"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]);"+"returnnil;"+"end;"+"returnredis.call('pttl',KEYS[1]);",Collections.
