分散式：这里的分散式指的是分散式系统，波及到好多时期和表面，包括CAP 表面、分散式存储、分散式事务、分散式锁...

分散式系统是由一组通过收集进行通讯、为了完成共同的任务而结合使命的筹画机节点构成的系统。

分散式系统的出现是为了用低价的、普通的机器完成单个筹画机无法完成的筹画、存储任务。其方向是期骗更多的机器，处理更多的数据。

Java 低级口试问题，来拼写下 赛克瑞纳挨日的

从锁的使用场景有来看下边这 3 种锁：

线程锁：synchronized 是用在方法或代码块中的，咱们把它叫『线程锁』，线程锁的已矣其实是靠线程之间分享内存已矣的，说白了即是内存中的一个整型数，有幽闲、上锁这类情状，比如 synchronized 是在对象头中的 Mark Word 有个锁情状标志，Lock 的已矣类大部分都有个叫 volatile int state 的分享变量来作念情状标志。

程度锁：为了次第吞并操作系统中多个程度造访某个分享资源，因为程度具有零丁性，各个程度无法造访其他程度的资源，因此无法通过 synchronized 等线程锁已矣程度锁。比如说，咱们的吞并个 linux 职业器，部署了好几个 Java 面容，有可能同期造访或操作职业器上的疏导数据，这就需要程度锁，一般不错用『文献锁』来达到程度互斥。

分散式锁：随着用户越来越多，咱们上了好多职业器，底本有个定时给客户发邮件的任务，若是不加以次第的话，到点后每台机器跑一次任务，客户就会收到 N 条邮件，这就需要通过分散式锁来互斥了。

书面解释：分散式锁是次第分散式系统或不同系统之间共同造访分享资源的一种锁已矣，若是不同的系统或吞并个系统的不同主机之间分享了某个资源时，时常需要互斥来辞谢相互干豫来保证一致性。

知谈了什么是分散式锁，接下来就到了时期选型要津

要已矣一个分散式锁，咱们一般弃取集群机器都不错操作的外部系统，然后各个机器都去这个外部系统肯求锁。

看过一项针对不同国家的人的调查，问他们：“你对自己感到满意吗？”

这个外部系合伙般需要骄贵如下条目才调胜任：

我以为不错这样类比：

好多商贩要租用某个仓库，吞并时刻，只可给一个商贩租用，且只可有一把钥匙，还得有固定的“租期”，到期后要回收的，诚然最迫切的是仓库门不成坏了，要不锁都锁不住。这不即是分散式锁吗?

欷歔我方竟然个爱时期爱生存的次第猿~~

其实锁，本色上即是用来进行防重操作的(数据一致性)，像查询这种幂等操作，就不需要费这劲

径直上论断：

分散式锁一般有三种已矣格式：1. 数据库乐不雅锁;2. 基于 Redis 的分散式锁;3. 基于 ZooKeeper 的分散式锁。

但为了追求更好的性能，咱们凡俗会弃取使用 Redis 或 Zookeeper 来作念。

想必也有心爱问为什么的同学，那数据库客不雅锁如何就性能不好了?

使用数据库乐不雅锁，包括主键防重，版块号次第。然而这两种方法各故意弊。

使用主键打破的计策进行防重，在并发量至极高的情况下对数据库性能会有影响，尤其是应用数据表和主键打破表在一个库的时候，弘扬愈加显着。还有即是在 MySQL 数据库中采用主键打破防重，在大并发情况下有可能会形成锁表征象，相比好的见解是在次第中分娩主键进行防重。

使用版块号计策

这个计策源于 MySQL 的 MVCC 机制，使用这个计策其实自己莫得什么问题，唯独的问题即是对数据表侵入较大，咱们要为每个表蓄意一个版块号字段，然后写一条判断 SQL 每次进行判断。

第三趴，编码

其实 Redis 官网还是给出了已矣：https://redis.io/topics/distlock，说各式竹素和博客用了各式妙技去用 Redis 已矣分散式锁，建议用 Redlock 已矣，这样更程序、更安全。咱们循序渐进来看

咱们默许指定大师用的是 Redis 2.6.12 及更高的版块，就不再去讲 setnx、expire 这种了，径直 set 号召加锁

set 亚新娱乐城key value[expiration EX seconds|PX milliseconds] [NX|XX] 

eg:

SET resource_name my_random_value NX PX 30000 

SET 号召的举止不错通过一系列参数来修改

这条教导的钦慕：当 key——resource_name 不存在时创建这样的key，设值为 my_random_value，并设立逾期时期 30000 毫秒。

别看这干了两件事，因为 Redis 是单线程的，这一条教导不会被打断，是以是原子性的操作。

Redis 已矣分散式锁的主要才略：

设立一个马上值的钦慕是在解锁时候判断 key 的值和咱们存储的马上数是不是相似，相似的话，才是我方的锁，径直 del 解锁就行。

诚然这个两个操作要保证原子性，是以 Redis 给出了一段 lua 剧本(Redis 职业器会单线程原子性试验 lua 剧本，保证 lua 剧本在处理的过程中不会被纵情其它请求打断。)：

if redis.call("get",KEYS[1]) == ARGV[1] then     return redis.call("del",KEYS[1]) else     return 0 end 

咱们先抛出两个问题想考：

获取锁时，逾期时期要设立若干适合呢?

预估一个适合的时期，其实没那么容易，比如操作资源的时期最慢可能要 10 s，而咱们只设立了 5 s 就逾期，那就存在锁提前逾期的风险。这个问题先记下，咱们先看下 Javaer 要如何在代码顶用 Redis 锁。

容错性如何保证呢?

Redis 挂了如何办，你可能会说上主从、上集群，但也会出现这样的顶点情况，当咱们上锁后，主节点就挂了，这个时候还没来的急同步到从节点，主从切换后锁照旧丢了

带着这两个问题，咱们接着看

redisson 是 Redis 官方的分散式锁组件。GitHub 地址：https://github.com/redisson/redisson

Redisson 是一个在 Redis 的基础上已矣的 Java 驻内存数据网格(In-Memory Data Grid)。它不仅提供了一系列的分散式的 Java 常用对象，还已矣了可重入锁(Reentrant Lock)、自制锁(Fair Lock、联锁(MultiLock)、 红锁(RedLock)、 读写锁(ReadWriteLock)等，还提供了很多分散式职业。Redisson 提供了使用 Redis 的最毛糙和最浅显的方法。Redisson 的目的是促进使用者对 Redis 的热心分离(Separation of Concern)，从而让使用者或者将元气心灵更连合地放在处理业务逻辑上。

redisson 当今还是很宽绰了，github 的 wiki 也很详备，分散式锁的先容径直戳 Distributed locks and synchronizers

Redisson 撑执单点模式、主从模式、哨兵模式、集群模式，仅仅成立的不同，咱们以单点模式来看下如何使用，代码很毛糙，都还是为咱们封装好了，径直拿来用就好，详备的demo，我放在了 github: starfish-learn-redisson 上，这里就不一步步来了

RLock lock = redisson.getLock("myLock"); 

RLock 提供了各式锁方法，咱们来解读下这个接口方法，

注：代码为 3.16.2 版块，不错看到袭取自 JDK 的 Lock 接口，和 Reddsion 的异步锁接口 RLockAsync(这个咱们先不磋议)

RLock

public interface RLock extends Lock, RLockAsync {      /**      * 获取锁的名字      */     String getName();          /**      * 这个叫终局锁操作，示意该锁不错被中断 假如A和B同期调这个方法，A获取锁，B为获取锁，那么B线程不错通过      * Thread.currentThread().interrupt(); 方法信得过中断该线程      */     void lockInterruptibly(long leaseTime, TimeUnit unit) throws InterruptedException;      /**      * 这个应该是最常用的，尝试获取锁      * waitTimeout 尝试获取锁的最大恭候时期，杰出这个值，则认为获取锁失败      * leaseTime   锁的执偶然期,杰出这个时期锁会自动失效（值应设立为大于业务处理的时期，确保在锁灵验期内业务能处理完）      */     boolean tryLock(long waitTime, long leaseTime, TimeUnit unit) throws InterruptedException;      /**      * 锁的灵验期设立为 leaseTime，逾期后自动失效      * 若是 leaseTime 设立为 -1, 示意不主动逾期      */     void lock(long leaseTime, TimeUnit unit);      /**      * Unlocks the lock independently of its state      */     boolean forceUnlock();      /**      * 查验是否被另一个线程锁住      */     boolean isLocked();      /**      * 查验刻下哨线程是否执有该锁      */     boolean isHeldByCurrentThread();         /**      *  这个就明明晰，查验指定线程是否执有锁      */     boolean isHeldByThread(long threadId);      /**      * 复返刻下哨程执有锁的次数      */     int getHoldCount();      /**      * 复返锁的剩余时期      * @return time in milliseconds      *          -2 if the lock does not exist.      *          -1 if the lock exists but has no associated expire.      */     long remainTimeToLive();      } 

Demo

即是这样毛糙，Redisson 还是作念好了封装，使用起来 so easy，若是使用主从、哨兵、集群这种也仅仅成立不同。

旨趣

看源码小 tips，最佳是 fork 到我方的仓库，然后拉到腹地，边看边可贵，然后提交到我方的仓库，也方便之后再看，不想这样发奋的，也不错径直看我的 Jstarfish/redisson

先看下 RLock 的类筹商

随着源码，不错发现 RedissonLock 是 RLock 的径直已矣，亦然咱们加锁、解锁操作的中枢类

加锁

主要的加锁方法就下边这两个，区别也很毛糙，一个有恭候时期，一个莫得，是以咱们挑个复杂的看(源码包含了另一个的绝大部分)

boolean tryLock(long waitTime, long leaseTime, TimeUnit unit) throws InterruptedException; void lock(long leaseTime, TimeUnit unit); 

RedissonLock.tryLock

@Override public boolean tryLock(long waitTime, long leaseTime, TimeUnit unit) throws InterruptedException {     // 获取等锁的最永劫期     long time = unit.toMillis(waitTime);     long current = System.currentTimeMillis();     //取适合前哨程id（判断是否可重入锁的枢纽）     long threadId = Thread.currentThread().getId();     // 【中枢点1】尝试获取锁，若复返值为null，则示意已获取到锁，复返的ttl即是key的剩孑遗活时期     Long ttl = tryAcquire(waitTime, leaseTime, unit, threadId);     if (ttl == null) {         return true;     }     // 还不错容忍的恭候时长 = 获取锁能容忍的最大恭候时长 - 试验完上述操作经过的时期     time -= System.currentTimeMillis() - current;     if (time <= 0) {         //等不到了，径直复返失败         acquireFailed(waitTime, unit, threadId);         return false;     }      current = System.currentTimeMillis();     /**      * 【中枢点2】      * 订阅解锁音信 redisson_lock__channel:{$KEY}，并通过await方法禁绝恭候锁开释，处理了无效的锁肯求顿然资源的问题：      * 基于信息量，当锁被其它资源占用时，刻下哨程通过 Redis 的 channel 订阅锁的开释事件，一朝锁开释会发音信见告待恭候的线程进行竞争      * 当 this.await复返false，诠释恭候时期还是超出获取锁最大恭候时期，取消订阅并复返获取锁失败      * 当 this.await复返true，参预轮回尝试获取锁      */     RFuture<RedissonLockEntry> subscribeFuture = subscribe(threadId);     //await 方法里面是用CountDownLatch来已矣禁绝，获取subscribe异步试验的终结（应用了Netty 的 Future）     if (!subscribeFuture.await(time, TimeUnit.MILLISECONDS)) {         if (!subscribeFuture.cancel(false)) {             subscribeFuture.onComplete((res, e) -> {                 if (e == null) {                     unsubscribe(subscribeFuture, threadId);                 }             });         }         acquireFailed(waitTime, unit, threadId);         return false;     }      // ttl 不为空，示意还是有这样的key了，只可禁绝恭候     try {         time -= System.currentTimeMillis() - current;         if (time <= 0) {             acquireFailed(waitTime, unit, threadId);             return false;         }          // 来个死轮回，欧博注册官网不竭尝试着获取锁         while (true) {             long currentTime = System.currentTimeMillis();             ttl = tryAcquire(waitTime, leaseTime, unit, threadId);             if (ttl == null) {                 return true;             }              time -= System.currentTimeMillis() - currentTime;             if (time <= 0) {                 acquireFailed(waitTime, unit, threadId);                 return false;             }              currentTime = System.currentTimeMillis();             /**             * 【中枢点3】凭证锁TTL，调养禁绝恭候时长；             * 1、latch其实是个信号量Semaphore，调用其tryAcquire方法会让刻下哨程禁绝一段时期，幸免在while轮回中时常请求获锁；             *  当其他线程开释了占用的锁，会播送解锁音信，监听器禁受解锁音信，并开释信号量，最终会叫醒禁绝在这里的线程             * 2、该Semaphore的release方法，会在订阅解锁音信的监听器音信处理方法org.redisson.pubsub.LockPubSub#onMessage调用；             */             //调用信号量的方法来禁绝线程，时长为锁恭候时期和租期时期中较小的阿谁             if (ttl >= 0 && ttl < time) {                 subscribeFuture.getNow().getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);             } else {                 subscribeFuture.getNow().getLatch().tryAcquire(time, TimeUnit.MILLISECONDS);             }              time -= System.currentTimeMillis() - currentTime;             if (time <= 0) {                 acquireFailed(waitTime, unit, threadId);                 return false;             }         }     } finally {         // 获取到锁或者抛出中断很是，退订redisson_lock__channel:{$KEY}，不再热心解锁事件         unsubscribe(subscribeFuture, threadId);     } } 

接着看可贵中提到的 3 个中枢点

中枢点1-尝试加锁：RedissonLock.tryAcquireAsync

private <T> RFuture<Long> tryAcquireAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId) {     RFuture<Long> ttlRemainingFuture;     // leaseTime != -1 诠释没逾期     if (leaseTime != -1) {         // 实质是异步试验加锁Lua剧本         ttlRemainingFuture = tryLockInnerAsync(waitTime, leaseTime, unit, threadId, RedisCommands.EVAL_LONG);     } else {         // 不然，还是逾期了,传参变为新的时期（续期后）         ttlRemainingFuture = tryLockInnerAsync(waitTime, internalLockLeaseTime,                 TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_LONG);     }     ttlRemainingFuture.onComplete((ttlRemaining, e) -> {         if (e != null) {             return;         }          // lock acquired         if (ttlRemaining == null) {             if (leaseTime != -1) {                 internalLockLeaseTime = unit.toMillis(leaseTime);             } else {                 // 续期                 scheduleExpirationRenewal(threadId);             }         }     });     return ttlRemainingFuture; } 

异步试验加锁 Lua 剧本：RedissonLock.tryLockInnerAsync

<T> RFuture<T> tryLockInnerAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> command) {     return evalWriteAsync(getRawName(), LongCodec.INSTANCE, command,             // 1.若是缓存中的key不存在，则试验 hincrby 号召(hincrby key UUID+threadId 1), 设值重入次数1             // 然后通过 pexpire 号召设立锁的逾期时期(即锁的租约时期)             // 复返空值 nil ，示意获取锁胜利             "if (redis.call('exists', KEYS[1]) == 0) then " +                     "redis.call('hincrby', KEYS[1], ARGV[2], 1); " +                     "redis.call('pexpire', KEYS[1], ARGV[1]); " +                     "return nil; " +                     "end; " +                     // 若是key还是存在，而且value也匹配，示意是刻下哨程执有的锁，则试验 hincrby 号召，重入次数加1，而且设立失效时期                     "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; " +                     //若是key还是存在，然而value不匹配，诠释锁还是被其他线程执有，通过 pttl 号召获取锁的剩孑遗活时期并复返，至此获取锁失败                     "return redis.call('pttl', KEYS[1]);",             Collections.singletonList(getRawName()), unit.toMillis(leaseTime), getLockName(threadId)); } 

看门狗续期：RedissonBaseLock.scheduleExpirationRenewal

// 基于线程ID定时转机和续期 protected void scheduleExpirationRenewal(long threadId) {     // 新建一个ExpirationEntry纪录线程重入计数     ExpirationEntry entry = new ExpirationEntry();     ExpirationEntry oldEntry = EXPIRATION_RENEWAL_MAP.putIfAbsent(getEntryName(), entry);     if (oldEntry != null) {         // 刻下进行确刻下哨程重入加锁         oldEntry.addThreadId(threadId);     } else {         // 刻下进行确刻下哨程初度加锁         entry.addThreadId(threadId);         // 初度新建ExpirationEntry需要触发续期方法，纪录续期的任务句柄         renewExpiration();     } }  // 处理续期 private void renewExpiration() {   // 凭证entryName获取ExpirationEntry实例，若是为空，诠释在cancelExpirationRenewal()方法还是被移除，一般是解锁的时候触发   ExpirationEntry ee = EXPIRATION_RENEWAL_MAP.get(getEntryName());   if (ee == null) {     return;   }    // 新建一个定时任务，这个即是看门狗的已矣，io.netty.util.Timeout是Netty结合时期轮使用的定时任求实例   Timeout task = commandExecutor.getConnectionManager().newTimeout(new TimerTask() {     @Override     public void run(Timeout timeout) throws Exception {       // 这里是重迭外面的阿谁逻辑，       ExpirationEntry ent = EXPIRATION_RENEWAL_MAP.get(getEntryName());       if (ent == null) {         return;       }       // 获取ExpirationEntry中首个线程ID，若是为空诠释调用过cancelExpirationRenewal()方法清空执有的线程重入计数，一般是锁还是开释的场景       Long threadId = ent.getFirstThreadId();       if (threadId == null) {         return;       }       // 向Redis异步发送续期的号召       RFuture<Boolean> future = renewExpirationAsync(threadId);       future.onComplete((res, e) -> {         // 抛出很是，续期失败，只打印日记和径直断绝任务         if (e != null) {           log.error("Can't update lock " + getRawName() + " expiration", e);           EXPIRATION_RENEWAL_MAP.remove(getEntryName());           return;         }         // 复返true讲解续期胜利，则递归调用续期方法（重新转机我方），续期失败诠释对应的锁还是不存在，径直复返，不再递归         if (res) {           // reschedule itself           renewExpiration();         } else {           cancelExpirationRenewal(null);         }       });     }// 这里的试验频率为leaseTime转机为ms单元下的三分之一，由于leaseTime动身点值为-1的情况下才会参预续期逻辑，那么这里的试验频率为lockWatchdogTimeout的三分之一   }, internalLockLeaseTime / 3, TimeUnit.MILLISECONDS);   // ExpirationEntry实例执有转机任求实例   ee.setTimeout(task); } 

中枢点2-订阅解锁音信：RedissonLock.subscribe

protected final LockPubSub pubSub;  public RedissonLock(CommandAsyncExecutor commandExecutor, String name) {   super(commandExecutor, name);   this.commandExecutor = commandExecutor;   this.internalLockLeaseTime = commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout();   //在构造器中动身点化pubSub,随着这几个get方法会发现他们都是在构造器中动身点化的，在PublishSubscribeService中会有   // private final AsyncSemaphore[] locks = new AsyncSemaphore[50]; 这样一段代码，动身点化了一组信号量   this.pubSub = commandExecutor.getConnectionManager().getSubscribeService().getLockPubSub(); }  protected RFuture<RedissonLockEntry> subscribe(long threadId) {   return pubSub.subscribe(getEntryName(), getChannelName()); }  // 在LockPubSub中注册一个entryName -> RedissonLockEntry的哈希映射，RedissonLockEntry实例中存放着RPromise<RedissonLockEntry>终结，一个信号量形势的锁和订阅方法重入计数器 public RFuture<E> subscribe(String entryName, String channelName) {   AsyncSemaphore semaphore = service.getSemaphore(new ChannelName(channelName));   RPromise<E> newPromise = new RedissonPromise<>();   semaphore.acquire(() -> {     if (!newPromise.setUncancellable()) {       semaphore.release();       return;     }      E entry = entries.get(entryName);     if (entry != null) {       entry.acquire();       semaphore.release();       entry.getPromise().onComplete(new TransferListener<E>(newPromise));       return;     }      E value = createEntry(newPromise);     value.acquire();      E oldValue = entries.putIfAbsent(entryName, value);     if (oldValue != null) {       oldValue.acquire();       semaphore.release();       oldValue.getPromise().onComplete(new TransferListener<E>(newPromise));       return;     }      RedisPubSubListener<Object> listener = createListener(channelName, value);     service.subscribe(LongCodec.INSTANCE, channelName, semaphore, listener);   });    return newPromise; } 

中枢点 3 相比毛糙，就不说了

解锁

RedissonLock.unlock()

@Override public void unlock() {   try {     // 获取刻下调用解锁操作的线程ID     get(unlockAsync(Thread.currentThread().getId()));   } catch (RedisException e) {     // IllegalMonitorStateException一般是A线程加锁，B线程解锁，里面判断线程情状不一致抛出的     if (e.getCause() instanceof IllegalMonitorStateException) {       throw (IllegalMonitorStateException) e.getCause();     } else {       throw e;     }   } } 

RedissonBaseLock.unlockAsync

@Override public RFuture<Void> unlockAsync(long threadId) {   // 构建一个终结RedissonPromise   RPromise<Void> result = new RedissonPromise<>();   // 复返的RFuture若是执有的终结为true，诠释解锁胜利，复返NULL诠释线程ID很是，加锁息争锁的客户端线程不是吞并个线程   RFuture<Boolean> future = unlockInnerAsync(threadId);    future.onComplete((opStatus, e) -> {     // 取消看门狗的续期任务     cancelExpirationRenewal(threadId);      if (e != null) {       result.tryFailure(e);       return;     }      if (opStatus == null) {       IllegalMonitorStateException cause = new IllegalMonitorStateException("attempt to unlock lock, not locked by current thread by node id: "                                                                             + id + " thread-id: " + threadId);       result.tryFailure(cause);       return;     }      result.trySuccess(null);   });    return result; } 

RedissonLock.unlockInnerAsync

// 信得过的里面解锁的方法，试验解锁的Lua剧本 protected RFuture<Boolean> unlockInnerAsync(long threadId) {   return evalWriteAsync(getRawName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,                         //若是分散式锁存在，然而value不匹配，示意锁还是被其他线程占用，无权开释锁，那么径直复返空值（解铃还须系铃东谈主）                         "if (redis.call('hexists', KEYS[1], ARGV[3]) == 0) then " +                         "return nil;" +                         "end; " +                         //若是value匹配，则即是刻下哨程占有分散式锁，那么将重入次数减1                         "local counter = redis.call('hincrby', KEYS[1], ARGV[3], -1); " +                         //重入次数减1后的值若是大于0，示意分散式锁有重入过，那么只可更新失效时期，还不成删除                         "if (counter > 0) then " +                         "redis.call('pexpire', KEYS[1], ARGV[2]); " +                         "return 0; " +                         "else " +                         //重入次数减1后的值若是为0，这时就不错删除这个KEY，并发布解锁音信，复返1                         "redis.call('del', KEYS[1]); " +                         "redis.call('publish', KEYS[2], ARGV[1]); " +                         "return 1; " +                         "end; " +                         "return nil;",                         //这5个参数分别对应KEYS[1]，KEYS[2]，ARGV[1]，ARGV[2]和ARGV[3]                         Arrays.asList(getRawName(), getChannelName()), LockPubSub.UNLOCK_MESSAGE, internalLockLeaseTime, getLockName(threadId)); } 

我只列出了一小部分代码，更多的内容照旧得我方动身点

从源码中，咱们不错看到 Redisson 帮咱们处理了抛出的第一个问题：失效时期设立多永劫期为好?

Redisson 提供了看门狗，每得到一个锁时，只设立一个很短的超时常期，同期起一个线程在每次将近到超时常期时去刷新锁的超时常期。在开释锁的同期终结这个线程。

然而莫得处理节点挂掉，丢失锁的问题，接着来~

咱们上边先容的分散式锁，在某些顶点情况下仍然是有弊端的

客户端永劫期内禁绝导致锁失效

客户端 1 得到了锁，因为收集问题或者 GC 等原因导致永劫期禁绝，然后业务次第还没试验完锁就逾期了，这时候客户端 2 也能日常拿到锁，可能会导致线程安全的问题。

Redis 职业器时钟漂移

若是 Redis 职业器的机器时期发生了上前杰出，就会导致这个 key 过早超时失效，比如说客户端 1 拿到锁后，key 还莫得到逾期时期，然而 Redis 职业器的时期比客户端快了 2 分钟，导致 key 提前就失效了，这时候，若是客户端 1 还莫得开释锁的话，就可能导致多个客户端同期执有吞并把锁的问题。

单点实例安全问题

若是 Redis 是单机模式的，若是挂了的话，那悉数的客户端都获取不到锁了，假定你是主从模式，但 Redis 的主从同步是异步进行的，若是 Redis 主宕机了，这个时候从机并莫得同步到这一把锁，那么机器 B 再次肯求的时候就会再次肯求到这把锁，这亦然问题

为了处理这些个问题 Redis 作家建议了 RedLock 红锁的算法，在 Redission 中也对 RedLock 进行了已矣。

Redis 官网对 redLock 算法的先容约莫如下：The Redlock algorithm

在分散式版块的算法里咱们假定咱们有 N 个 Redis master 节点，这些节点都是都备零丁的，咱们无须任何复制或者其他隐含的分散式结合机制。之前咱们还是描摹了在 Redis 单实例下如何安全地获取和开释锁。咱们确保将在每(N) 个实例上使用此方法获取和开释锁。在咱们的例子里面咱们设立 N=5，这是一个相比合理的设立，是以咱们需要在 5 台机器或者捏造机上头运行这些实例，这样保证他们不会同期都宕掉。为了取到锁，客户端应该试验以下操作:

获取刻下 Unix 时期，以毫秒为单元。

循序尝试从 5 个实例，使用疏导的 key 和具有唯独性的 value(举例UUID)获取锁。当向 Redis 请求获取锁时，客户端应该设立一个尝试从某个 Reids 实例获取锁的最大恭候时期(杰出这个时期，则立马接洽下一个实例)，这个超时常期应该小于锁的失效时期。举例你的锁自动失效时期为 10 秒，则超时常期应该在 5-50 毫秒之间。这样不错幸免职业器端 Redis 还是挂掉的情况下，客户端还在死死地恭候反映终结。若是职业器端莫得在国法时期内反映，客户端应该尽快尝试去另外一个 Redis 实例请求获取锁。

客户端使用刻下时期减去动身点获取锁时期(才略1纪录的时期)就得到获取锁耗尽的时期。当且仅当从大巨额(N/2+1，这里是3个节点)的 Redis 节点都取到锁，而且使用的总耗时小于锁失效时期时，锁才算获取胜利。

若是取到了锁，key 的信得过灵验时期 = 灵验时期(获取锁时设立的 key 的自动超时常期) - 获取锁的总耗时(接洽各个 Redis 实例的总耗时之和)(才略 3 筹画的终结)。

若是因为某些原因，最终获取锁失败(即莫得在至少 “N/2+1 ”个 Redis 实例取到锁或者“获取锁的总耗时”杰出了“灵验时期”)，客户端应该在悉数的 Redis 实例上进行解锁(即便某些 Redis 实例根蒂就莫得加锁胜利，这样不错辞谢某些节点获取到锁然而客户端莫得得到反映而导致接下来的一段时期不成被重新获取锁)。

追思下即是：

客户端在多个 Redis 实例上肯求加锁，必须保证大巨额节点加锁胜利

处理容错性问题，部分实例很是，剩下的还能加锁胜利

大巨额节点加锁的总耗时，要小于锁设立的逾期时期

多实例操作，可能存在收集延伸、丢包、超时等问题，是以就算是大巨额节点加锁胜利，若是加锁的积累耗时杰出了锁的逾期时期，那有些节点上的锁可能也还是失效了，照旧没挑升旨的

开释锁，要向一都节点发起开释锁请求

若是部分节点加锁胜利，但临了由于很是导致大部分节点没加锁胜利，就要开释掉悉数的，各节点要保执一致

对于 RedLock，两位分散式大佬，Antirez 和 Martin 还进行过一场争论，感钦慕的也不错望望

皇冠体育hg86a

Config config1 = new Config(); config1.useSingleServer().setAddress("127.0.0.1:6379"); RedissonClient redissonClient1 = Redisson.create(config1);  Config config2 = new Config(); config2.useSingleServer().setAddress("127.0.0.1:5378"); RedissonClient redissonClient2 = Redisson.create(config2);  Config config3 = new Config(); config3.useSingleServer().setAddress("127.0.0.1:5379"); RedissonClient redissonClient3 = Redisson.create(config3);  /**  * 获取多个 RLock 对象  */ RLock lock1 = redissonClient1.getLock(lockKey); RLock lock2 = redissonClient2.getLock(lockKey); RLock lock3 = redissonClient3.getLock(lockKey);  /**  * 凭证多个 RLock 对象构建 RedissonRedLock （最中枢的别离就在这里）  */ RedissonRedLock redLock = new RedissonRedLock(lock1, lock2, lock3);  try {     /**      * 4.尝试获取锁      * waitTimeout 尝试获取锁的最大恭候时期，杰出这个值，则认为获取锁失败      * leaseTime   锁的执偶然期,杰出这个时期锁会自动失效（值应设立为大于业务处理的时期，确保在锁灵验期内业务能处理完）      */     boolean res = redLock.tryLock(100, 10, TimeUnit.SECONDS);     if (res) {         //胜利得到锁，在这里处理业务     } } catch (Exception e) {     throw new RuntimeException("aquire lock fail"); }finally{     //不管如何, 临了都要解锁     redLock.unlock(); } 

最中枢的变化即是需要构建多个 RLock ，然后凭证多个 RLock 构建成一个 RedissonRedLock，因为 redLock 算法是栽植在多个相互零丁的 Redis 环境之上的(为了永别不错叫为 Redission node)，Redission node 节点既不错是单机模式(single)，也不错是主从模式(master/salve)，哨兵模式(sentinal)，或者集群模式(cluster)。这就意味着，不成跟以往这样只搭建 1个 cluster、或 1个 sentinel 集群，或是1套主从架构就了事了，需要为 RedissonRedLock 非常搭建多几套零丁的 Redission 节点。

RedissonMultiLock.tryLock

@Override public boolean tryLock(long waitTime, long leaseTime, TimeUnit unit) throws InterruptedException {   //        try {   //            return tryLockAsync(waitTime, leaseTime, unit).get();   //        } catch (ExecutionException e) {   //            throw new IllegalStateException(e);   //        }   long newLeaseTime = -1;   if (leaseTime != -1) {     if (waitTime == -1) {       newLeaseTime = unit.toMillis(leaseTime);     } else {       newLeaseTime = unit.toMillis(waitTime)*2;     }   }    long time = System.currentTimeMillis();   long remainTime = -1;   if (waitTime != -1) {     remainTime = unit.toMillis(waitTime);   }   long lockWaitTime = calcLockWaitTime(remainTime);    //允许加锁失败节点个数截止（N-(N/2+1)）   int failedLocksLimit = failedLocksLimit();   List<RLock> acquiredLocks = new ArrayList<>(locks.size());   // 遍历悉数节点通过EVAL号召试验lua加锁   for (ListIterator<RLock> iterator = locks.listIterator(); iterator.hasNext();) {     RLock lock = iterator.next();     boolean lockAcquired;     try {       // 对节点尝试加锁       if (waitTime == -1 && leaseTime == -1) {         lockAcquired = lock.tryLock();       } else {         long awaitTime = Math.min(lockWaitTime, remainTime);         lockAcquired = lock.tryLock(awaitTime, newLeaseTime, TimeUnit.MILLISECONDS);       }     } catch (RedisResponseTimeoutException e) {       // 若是抛出这类很是，为了辞谢加锁胜利，然而反映失败，需要解锁悉数节点       unlockInner(Arrays.asList(lock));       lockAcquired = false;     } catch (Exception e) {       lockAcquired = false;     }      if (lockAcquired) {       acquiredLocks.add(lock);     } else {       /*        *  筹画还是肯求锁失败的节点是否还是到达 允许加锁失败节点个数截止 （N-(N/2+1)）        * 若是还是到达， 就认定最终肯求锁失败，则莫得必要不竭从背面的节点肯求了        * 因为 Redlock 算法条目至少N/2+1 个节点都加锁胜利，才算最终的锁肯求胜利        */       if (locks.size() - acquiredLocks.size() == failedLocksLimit()) {         break;       }        if (failedLocksLimit == 0) {         unlockInner(acquiredLocks);         if (waitTime == -1) {           return false;         }         failedLocksLimit = failedLocksLimit();         acquiredLocks.clear();         // reset iterator         while (iterator.hasPrevious()) {           iterator.previous();         }       } else {         failedLocksLimit--;       }     }     //筹画 咫尺从各个节点获取锁还是耗尽的总时期，若是还是等于最大恭候时期，则认定最终肯求锁失败，复返false     if (remainTime != -1) {       remainTime -= System.currentTimeMillis() - time;       time = System.currentTimeMillis();       if (remainTime <= 0) {         unlockInner(acquiredLocks);         return false;       }     }   }    if (leaseTime != -1) {     acquiredLocks.stream()       .map(l -> (RedissonLock) l)       .map(l -> l.expireAsync(unit.toMillis(leaseTime), TimeUnit.MILLISECONDS))       .forEach(f -> f.syncUninterruptibly());   }    return true; } 

参考与感谢

《Redis —— Distributed locks with Redis》

《Redisson —— Distributed locks and synchronizers》

慢谈 Redis 已矣分散式锁 以及 Redisson 源码确认

领会Redisson均分散式锁的已矣