一、关于 在某些场景中，多个进程必须以互斥的方式独占共享资源，这时用分布式锁是最直接有效的。 随着技术快速发展，数据规模增大，分布式系统越来越普及，一个应用往往会部署在多台机器上（多节点），在有些场景中，为了保证数据不重复，要求在同一时刻，同一任务只在一个节点上运行，即保证某一方法同一时刻只能被一个线程执行。在单机环境中，应用是在同一进程下的，只需要保证单进程多线程环境中的线程安全性，通过 JAVA 提供的 volatile、ReentrantLock、synchronized 以及 concurrent 并发包下一些线程安全的类等就可以做到。而在多机部署环境中，不同机器不同进程，就需要在多进程下保证线程的安全性了。因此就需要分布式锁。 以往的工作中看到或用到几种实现方案，有基于 zk 的，也有基于 redis 的。由于实现上逻辑不严谨，线上时不时会爆出几个死锁 case。那么，什么样的分布式锁实现，才算是比较好的方案？ 常见分布式锁方案对比： 表格中对比了几种常见的方案，redis+lua 基本可应付工作中分布式锁的需求。然而，当偶然看到 redisson 分布式锁实现方案，相比以上方案，redisson 保持了简单易用、支持锁重入、支持阻塞等待、Lua 脚本原子操作。 分布式锁需满足四个条件： 首先，为了确保分布式锁可用，我们至少要确保锁的实现同时满足以下四个条件： 1、互斥性。在任意时刻，只有一个客户端能持有锁。 2、不会发生死锁。即使有一个客户端在持有锁的期间崩溃而没有主动解锁，也能保证后续其他客户端能加锁。 3、加锁和解锁必须是同一个客户端，客户端自己不能把别人加的锁给解了，即不能误解锁。 4、具有容错性。只要大多数 Redis 节点正常运行，客户端就能够获取和释放锁。 二、Redisson 分布式锁的实现 2.1、Redisson 分布式重入锁用法 Redisson 支持单点模式、主从模式、哨兵模式、集群模式，这里以单点模式为例： // 1、构造 redisson 实现分布式锁必要的 Config Config config = new Config (); config.useSingleServer ().setAddress ("redis://127.0.0.1:5379").setPassword ("123456").setDatabase (0); // 2、构造 RedissonClient RedissonClient redissonClient = Redisson.create (config); // 3、获取锁对象实例（无法保证是按线程的顺序获取到） RLock rLock = redissonClient.getLock (lockKey); try { /** * 4、尝试获取锁 * waitTimeout 尝试获取锁的最大等待时间，超过这个值，则认为获取锁失败 * leaseTime 锁的持有时间，超过这个时间锁会自动失效（值应设置为大于业务处理的时间，确保在锁有效期内业务能处理完） */ boolean res = rLock.tryLock ((long) waitTimeout, (long) leaseTime, TimeUnit.SECONDS); if (res) { // 成功获得锁，在这里处理业务 } } catch (Exception e) { throw new RuntimeException ("aquire lock fail"); } finally { // 无论如何，最后都要解锁 rLock.unlock (); } redisson 这个框架重度依赖了 Lua 脚本和 Netty，各种 Future 及 FutureListener 的异步、同步操作转换。 肯定要定义 2 个接口：加锁、解锁；大道至简，redisson 的作者就是在加锁和解锁的执行层面采用 Lua 脚本，逼格高，而且重要有原子性保证。加锁和解锁过程中还巧妙地利用了 redis 的发布订阅功能。 2.2、加锁 & 解锁 Lua 脚本 加锁、解锁 Lua 脚本是 redisson 分布式锁实现最重要的组成部分。 1、加锁Lua脚本 脚本入参： 脚本内容： -- 若锁不存在：则新增锁，并设置锁重入计数为1、设置锁过期时间 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; -- 若锁存在，且唯一标识也匹配：则表明当前加锁请求为锁重入请求，故锁重入计数+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; -- 若锁存在，但唯一标识不匹配：表明锁是被其他线程占用，当前线程无权解他人的锁，直接返回锁剩余过期时间 return redis.call('pttl', KEYS[1]); 脚本解读： Q：返回 nil、返回剩余过期时间有什么目的？ A：当且仅当返回 nil，才表示加锁成功；客户端需要感知加锁是否成功的结果。 2、解锁 Lua 脚本 脚本入参： 脚本内容： -- 若锁不存在：则直接广播解锁消息，并返回1 if (redis.call('exists', KEYS[1]) == 0) then redis.call('publish', KEYS[2], ARGV[1]); return 1; end; -- 若锁存在，但唯一标识不匹配：则表明锁被其他线程占用，当前线程不允许解锁其他线程持有的锁 if (redis.call('hexists', KEYS[1], ARGV[3]) == 0) then return nil; end; -- 若锁存在，且唯一标识匹配：则先将锁重入计数减1 local counter = redis.call('hincrby', KEYS[1], ARGV[3], -1); if (counter > 0) then -- 锁重入计数减1后还大于0：表明当前线程持有的锁还有重入，不能进行锁删除操作，但可以友好地帮忙设置下过期时期 redis.call('pexpire', KEYS[1], ARGV[2]); return 0; else -- 锁重入计数已为0：间接表明锁已释放了。直接删除掉锁，并广播解锁消息，去唤醒那些争抢过锁但还处于阻塞中的线程 redis.call('del', KEYS[1]); redis.call('publish', KEYS[2], ARGV[1]); return 1; end; return nil; 脚本解读: Q1：广播解锁消息有什么用？ A：是为了通知其他争抢锁阻塞住的线程，从阻塞中解除，并再次去争抢锁。 Q2：返回值 0、1、nil 有什么不一样？ A：当且仅当返回 1，才表示当前请求真正触发了解锁 Lua 脚本；但客户端又并不关心解锁请求的返回值，好像没什么用？ 三、源码 2.1、加锁流程源码 读加锁源码时，可以把 tryAcquire(leaseTime, unit, threadId) 方法直接视为执行加锁 Lua 脚本。直接进入 org.redisson.RedissonLock#tryLock(long, long, java.util.concurrent.TimeUnit) 源码 。 @Override public boolean tryLock(long waitTime, long leaseTime, TimeUnit unit) throws InterruptedException { // 获取锁能容忍的最大等待时长 long time = unit.toMillis(waitTime); long current = System.currentTimeMillis(); final long threadId = Thread.currentThread().getId(); // 【核心点1】尝试获取锁，若返回值为null，则表示已获取到锁 Long ttl = tryAcquire(leaseTime, unit, threadId); // lock acquired if (ttl == null) { return true; } // 还可以容忍的等待时长=获取锁能容忍的最大等待时长 - 执行完上述操作流逝的时间 time -= (System.currentTimeMillis() - current); if (time <= 0) { acquireFailed(threadId); return false; } current = System.currentTimeMillis(); // 【核心点2】订阅解锁消息，见org.redisson.pubsub.LockPubSub#onMessage /** * 4.订阅锁释放事件，并通过await方法阻塞等待锁释放，有效的解决了无效的锁申请浪费资源的问题： * 基于信息量，当锁被其它资源占用时，当前线程通过 Redis 的 channel 订阅锁的释放事件，一旦锁释放会发消息通知待等待的线程进行竞争 * 当 this.await返回false，说明等待时间已经超出获取锁最大等待时间，取消订阅并返回获取锁失败 * 当 this.await返回true，进入循环尝试获取锁 */ final RFuture<RedissonLockEntry> subscribeFuture = subscribe(threadId); //await 方法内部是用CountDownLatch来实现阻塞，获取subscribe异步执行的结果（应用了Netty 的 Future） if (!await(subscribeFuture, time, TimeUnit.MILLISECONDS)) { if (!subscribeFuture.cancel(false)) { subscribeFuture.addListener(new FutureListener<RedissonLockEntry>() { @Override public void operationComplete(Future<RedissonLockEntry> future) throws Exception { if (subscribeFuture.isSuccess()) { unsubscribe(subscribeFuture, threadId); } } }); } acquireFailed(threadId); return false; } // 订阅成功 try { // 还可以容忍的等待时长=获取锁能容忍的最大等待时长 - 执行完上述操作流逝的时间 time -= (System.currentTimeMillis() - current); if (time <= 0) { // 超出可容忍的等待时长，直接返回获取锁失败 acquireFailed(threadId); return false; } while (true) { long currentTime = System.currentTimeMillis(); // 尝试获取锁；如果锁被其他线程占用，就返回锁剩余过期时间【同上】 ttl = tryAcquire(leaseTime, unit, threadId); // lock acquired if (ttl == null) { return true; } time -= (System.currentTimeMillis() - currentTime); if (time <= 0) { acquireFailed(threadId); return false; } // waiting for message currentTime = System.currentTimeMillis(); // 【核心点3】根据锁TTL，调整阻塞等待时长； // 注意：这里实现非常巧妙，1、latch其实是个信号量Semaphore，调用其tryAcquire方法会让当前线程阻塞一段时间，避免了在while循环中频繁请求获取锁； //2、该Semaphore的release方法，会在订阅解锁消息的监听器消息处理方法org.redisson.pubsub.LockPubSub#onMessage调用；当其他线程释放了占用的锁，会广播解锁消息，监听器接收解锁消息，并释放信号量，最终会唤醒阻塞在这里的线程。 if (ttl >= 0 && ttl < time) { getEntry(threadId).getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS); } else { getEntry(threadId).getLatch().tryAcquire(time, TimeUnit.MILLISECONDS); } time -= (System.currentTimeMillis() - currentTime); if (time <= 0) { acquireFailed(threadId); return false; } } } finally { // 取消解锁消息的订阅 unsubscribe(subscribeFuture, threadId); } } 接下的再获取锁方法 tryAcquire 的实现，真的就是执行 Lua 脚本。 private Long tryAcquire(long leaseTime, TimeUnit unit, long threadId) { // tryAcquireAsync异步执行Lua脚本，get方法同步获取返回结果 return get(tryAcquireAsync(leaseTime, unit, threadId)); } // 见org.redisson.RedissonLock#tryAcquireAsync private <T> RFuture<Long> tryAcquireAsync(long leaseTime, TimeUnit unit, final long threadId) { if (leaseTime != -1) { // 实质是异步执行加锁Lua脚本 return tryLockInnerAsync(leaseTime, unit, threadId, RedisCommands.EVAL_LONG); } RFuture<Long> ttlRemainingFuture = tryLockInnerAsync(commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout(), TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_LONG); ttlRemainingFuture.addListener(new FutureListener<Long>() { @Override public void operationComplete(Future<Long> future) throws Exception { //先判断这个异步操作有没有执行成功，如果没有成功，直接返回，如果执行成功了，就会同步获取结果 if (!future.isSuccess()) { return; } Long ttlRemaining = future.getNow(); // lock acquired //如果ttlRemaining为null，则会执行一个定时调度的方法scheduleExpirationRenewal if (ttlRemaining == null) { scheduleExpirationRenewal(threadId); } } }); return ttlRemainingFuture; } // 见org.redisson.RedissonLock#tryLockInnerAsync <T> RFuture<T> tryLockInnerAsync(long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> 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.<Object>singletonList(getName()), internalLockLeaseTime, getLockName(threadId)); } 2.2、加锁过程小结 1、锁其实也是一种资源，各线程争抢锁操作对应到 redisson 中就是争抢着去创建一个 hash 结构，谁先创建就代表谁获得锁；hash 的名称为锁名，hash 里面内容仅包含一条键值对，键为 redisson 客户端唯一标识 + 持有锁线程 id，值为锁重入计数；给 hash 设置的过期时间就是锁的过期时间。放个图直观感受下： 2、加锁流程核心就 3 步 Step1：尝试获取锁，这一步是通过执行加锁 Lua 脚本来做； Step2：若第一步未获取到锁，则去订阅解锁消息，当获取锁到剩余过期时间后，调用信号量方法阻塞住，直到被唤醒或等待超时。 Step3：一旦持有锁的线程释放了锁，就会广播解锁消息。于是，第二步中的解锁消息的监听器会释放信号量，获取锁被阻塞的那些线程就会被唤醒，并重新尝试获取锁。 比如 RedissonLock 中的变量 internalLockLeaseTime, 默认值是 30000 毫秒，还有调用 tryLockInnerAsync () 传入的一个从连接管理器获取的 getLockWatchdogTimeout (), 他的默认值也是 30000 毫秒，这些都和 redisson 官方文档所说的 watchdog 机制有关，看门狗，还是很形象的描述这一机制，那么看门狗到底做了什么，为什么这么做，来看下核心代码。 先思考一个问题，假设在一个分布式环境下，多个服务实例请求获取锁，其中服务实例 1 成功获取到了锁，在执行业务逻辑的过程中，服务实例突然挂掉了或者 hang 住了，那么这个锁会不会释放，什么时候释放？回答这个问题，自然想起来之前我们分析的 lua 脚本，其中第一次加锁的时候使用 pexpire 给锁 key 设置了过期时间，默认 30000 毫秒，由此来看如果服务实例宕机了，锁最终也会释放，其他服务实例也是可以继续获取到锁执行业务。但是要是 30000 毫秒之后呢，要是服务实例 1 没有宕机但是业务执行还没有结束，锁释放掉了就会导致线程问题，这个 redisson 是怎么解决的呢？这个就一定要实现自动延长锁有效期的机制。 异步执行完 lua 脚本执行完成之后，设置了一个监听器，来处理异步执行结束之后的一些工作。在操作完成之后会去执行 operationComplete 方法，先判断这个异步操作有没有执行成功，如果没有成功，直接返回，如果执行成功了，就会同步获取结果，如果 ttlRemaining 为 null，则会执行一个定时调度的方法 scheduleExpirationRenewal，回想一下之前的 lua 脚本，当加锁逻辑。 处理结束，返回了一个 nil; 如此说来 就一定会走定时任务了。来看下定时调度 scheduleExpirationRenewal 代码。 private void scheduleExpirationRenewal(final long threadId) { if (expirationRenewalMap.containsKey(getEntryName())) { return; } Timeout task = commandExecutor.getConnectionManager().newTimeout(new TimerTask() { @Override public void run(Timeout timeout) throws Exception { RFuture<Boolean> future = commandExecutor.evalWriteAsync(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.<Object>singletonList(getName()), internalLockLeaseTime, getLockName(threadId)); future.addListener(new FutureListener<Boolean>() { @Override public void operationComplete(Future<Boolean> future) throws Exception { expirationRenewalMap.remove(getEntryName()); if (!future.isSuccess()) { log.error("Can't update lock " + getName() + " expiration", future.cause()); return; } if (future.getNow()) { // reschedule itself scheduleExpirationRenewal(threadId); } } }); } }, internalLockLeaseTime / 3, TimeUnit.MILLISECONDS); if (expirationRenewalMap.putIfAbsent(getEntryName(), task) != null) { task.cancel(); } } 首先，会先判断在 expirationRenewalMap 中是否存在了 entryName，这是个 map 结构，主要还是判断在这个服务实例中的加锁客户端的锁 key 是否存在，如果已经存在了，就直接返回；第一次加锁，肯定是不存在的，接下来就是搞了一个 TimeTask，延迟 internalLockLeaseTime/3 之后执行，这里就用到了文章一开始就提到奇妙的变量，算下来就是大约 10 秒钟执行一次，调用了一个异步执行的方法。 如图也是调用异步执行了一段 lua 脚本，首先判断这个锁 key 的 map 结构中是否存在对应的 key8a9649f5-f5b5-48b4-beaa-d0c24855f9ab:anyLock:1, 如果存在，就直接调用 pexpire 命令设置锁 key 的过期时间，默认 30000 毫秒。 OK，现在思路就清晰了，在上面任务调度的方法中，也是异步执行并且设置了一个监听器，在操作执行成功之后，会回调这个方法，如果调用失败会打一个错误日志并返回，更新锁过期时间失败；然后获取异步执行的结果，如果为 true，就会调用本身，如此说来又会延迟 10 秒钟去执行这段逻辑，所以，这段逻辑在你成功获取到锁之后，会每隔十秒钟去执行一次，并且，在锁 key 还没有失效的情况下，会把锁的过期时间继续延长到 30000 毫秒，也就是说只要这台服务实例没有挂掉，并且没有主动释放锁，看门狗都会每隔十秒给你续约一下，保证锁一直在你手中。完美的操作。 到现在来说，加锁，锁自动延长过期时间，都 OK 了，然后就是说在你执行业务，持有锁的这段时间，别的服务实例来尝试加锁又会发生什么情况呢？或者当前客户端的别的线程来获取锁呢？很显然，肯定会阻塞住，我们来通过代码看看是怎么做到的。还是把眼光放到之前分析的那段加锁 lua 代码上，当加锁的锁 key 存在的时候并且锁 key 对应的 map 结构中当前客户端的唯一 key 也存在时，会去调用 hincrby 命令，将唯一 key 的值自增一，并且会 pexpire 设置 key 的过期时间为 30000 毫秒，然后返回 nil，可以想象这里也是加锁成功的，也会继续去执行定时调度任务，完成锁 key 过期时间的续约，这里呢，就实现了锁的可重入性。 那么当以上这种情况也没有发生呢，这里就会直接返回当前锁的剩余有效期，相应的也不会去执行续约逻辑。此时一直返回到上面的方法，如果加锁成功就直接返回；否则就会进入一个死循环，去尝试加锁，并且也会在等待一段时间之后一直循环尝试加锁，阻塞住，直到第一个服务实例释放锁。对于不同的服务实例尝试会获取一把锁，也和上面的逻辑类似，都是这样实现了锁的互斥。 紧接着，我们来看看锁释放的逻辑，其实也很简单，调用了 lock.unlock () 方法，跟着代码走流程发现，也是异步调用了一段 lua 脚本，lua 脚本，应该就比较清晰，也就是通过判断锁 key 是否存在，如果不存在直接返回；否则就会判断当前客户端对应的唯一 key 的值是否存在，如果不存在就会返回 nil; 否则，值自增 - 1，判断唯一 key 的值是否大于零，如果大于零，则返回 0；否则删除当前锁 key，并返回 1；返回到上一层方法，也是针对返回值进行了操作，如果返回值是 1，则会去取消之前的定时续约任务，如果失败了，则会做一些类似设置状态的操作，这一些和解锁逻辑也没有什么关系，可以不去看他。 2.3、解锁流程源码 解锁流程相对比较简单，完全就是执行解锁 Lua 脚本，无额外的代码逻辑，直接看 org.redisson.RedissonLock#unlock 代码。 @Override public void unlock() { // 执行解锁Lua脚本，这里传入线程id，是为了保证加锁和解锁是同一个线程，避免误解锁其他线程占有的锁 Boolean opStatus = get(unlockInnerAsync(Thread.currentThread().getId())); if (opStatus == null) { throw new IllegalMonitorStateException("attempt to unlock lock, not locked by current thread by node id: " + id + " thread-id: " + Thread.currentThread().getId()); } if (opStatus) { cancelExpirationRenewal(); } } // 见org.redisson.RedissonLock#unlockInnerAsync protected RFuture<Boolean> unlockInnerAsync(long threadId) { return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN, "if (redis.call('exists', KEYS[1]) == 0) then " + "redis.call('publish', KEYS[2], ARGV[1]); " + "return 1; " + "end;" + "if (redis.call('hexists', KEYS[1], ARGV[3]) == 0) then " + "return nil;" + "end; " + "local counter = redis.call('hincrby', KEYS[1], ARGV[3], -1); " + "if (counter > 0) then " + "redis.call('pexpire', KEYS[1], ARGV[2]); " + "return 0; " + "else " + "redis.call('del', KEYS[1]); " + "redis.call('publish', KEYS[2], ARGV[1]); " + "return 1; "+ "end; " + "return nil;", Arrays.<Object>asList(getName(), getChannelName()), LockPubSub.unlockMessage, internalLockLeaseTime, getLockName(threadId)); } 2.4、加锁 & 解锁流程串起来 上面结合 Lua 脚本和源码，分别分析了加锁流程和解锁流程。下面升级下挑战难度，模拟下多个线程争抢锁会是怎样的流程。示意图如下，比较关键的三处已用红色字体标注。 概括下整个流程： 1、线程 A 和线程 B 两个线程同时争抢锁。线程 A 很幸运，最先抢到了锁。线程 B 在获取锁失败后，并未放弃希望，而是主动订阅了解锁消息，然后再尝试获取锁，顺便看看没有抢到的这把锁还有多久就过期，线程 B 就按需阻塞等锁释放。 2、线程 A 拿着锁干完了活，自觉释放了持有的锁，于此同时广播了解锁消息，通知其他抢锁的线程再来枪； 3、解锁消息的监听者 LockPubSub 收到消息后，释放自己持有的信号量；线程 B 就瞬间从阻塞中被唤醒了，接着再抢锁，这次终于抢到锁了！后面再按部就班，干完活，解锁 2.5、其他 Q1：订阅频道名称 (如:redisson_lock__channel:{my_first_lock_name}) 为什么有大括号？ A： 1、在 redis 集群方案中，如果 Lua 脚本涉及多个 key 的操作，则需限制这些 key 在同一个 slot 中，才能保障 Lua 脚本执行的原子性。否则运行会报错 Lua script attempted to access a non local key in a cluster node . channel； 2、HashTag 是用 {} 包裹 key 的一个子串，若设置了 HashTag，集群会根据 HashTag 决定 key 分配到哪个 slot；HashTag 不支持嵌套，只有第一个左括号 {和第一个右括号} 里面的内容才当做 HashTag 参与 slot 计算；通常，客户端都会封装这个计算逻辑。 // 见 org.redisson.cluster.ClusterConnectionManager#calcSlot @Override public int calcSlot(String key) { if (key == null) { return 0; } int start = key.indexOf('{'); if (start != -1) { int end = key.indexOf('}'); key = key.substring(start+1, end); } int result = CRC16.crc16(key.getBytes()) % MAX_SLOT; log.debug("slot {} for {}", result, key); return result; } 3、在解锁 Lua 脚本中，操作了两个 key：一个是锁名 my_lock_name，一个是解锁消息发布订阅频道 redisson_lock__channel:{my_first_lock_name}，按照上面 slot 计算方式，两个 key 都会按照内容 my_first_lock_name 来计算，故能保证落到同一个 slot Q2：redisson 代码几乎都是以 Lua 脚本方式与 redis 服务端交互，如何跟踪这些脚本执行过程？ A：启动一个 redis 客户端终端，执行 monitor 命令以便在终端上实时打印 redis 服务器接收到的命令；然后 debug 执行 redisson 加锁 / 解锁测试用例，即可看到代码运行过程中实际执行了哪些 Lua 脚本。 eg：上面整体流程示意图的测试用例位: @RunWith(SpringRunner.class) @SpringBootTest public class RedissonDistributedLockerTest { private static final Logger log = LoggerFactory.getLogger(RedissonDistributedLocker.class); @Resource private DistributedLocker distributedLocker; private static final ExecutorService executorServiceB = Executors.newSingleThreadExecutor(); private static final ExecutorService executorServiceC = Executors.newSingleThreadExecutor(); @Test public void tryLockUnlockCost() throws Exception { StopWatch stopWatch = new StopWatch("加锁解锁耗时统计"); stopWatch.start(); for (int i = 0; i < 10000; i++) { String key = "mock-key:" + UUID.randomUUID().toString().replace("-", ""); Optional<LockResource> optLocked = distributedLocker.tryLock(key, 600000, 600000); Assert.assertTrue(optLocked.isPresent()); optLocked.get().unlock(); } stopWatch.stop(); log.info(stopWatch.prettyPrint()); } @Test public void tryLock() throws Exception { String key = "mock-key:" + UUID.randomUUID().toString().replace("-", ""); Optional<LockResource> optLocked = distributedLocker.tryLock(key, 600000, 600000); Assert.assertTrue(optLocked.isPresent()); Optional<LockResource> optLocked2 = distributedLocker.tryLock(key, 600000, 600000); Assert.assertTrue(optLocked2.isPresent()); optLocked.get().unlock(); } /** * 模拟2个线程争抢锁：A先获取到锁，A释放锁后，B再获得锁 */ @Test public void tryLock2() throws Exception { String key = "mock-key:" + UUID.randomUUID().toString().replace("-", ""); CountDownLatch countDownLatch = new CountDownLatch(1); Future<Optional<LockResource>> submit = executorServiceB.submit(() -> { countDownLatch.await(); log.info("B尝试获得锁：thread={}", currentThreadId()); return distributedLocker.tryLock(key, 600000, 600000); } ); log.info("A尝试获得锁：thread={}", currentThreadId()); Optional<LockResource> optLocked = distributedLocker.tryLock(key, 300000, 600000); Assert.assertTrue(optLocked.isPresent()); log.info("A已获得锁：thread={}", currentThreadId()); countDownLatch.countDown(); optLocked.get().unlock(); log.info("A已释放锁：thread={}", currentThreadId()); Optional<LockResource> lockResource2 = submit.get(); Assert.assertTrue(lockResource2.isPresent()); executorServiceB.submit(() -> { log.info("B已获得锁：thread={}", currentThreadId()); lockResource2.get().unlock(); log.info("B已释放锁：thread={}", currentThreadId()); }); } /** * 模拟3个线程争抢锁：A先获取到锁，A释放锁后，B和C同时争抢锁 */ @Test public void tryLock3() throws Exception { String key = "mock-key:" + UUID.randomUUID().toString().replace("-", ""); log.info("A尝试获得锁：thread={}", currentThreadId()); Optional<LockResource> optLocked = distributedLocker.tryLock(key, 600000, 600000); if (optLocked.isPresent()) { log.info("A已获得锁：thread={}", currentThreadId()); } Assert.assertTrue(optLocked.isPresent()); CyclicBarrier cyclicBarrier = new CyclicBarrier(2); Future<Optional<LockResource>> submitB = executorServiceB.submit(() -> { cyclicBarrier.await(); log.info("B尝试获得锁：thread={}", currentThreadId()); return distributedLocker.tryLock(key, 600000, 600000); } ); Future<Optional<LockResource>> submitC = executorServiceC.submit(() -> { cyclicBarrier.await(); log.info("C尝试获得锁：thread={}", currentThreadId()); return distributedLocker.tryLock(key, 600000, 600000); } ); optLocked.get().unlock(); log.info("A已释放锁：thread={}", currentThreadId()); CountDownLatch countDownLatch = new CountDownLatch(2); executorServiceB.submit(() -> { log.info("B已获得锁：thread={}", currentThreadId()); try { submitB.get().get().unlock(); } catch (InterruptedException | ExecutionException e) { e.printStackTrace(); } log.info("B已释放锁：thread={}", currentThreadId()); countDownLatch.countDown(); }); executorServiceC.submit(() -> { log.info("C已获得锁：thread={}", currentThreadId()); try { submitC.get().get().unlock(); } catch (InterruptedException | ExecutionException e) { e.printStackTrace(); } log.info("C已释放锁：thread={}", currentThreadId()); countDownLatch.countDown(); }); countDownLatch.await(); } private static Long currentThreadId() { return Thread.currentThread().getId(); } @Test public void tryLockWaitTimeout() throws Exception { String key = "mock-key:" + UUID.randomUUID().toString(); Optional<LockResource> optLocked = distributedLocker.tryLock(key, 10, 2000); Assert.assertTrue(optLocked.isPresent()); Optional<LockResource> optLockResource = CompletableFuture.supplyAsync(() -> { long now = System.currentTimeMillis(); Optional<LockResource> optLockedAgain = distributedLocker.tryLock(key, 1000, 10); long cost = System.currentTimeMillis() - now; log.info("cost={}", cost); return optLockedAgain; }).exceptionally(th -> { log.error("Exception: ", th); return Optional.empty(); }).join(); Assert.assertTrue(!optLockResource.isPresent()); } @Test public void tryLockWithLeaseTime() throws Exception { String key = "mock-key-with-leaseTime:" + UUID.randomUUID().toString(); Optional<LockResource> optLocked = distributedLocker.tryLock(key, 3000, 1000); Assert.assertTrue(optLocked.isPresent()); // 可重入 Optional<LockResource> optLockedAgain = distributedLocker.tryLock(key, 3000, 1000); Assert.assertTrue(optLockedAgain.isPresent()); } /** * 模拟1000个并发请求枪一把锁 */ @Test public void tryLockWithLeaseTimeOnMultiThread() throws Exception { int totalThread = 1000; String key = "mock-key-with-leaseTime:" + UUID.randomUUID().toString(); AtomicInteger tryAcquireLockTimes = new AtomicInteger(0); AtomicInteger acquiredLockTimes = new AtomicInteger(0); ExecutorService executor = Executors.newFixedThreadPool(totalThread); for (int i = 0; i < totalThread; i++) { executor.submit(new Runnable() { @Override public void run() { tryAcquireLockTimes.getAndIncrement(); Optional<LockResource> optLocked = distributedLocker.tryLock(key, 10, 10000); if (optLocked.isPresent()) { acquiredLockTimes.getAndIncrement(); } } }); } executor.awaitTermination(15, TimeUnit.SECONDS); Assert.assertTrue(tryAcquireLockTimes.get() == totalThread); Assert.assertTrue(acquiredLockTimes.get() == 1); } @Test public void tryLockWithLeaseTimeOnMultiThread2() throws Exception { int totalThread = 100; String key = "mock-key-with-leaseTime:" + UUID.randomUUID().toString(); AtomicInteger tryAcquireLockTimes = new AtomicInteger(0); AtomicInteger acquiredLockTimes = new AtomicInteger(0); ExecutorService executor = Executors.newFixedThreadPool(totalThread); for (int i = 0; i < totalThread; i++) { executor.submit(new Runnable() { @Override public void run() { long now = System.currentTimeMillis(); Optional<LockResource> optLocked = distributedLocker.tryLock(key, 10000, 5); long cost = System.currentTimeMillis() - now; log.info("tryAcquireLockTimes={}||wait={}", tryAcquireLockTimes.incrementAndGet(), cost); if (optLocked.isPresent()) { acquiredLockTimes.getAndIncrement(); // 主动释放锁 optLocked.get().unlock(); } } }); } executor.awaitTermination(20, TimeUnit.SECONDS); log.info("tryAcquireLockTimes={}, acquireLockTimes={}", tryAcquireLockTimes.get(), acquiredLockTimes.get()); Assert.assertTrue(tryAcquireLockTimes.get() == totalThread); Assert.assertTrue(acquiredLockTimes.get() == totalThread); } } public interface DistributedLocker { Optional<LockResource> tryLock(String lockKey, int waitTime); Optional<LockResource> tryLock(String lockKey, int waitTime, int leaseTime); } public interface LockResource { void unlock(); } 执行的 Lua 脚本如下: 加锁：redissonClient.getLock("my_first_lock_name").tryLock(600000, 600000); 解锁：redissonClient.getLock("my_first_lock_name").unlock(); 其中： # 线程A ## 1.1.1尝试获取锁 -> 成功 1568357723.205362 [0 127.0.0.1:56419] "EVAL" "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]);" "1" "my_first_lock_name" "600000" "58c62432-bb74-4d14-8a00-9908cc8b828f:1" 1568357723.205452 [0 lua] "exists" "my_first_lock_name" 1568357723.208858 [0 lua] "hset" "my_first_lock_name" "58c62432-bb74-4d14-8a00-9908cc8b828f:1" "1" 1568357723.208874 [0 lua] "pexpire" "my_first_lock_name" "600000" # 线程B ### 2.1.1尝试获取锁，未获取到，返回锁剩余过期时间 1568357773.338018 [0 127.0.0.1:56417] "EVAL" "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]);" "1" "my_first_lock_name" "600000" "58c62432-bb74-4d14-8a00-9908cc8b828f:26" 1568357773.338161 [0 lua] "exists" "my_first_lock_name" 1568357773.338177 [0 lua] "hexists" "my_first_lock_name" "58c62432-bb74-4d14-8a00-9908cc8b828f:26" 1568357773.338197 [0 lua] "pttl" "my_first_lock_name" ## 2.1.1.3 添加订阅（非Lua脚本） -> 订阅成功 1568357799.403341 [0 127.0.0.1:56421] "SUBSCRIBE" "redisson_lock__channel:{my_first_lock_name}" ## 2.1.1.4 再次尝试获取锁 -> 未获取到，返回锁剩余过期时间 1568357830.683631 [0 127.0.0.1:56418] "EVAL" "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]);" "1" "my_first_lock_name" "600000" "58c62432-bb74-4d14-8a00-9908cc8b828f:26" 1568357830.684371 [0 lua] "exists" "my_first_lock_name" 1568357830.684428 [0 lua] "hexists" "my_first_lock_name" "58c62432-bb74-4d14-8a00-9908cc8b828f:26" 1568357830.684485 [0 lua] "pttl" "my_first_lock_name" # 线程A ## 3.1.1 释放锁并广播解锁消息，0代表解锁消息 1568357922.122454 [0 127.0.0.1:56420] "EVAL" "if (redis.call('exists', KEYS[1]) == 0) then redis.call('publish', KEYS[2], ARGV[1]); return 1; end;if (redis.call('hexists', KEYS[1], ARGV[3]) == 0) then return nil;end; local counter = redis.call('hincrby', KEYS[1], ARGV[3], -1); if (counter > 0) then redis.call('pexpire', KEYS[1], ARGV[2]); return 0; else redis.call('del', KEYS[1]); redis.call('publish', KEYS[2], ARGV[1]); return 1; end; return nil;" "2" "my_first_lock_name" "redisson_lock__channel:{my_first_lock_name}" "0" "30000" "58c62432-bb74-4d14-8a00-9908cc8b828f:1" 1568357922.123645 [0 lua] "exists" "my_first_lock_name" 1568357922.123701 [0 lua] "hexists" "my_first_lock_name" "58c62432-bb74-4d14-8a00-9908cc8b828f:1" 1568357922.123741 [0 lua] "hincrby" "my_first_lock_name" "58c62432-bb74-4d14-8a00-9908cc8b828f:1" "-1" 1568357922.123775 [0 lua] "del" "my_first_lock_name" 1568357922.123799 [0 lua] "publish" "redisson_lock__channel:{my_first_lock_name}" "0" # 线程B ## 监听到解锁消息消息 -> 释放信号量，阻塞被解除；4.1.1.1 再次尝试获取锁 -> 获取成功 1568357975.015206 [0 127.0.0.1:56419] "EVAL" "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]);" "1" "my_first_lock_name" "600000" "58c62432-bb74-4d14-8a00-9908cc8b828f:26" 1568357975.015579 [0 lua] "exists" "my_first_lock_name" 1568357975.015633 [0 lua] "hset" "my_first_lock_name" "58c62432-bb74-4d14-8a00-9908cc8b828f:26" "1" 1568357975.015721 [0 lua] "pexpire" "my_first_lock_name" "600000" ## 4.1.1.3 取消订阅（非Lua脚本） 1568358031.185226 [0 127.0.0.1:56421] "UNSUBSCRIBE" "redisson_lock__channel:{my_first_lock_name}" # 线程B ## 5.1.1 释放锁并广播解锁消息 1568358255.551896 [0 127.0.0.1:56417] "EVAL" "if (redis.call('exists', KEYS[1]) == 0) then redis.call('publish', KEYS[2], ARGV[1]); return 1; end;if (redis.call('hexists', KEYS[1], ARGV[3]) == 0) then return nil;end; local counter = redis.call('hincrby', KEYS[1], ARGV[3], -1); if (counter > 0) then redis.call('pexpire', KEYS[1], ARGV[2]); return 0; else redis.call('del', KEYS[1]); redis.call('publish', KEYS[2], ARGV[1]); return 1; end; return nil;" "2" "my_first_lock_name" "redisson_lock__channel:{my_first_lock_name}" "0" "30000" "58c62432-bb74-4d14-8a00-9908cc8b828f:26" 1568358255.552125 [0 lua] "exists" "my_first_lock_name" 1568358255.552156 [0 lua] "hexists" "my_first_lock_name" "58c62432-bb74-4d14-8a00-9908cc8b828f:26" 1568358255.552200 [0 lua] "hincrby" "my_first_lock_name" "58c62432-bb74-4d14-8a00-9908cc8b828f:26" "-1" 1568358255.552258 [0 lua] "del" "my_first_lock_name" 1568358255.552304 [0 lua] "publish" "redisson_lock__channel:{my_first_lock_name}" "0" 需要特别注意的是，RedissonLock 同样没有解决 节点挂掉的时候，存在丢失锁的风险的问题。而现实情况是有一些场景无法容忍的，所以 Redisson 提供了实现了 redlock 算法的 RedissonRedLock，RedissonRedLock 真正解决了单点失败的问题，代价是需要额外的为 RedissonRedLock 搭建 Redis 环境。 所以，如果业务场景可以容忍这种小概率的错误，则推荐使用 RedissonLock， 如果无法容忍，则推荐使用 RedissonRedLock。 三、redlock算法 Redis 官网对 redLock 算法的介绍大致如下： The Redlock algorithm 在分布式版本的算法里我们假设我们有 N 个 Redis master 节点，这些节点都是完全独立的，我们不用任何复制或者其他隐含的分布式协调机制。之前我们已经描述了在 Redis 单实例下怎么安全地获取和释放锁。我们确保将在每（N) 个实例上使用此方法获取和释放锁。在我们的例子里面我们把 N 设成 5，这是一个比较合理的设置，所以我们需要在 5 台机器上面或者 5 台虚拟机上面运行这些实例，这样保证他们不会同时都宕掉。为了取到锁，客户端应该执行以下操作: 1、获取当前 Unix 时间，以毫秒为单位。 2、依次尝试从 5 个实例，使用相同的 key 和具有唯一性的 value（例如 UUID）获取锁。当向 Redis 请求获取锁时，客户端应该设置一个尝试从某个 Reids 实例获取锁的最大等待时间（超过这个时间，则立马询问下一个实例），这个超时时间应该小于锁的失效时间。例如你的锁自动失效时间为 10 秒，则超时时间应该在 5-50 毫秒之间。这样可以避免服务器端 Redis 已经挂掉的情况下，客户端还在死死地等待响应结果。如果服务器端没有在规定时间内响应，客户端应该尽快尝试去另外一个 Redis 实例请求获取锁。 3、客户端使用当前时间减去开始获取锁时间（步骤 1 记录的时间）就得到获取锁消耗的时间。当且仅当从大多数（N/2+1，这里是 3 个节点）的 Redis 节点都取到锁，并且使用的总耗时小于锁失效时间时，锁才算获取成功。 4、如果取到了锁，key 的真正有效时间 = 有效时间（获取锁时设置的 key 的自动超时时间） - 获取锁的总耗时（询问各个 Redis 实例的总耗时之和）（步骤 3 计算的结果）。 5、如果因为某些原因，最终获取锁失败（即没有在至少 “N/2+1” 个 Redis 实例取到锁或者 “获取锁的总耗时” 超过了 “有效时间”），客户端应该在所有的 Redis 实例上进行解锁（即便某些 Redis 实例根本就没有加锁成功，这样可以防止某些节点获取到锁但是客户端没有得到响应而导致接下来的一段时间不能被重新获取锁）。 用 Redisson 实现分布式锁 (红锁 RedissonRedLock) 及源码分析（实现三） 这里以三个单机模式为例，需要特别注意的是他们完全互相独立，不存在主从复制或者其他集群协调机制。 Config config1 = new Config(); config1.useSingleServer().setAddress("redis://172.0.0.1:5378").setPassword("a123456").setDatabase(0); RedissonClient redissonClient1 = Redisson.create(config1); Config config2 = new Config(); config2.useSingleServer().setAddress("redis://172.0.0.1:5379").setPassword("a123456").setDatabase(0); RedissonClient redissonClient2 = Redisson.create(config2); Config config3 = new Config(); config3.useSingleServer().setAddress("redis://172.0.0.1:5380").setPassword("a123456").setDatabase(0); 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((long)waitTimeout, (long)leaseTime, 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 节点。 比如可以搭建 3 个 或者 5 个 Redission 节点，具体可看视资源及业务情况而定。 下图是一个利用多个 Redission node 最终 组成 RedLock 分布式锁的例子，需要特别注意的是每个 Redission node 是互相独立的，不存在任何复制或者其他隐含的分布式协调机制。 四、Redisson 实现 redlock 算法源码分析（RedLock） 加锁核心代码 org.redisson.RedissonMultiLock#tryLock public boolean tryLock(long waitTime, long leaseTime, TimeUnit unit) throws InterruptedException { long newLeaseTime = -1; if (leaseTime != -1) { newLeaseTime = unit.toMillis(waitTime)*2; } long time = System.currentTimeMillis(); long remainTime = -1; if (waitTime != -1) { remainTime = unit.toMillis(waitTime); } long lockWaitTime = calcLockWaitTime(remainTime); /** * 1. 允许加锁失败节点个数限制（N-(N/2+1)） */ int failedLocksLimit = failedLocksLimit(); /** * 2. 遍历所有节点通过EVAL命令执行lua加锁 */ List<RLock> acquiredLocks = new ArrayList<>(locks.size()); for (ListIterator<RLock> iterator = locks.listIterator(); iterator.hasNext();) { RLock lock = iterator.next(); boolean lockAcquired; /** * 3.对节点尝试加锁 */ 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) { /** *4. 如果获取到锁则添加到已获取锁集合中 */ acquiredLocks.add(lock); } else { /** * 5. 计算已经申请锁失败的节点是否已经到达 允许加锁失败节点个数限制 （N-(N/2+1)） * 如果已经到达， 就认定最终申请锁失败，则没有必要继续从后面的节点申请了 * 因为 Redlock 算法要求至少N/2+1 个节点都加锁成功，才算最终的锁申请成功 */ if (locks.size() - acquiredLocks.size() == failedLocksLimit()) { break; } if (failedLocksLimit == 0) { unlockInner(acquiredLocks); if (waitTime == -1 && leaseTime == -1) { return false; } failedLocksLimit = failedLocksLimit(); acquiredLocks.clear(); // reset iterator while (iterator.hasPrevious()) { iterator.previous(); } } else { failedLocksLimit--; } } /** * 6.计算 目前从各个节点获取锁已经消耗的总时间，如果已经等于最大等待时间，则认定最终申请锁失败，返回false */ if (remainTime != -1) { remainTime -= System.currentTimeMillis() - time; time = System.currentTimeMillis(); if (remainTime <= 0) { unlockInner(acquiredLocks); return false; } } } if (leaseTime != -1) { List<RFuture<Boolean>> futures = new ArrayList<>(acquiredLocks.size()); for (RLock rLock : acquiredLocks) { RFuture<Boolean> future = ((RedissonLock) rLock).expireAsync(unit.toMillis(leaseTime), TimeUnit.MILLISECONDS); futures.add(future); } for (RFuture<Boolean> rFuture : futures) { rFuture.syncUninterruptibly(); } } /** * 7.如果逻辑正常执行完则认为最终申请锁成功，返回true */ return true; }
一、需具备的条件 1、获取 / 释放锁性能要好。 2、原子性。 3、网络中断或宕机无法释放锁时，锁必须被清除，不然会发生死锁。 4、可重入一个线程中可以多次获取同一把锁，比如一个线程在执行一个带锁的方法，该方法中又调用了另一个需要相同锁的方法，则该线程可以直接执行调用的方法，而无需重新获得锁。 5、阻塞锁 / 非阻塞锁，阻塞锁即没有获取到锁，则继续等待获取锁；非阻塞锁即没有获取到锁后，不继续等待，直接返回锁失败。 二、实现方式 2.1、数据库锁 一般很少使用数据库锁，性能不好并且容易产生死锁。 2.1.1、基于 MySQL 锁表 该实现方式完全依靠数据库唯一索引来实现，当想要获得锁时，即向数据库中插入一条记录，释放锁时就删除这条记录。这种方式存在以下几个问题： 1、锁没有失效时间，解锁失败会导致死锁，其他线程无法再获取到锁，因为唯一索引 insert 都会返回失败。 2、只能是非阻塞锁，insert 失败直接就报错了，无法进入队列进行重试。 3、不可重入，同一线程在没有释放锁之前无法再获取到锁。 2.1.2、 采用乐观锁增加版本号 根据版本号来判断更新之前有没有其他线程更新过，如果被更新过，则获取锁失败。 2.2、缓存锁 2.2.1、基于 setnx、expire 两个命令来实现 基于 setnx（set if not exist）的特点，当缓存里 key 不存在时，才会去 set，否则直接返回 false。如果返回 true 则获取到锁，否则获取锁失败，为了防止死锁，我们再用 expire 命令对这个 key 设置一个超时时间来避免。但是这里看似完美，实则有缺陷，当我们 setnx 成功后，线程发生异常中断，expire 还没来的及设置，那么就会产生死锁。 解决上述问题有两种方案： 第一种是采用 redis2.6.12 版本以后的 set，它提供了一系列选项： EX seconds – 设置键 key 的过期时间，单位时秒 PX milliseconds – 设置键 key 的过期时间，单位时毫秒 NX – 只有键 key 不存在的时候才会设置 key 的值 XX – 只有键 key 存在的时候才会设置 key 的值 第二种采用 setnx ()，get ()，getset () 实现，大体的实现过程如下： 1、线程 Asetnx，值为超时的时间戳 (t1)，如果返回 true，获得锁。 2、线程 B 用 get 命令获取 t1，与当前时间戳比较，判断是否超时，没超时 false，如果已超时执行步骤 3。 3、计算新的超时时间 t2，使用 getset 命令返回 t3 (这个值可能其他线程已经修改过)，如果 t1==t3, 获得锁，如果 t1!=t3 说明锁被其他线程获取了。 4、获取锁后，处理完业务逻辑，再去判断锁是否超时，如果没超时删除锁，如果已超时，不用处理（防止删除其他线程的锁）。 2、RedLock 算法 redlock 算法是 redis 作者推荐的一种分布式锁实现方式，算法的内容如下： 1、获取当前时间； 2、尝试从 5 个相互独立 redis 客户端获取锁； 3、计算获取所有锁消耗的时间，当且仅当客户端从多数节点获取锁，并且获取锁的时间小于锁的有效时间，认为获得锁； 4、重新计算有效期时间，原有效时间减去获取锁消耗的时间； 5、删除所有实例的锁。 redlock 算法相对于单节点 redis 锁可靠性要更高，但是实现起来条件也较为苛刻。 1、必须部署 5 个节点才能让 Redlock 的可靠性更强。 2、需要请求 5 个节点才能获取到锁，通过 Future 的方式，先并发向 5 个节点请求，再一起获得响应结果，能缩短响应时间，不过还是比单节点 redis 锁要耗费更多时间。 然后由于必须获取到 5 个节点中的 3 个以上，所以可能出现获取锁冲突，即大家都获得了 1-2 把锁，结果谁也不能获取到锁，这个问题，redis 作者借鉴了 raft 算法的精髓，通过冲突后在随机时间开始，可以大大降低冲突时间，但是这问题并不能很好的避免，特别是在第一次获取锁的时候，所以获取锁的时间成本增加了。 如果 5 个节点有 2 个宕机，此时锁的可用性会极大降低，首先必须等待这两个宕机节点的结果超时才能返回，另外只有 3 个节点，客户端必须获取到这全部 3 个节点的锁才能拥有锁，难度也加大了。 如果出现网络分区，那么可能出现客户端永远也无法获取锁的情况，介于这种情况，下面我们来看一种更可靠的分布式锁 zookeeper 锁。 2.3、zookeeper 分布式锁 zookeeper 是一个为分布式应用提供一致性服务的软件，它内部是一个分层的文件系统目录树结构，规定统一个目录下只能有一个唯一文件名。 数据模型： 永久节点：节点创建后，不会因为会话失效而消失。 临时节点：与永久节点相反，如果客户端连接失效，则立即删除节点。 顺序节点：与上述两个节点特性类似，如果指定创建这类节点时，zk 会自动在节点名后加一个数字后缀，并且是有序的。 监视器（watcher）： 当创建一个节点时，可以注册一个该节点的监视器，当节点状态发生改变时，watch 被触发时，ZooKeeper 将会向客户端发送且仅发送一条通知，因为 watch 只能被触发一次。 根据 zookeeper 的这些特性，我们来看看如何利用这些特性来实现分布式锁： 1、创建一个锁目录 lock。 2、希望获得锁的线程 A 就在 lock 目录下，创建临时顺序节点。 3、获取锁目录下所有的子节点，然后获取比自己小的兄弟节点，如果不存在，则说明当前线程顺序号最小，获得锁。 4、线程 B 获取所有节点，判断自己不是最小节点，设置监听 (watcher) 比自己次小的节点（只关注比自己次小的节点是为了防止发生 “羊群效应”）。 5、线程 A 处理完，删除自己的节点，线程 B 监听到变更事件，判断自己是最小的节点，获得锁。 三、小结 在分布式系统中，共享资源互斥访问问题非常普遍，而针对访问共享资源的互斥问题，常用的解决方案就是使用分布式锁，这里只介绍了几种常用的分布式锁，分布式锁的实现方式还有有很多种，根据业务选择合适的分布式锁，下面对上述几种锁进行一下比较： 数据库锁： 优点：直接使用数据库，使用简单。 缺点：分布式系统大多数瓶颈都在数据库，使用数据库锁会增加数据库负担。 缓存锁： 优点：性能高，实现起来较为方便，在允许偶发的锁失效情况，不影响系统正常使用，建议采用缓存锁。 缺点：通过锁超时机制不是十分可靠，当线程获得锁后，处理时间过长导致锁超时，就失效了锁的作用。 zookeeper 锁： 优点：不依靠超时时间释放锁；可靠性高；系统要求高可靠性时，建议采用 zookeeper 锁。 缺点：性能比不上缓存锁，因为要频繁的创建节点删除节点。
分布式事务的解决方案有如下几种： 全局事务 基于可靠消息服务的分布式事务 TCC 最大努力通知 一、方案 1：全局事务（DTP 模型） 全局事务基于 DTP 模型实现。DTP 是由 X/Open 组织提出的一种分布式事务模型 ——X/Open Distributed Transaction Processing Reference Model。它规定了要实现分布式事务，需要三种角色： AP：Application 应用系统它就是我们开发的业务系统，在我们开发的过程中，可以使用资源管理器提供的事务接口来实现分布式事务。 TM：Transaction Manager 事务管理器 分布式事务的实现由事务管理器来完成，它会提供分布式事务的操作接口供我们的业务系统调用。这些接口称为 TX 接口。 事务管理器还管理着所有的资源管理器，通过它们提供的 XA 接口来同一调度这些资源管理器，以实现分布式事务。 DTP 只是一套实现分布式事务的规范，并没有定义具体如何实现分布式事务，TM 可以采用 2PC、3PC、Paxos 等协议实现分布式事务。 RM：Resource Manager 资源管理器 能够提供数据服务的对象都可以是资源管理器，比如：数据库、消息中间件、缓存等。大部分场景下，数据库即为分布式事务中的资源管理器。 资源管理器能够提供单数据库的事务能力，它们通过 XA 接口，将本数据库的提交、回滚等能力提供给事务管理器调用，以帮助事务管理器实现分布式的事务管理。 XA 是 DTP 模型定义的接口，用于向事务管理器提供该资源管理器 (该数据库) 的提交、回滚等能力。 DTP 只是一套实现分布式事务的规范，RM 具体的实现是由数据库厂商来完成的。 1.1、实际方案：基于 XA 协议的两阶段提交 XA 是一个分布式事务协议，由 Tuxedo 提出。XA 中大致分为两部分：事务管理器和本地资源管理器。其中本地资源管理器往往由数据库实现，比如 Oracle、DB2 这些商业数据库都实现了 XA 接口，而事务管理器作为全局的调度者，负责各个本地资源的提交和回滚。XA 实现分布式事务的原理如下： 总的来说，XA 协议比较简单，而且一旦商业数据库实现了 XA 协议，使用分布式事务的成本也比较低。但是，XA 也有致命的缺点，那就是性能不理想，特别是在交易下单链路，往往并发量很高，XA 无法满足高并发场景。XA 目前在商业数据库支持的比较理想，在 mysql 数据库中支持的不太理想，mysql 的 XA 实现，没有记录 prepare 阶段日志，主备切换回导致主库与备库数据不一致。许多 nosql 也没有支持 XA，这让 XA 的应用场景变得非常狭隘。 二、方案 2：基于可靠消息服务的分布式事务（事务消息中间件） 这种实现分布式事务的方式需要通过消息中间件来实现。假设有 A 和 B 两个系统，分别可以处理任务 A 和任务 B。此时系统 A 中存在一个业务流程，需要将任务 A 和任务 B 在同一个事务中处理。下面来介绍基于消息中间件来实现这种分布式事务。 在系统 A 处理任务 A 前，首先向消息中间件发送一条消息。 消息中间件收到后将该条消息持久化，但并不投递。此时下游系统 B 仍然不知道该条消息的存在。 消息中间件持久化成功后，便向系统 A 返回一个确认应答。 系统 A 收到确认应答后，则可以开始处理任务 A。 任务 A 处理完成后，向消息中间件发送 Commit 请求。该请求发送完成后，对系统 A 而言，该事务的处理过程就结束了，此时它可以处理别的任务了。但 commit 消息可能会在传输途中丢失，从而消息中间件并不会向系统 B 投递这条消息，从而系统就会出现不一致性。这个问题由消息中间件的事务回查机制完成。 消息中间件收到 Commit 指令后，便向系统 B 投递该消息，从而触发任务 B 的执行。 当任务 B 执行完成后，系统 B 向消息中间件返回一个确认应答，告诉消息中间件该消息已经成功消费，此时，这个分布式事务完成。 上述过程可以得出如下几个结论： 消息中间件扮演者分布式事务协调者的角色。 系统 A 完成任务 A 后，到任务 B 执行完成之间，会存在一定的时间差。在这个时间差内，整个系统处于数据不一致的状态，但这短暂的不一致性是可以接受的，因为经过短暂的时间后，系统又可以保持数据一致性，满足 BASE 理论。 上述过程中，如果任务 A 处理失败，那么需要进入回滚流程，如下图所示： 若系统 A 在处理任务 A 时失败，那么就会向消息中间件发送 Rollback 请求。和发送 Commit 请求一样，系统 A 发完之后便可以认为回滚已经完成，它便可以去做其他的事情。 消息中间件收到回滚请求后，直接将该消息丢弃，而不投递给系统 B，从而不会触发系统 B 的任务 B。 此时系统又处于一致性状态，因为任务 A 和任务 B 都没有执行。 上面所介绍的 Commit 和 Rollback 都属于理想情况，但在实际系统中，Commit 和 Rollback 指令都有可能在传输途中丢失。那么当出现这种情况的时候，消息中间件是如何保证数据一致性呢？—— 答案就是超时询问机制。 系统 A 除了实现正常的业务流程外，还需提供一个事务询问的接口，供消息中间件调用。当消息中间件收到一条事务型消息后便开始计时，如果到了超时时间也没收到系统 A 发来的 Commit 或 Rollback 指令的话，就会主动调用系统 A 提供的事务询问接口询问该系统目前的状态。该接口会返回三种结果： 提交若获得的状态是 “提交”，则将该消息投递给系统 B。 回滚若获得的状态是 “回滚”，则直接将条消息丢弃。 处理中若获得的状态是 “处理中”，则继续等待。 消息中间件的超时询问机制能够防止上游系统因在传输过程中丢失 Commit/Rollback 指令而导致的系统不一致情况，而且能降低上游系统的阻塞时间，上游系统只要发出 Commit/Rollback 指令后便可以处理其他任务，无需等待确认应答。而 Commit/Rollback 指令丢失的情况通过超时询问机制来弥补，这样大大降低上游系统的阻塞时间，提升系统的并发度。 下面来说一说消息投递过程的可靠性保证。当上游系统执行完任务并向消息中间件提交了 Commit 指令后，便可以处理其他任务了，此时它可以认为事务已经完成，接下来消息中间件一定会保证消息被下游系统成功消费掉！那么这是怎么做到的呢？这由消息中间件的投递流程来保证。 消息中间件向下游系统投递完消息后便进入阻塞等待状态，下游系统便立即进行任务的处理，任务处理完成后便向消息中间件返回应答。消息中间件收到确认应答后便认为该事务处理完毕！ 如果消息在投递过程中丢失，或消息的确认应答在返回途中丢失，那么消息中间件在等待确认应答超时之后就会重新投递，直到下游消费者返回消费成功响应为止。当然，一般消息中间件可以设置消息重试的次数和时间间隔，比如：当第一次投递失败后，每隔五分钟重试一次，一共重试 3 次。如果重试 3 次之后仍然投递失败，那么这条消息就需要人工干预。 问题：消息投递失败后为什么不回滚消息，而是不断尝试重新投递？ 这就涉及到整套分布式事务系统的实现成本问题。我们知道，当系统 A 将向消息中间件发送 Commit 指令后，它便去做别的事情了。如果此时消息投递失败，需要回滚的话，就需要让系统 A 事先提供回滚接口，这无疑增加了额外的开发成本，业务系统的复杂度也将提高。对于一个业务系统的设计目标是，在保证性能的前提下，最大限度地降低系统复杂度，从而能够降低系统的运维成本。 不知大家是否发现，上游系统 A 向消息中间件提交 Commit/Rollback 消息采用的是异步方式，也就是当上游系统提交完消息后便可以去做别的事情，接下来提交、回滚就完全交给消息中间件来完成，并且完全信任消息中间件，认为它一定能正确地完成事务的提交或回滚。然而，消息中间件向下游系统投递消息的过程是同步的。也就是消息中间件将消息投递给下游系统后，它会阻塞等待，等下游系统成功处理完任务返回确认应答后才取消阻塞等待。为什么这两者在设计上是不一致的呢？ 首先，上游系统和消息中间件之间采用异步通信是为了提高系统并发度。业务系统直接和用户打交道，用户体验尤为重要，因此这种异步通信方式能够极大程度地降低用户等待时间。此外，异步通信相对于同步通信而言，没有了长时间的阻塞等待，因此系统的并发性也大大增加。但异步通信可能会引起 Commit/Rollback 指令丢失的问题，这就由消息中间件的超时询问机制来弥补。 那么，消息中间件和下游系统之间为什么要采用同步通信呢？ 异步能提升系统性能，但随之会增加系统复杂度；而同步虽然降低系统并发度，但实现成本较低。因此，在对并发度要求不是很高的情况下，或者服务器资源较为充裕的情况下，我们可以选择同步来降低系统的复杂度。我们知道，消息中间件是一个独立于业务系统的第三方中间件，它不和任何业务系统产生直接的耦合，它也不和用户产生直接的关联，它一般部署在独立的服务器集群上，具有良好的可扩展性，所以不必太过于担心它的性能，如果处理速度无法满足我们的要求，可以增加机器来解决。而且，即使消息中间件处理速度有一定的延迟那也是可以接受的，因为前面所介绍的 BASE 理论就告诉我们了，我们追求的是最终一致性，而非实时一致性，因此消息中间件产生的时延导致事务短暂的不一致是可以接受的。 三、方案 3：最大努力通知（定期校对）也叫本地消息表 最大努力通知也被称为定期校对，其实在方案二中已经包含，这里再单独介绍，主要是为了知识体系的完整性。这种方案也需要消息中间件的参与，其过程如下： 上游系统在完成任务后，向消息中间件同步地发送一条消息，确保消息中间件成功持久化这条消息，然后上游系统可以去做别的事情了； 消息中间件收到消息后负责将该消息同步投递给相应的下游系统，并触发下游系统的任务执行； 当下游系统处理成功后，向消息中间件反馈确认应答，消息中间件便可以将该条消息删除，从而该事务完成。 上面是一个理想化的过程，但在实际场景中，往往会出现如下几种意外情况： 消息中间件向下游系统投递消息失败 上游系统向消息中间件发送消息失败 对于第一种情况，消息中间件具有重试机制，我们可以在消息中间件中设置消息的重试次数和重试时间间隔，对于网络不稳定导致的消息投递失败的情况，往往重试几次后消息便可以成功投递，如果超过了重试的上限仍然投递失败，那么消息中间件不再投递该消息，而是记录在失败消息表中，消息中间件需要提供失败消息的查询接口，下游系统会定期查询失败消息，并将其消费，这就是所谓的 “定期校对”。 如果重复投递和定期校对都不能解决问题，往往是因为下游系统出现了严重的错误，此时就需要人工干预。 对于第二种情况，需要在上游系统中建立消息重发机制。可以在上游系统建立一张本地消息表，并将 任务处理过程 和 向本地消息表中插入消息 这两个步骤放在一个本地事务中完成。如果向本地消息表插入消息失败，那么就会触发回滚，之前的任务处理结果就会被取消。如果这量步都执行成功，那么该本地事务就完成了。接下来会有一个专门的消息发送者不断地发送本地消息表中的消息，如果发送失败它会返回重试。当然，也要给消息发送者设置重试的上限，一般而言，达到重试上限仍然发送失败，那就意味着消息中间件出现严重的问题，此时也只有人工干预才能解决问题。 对于不支持事务型消息的消息中间件，如果要实现分布式事务的话，就可以采用这种方式。它能够通过重试机制 + 定期校对实现分布式事务，但相比于第二种方案，它达到数据一致性的周期较长，而且还需要在上游系统中实现消息重试发布机制，以确保消息成功发布给消息中间件，这无疑增加了业务系统的开发成本，使得业务系统不够纯粹，并且这些额外的业务逻辑无疑会占用业务系统的硬件资源，从而影响性能。 因此，尽量选择支持事务型消息的消息中间件来实现分布式事务，如 RocketMQ。 四、方案 4：TCC（两阶段型、补偿型） 跨应用的业务操作原子性要求，其实是比较常见的。比如在第三方支付场景中的组合支付，用户在电商网站购物后，要同时使用余额和 红包支付该笔订单，而余额系统和红包系统分别是不同的应用系统，支付系统在调用这两个系统进行支付时，就需要保证余额扣减和红 包使用要么同时成功，要么同时失败。 TCC 事务的出现正是为了解决应用拆分带来的跨应用业务操作原子性的问题。当然，由于常规的 XA 事务 (2PC，2 Phase Commit, 两阶段提交) 性能上不尽如人意，也有通过 TCC 事务来解决数据库拆分的使用场景 (如账务拆分)，这个本文后续部分再详述。 故从整个系统架构的角度来看，分布式事务的不同方案是存在层次结构的。 4.1、TCC 的机制 明眼一看就知道，TCC 应该是三个英文单词的首字母缩写而来。没错，TCC 分别对应 Try、Confirm 和 Cancel 三种操作， 这三种操作的业务含义如下： Try：预留业务资源 Confirm：确认执行业务操作 Cancel：取消执行业务操作 稍稍对照下关系型数据库事务的三种操作：DML、Commit 和 Rollback，会发现和 TCC 有异曲同工之妙。在一个跨应用的业务操作中， Try 操作是先把多个应用中的业务资源预留和锁定住，为后续的确认打下基础，类似的，DML 操作要锁定数据库记录行，持有数据库资源； Confirm 操作是在 Try 操作中涉及的所有应用均成功之后进行确认，使用预留的业务资源，和 Commit 类似； 而 Cancel 则是当 Try 操作中涉及的所有应用没有全部成功，需要将已成功的应用进行取消 (即 Rollback 回滚)。 其中 Confirm 和 Cancel 操作是一对反向业务操作。 简而言之，TCC 是应用层的 2PC (2 Phase Commit, 两阶段提交)，如果你将应用看做资源管理器的话。 详细来说，TCC 每项操作需要做的事情如下： Try：尝试执行业务。 完成所有业务检查 (一致性) 预留必须业务资源 (准隔离性)。 Confirm：确认执行业务。 真正执行业务。 不做任何业务检查。 只使用 Try 阶段预留的业务资源。 Cancel：取消执行业务。 释放 Try 阶段预留的业务资源。 一个完整的 TCC 事务参与方包括三部分： 主业务服务：主业务服务为整个业务活动的发起方，如前面提到的组合支付场景，支付系统即是主业务服务。 从业务服务：从业务服务负责提供 TCC 业务操作，是整个业务活动的操作方。从业务服务必须实现 Try、Confirm 和 Cancel 三个接口，供主业务服务调用。 由于 Confirm 和 Cancel 操作可能被重复调用，故要求 Confirm 和 Cancel 两个接口必须是幂等的。前面的组合支付场景中的余额系统和红包系统即为从业务服务。 业务活动管理器：业务活动管理器管理控制整个业务活动，包括记录维护 TCC 全局事务的事务状态和每个从业务服务的子事务状态，并在业务活动提交时确认所有的 TCC 型操作的 confirm 操作，在业务活动取消时调用所有 TCC 型操作的 cancel 操作。 可见整个 TCC 事务对于主业务服务来说是透明的，其中业务活动管理器和从业务服务各自干了一部分工作。 4.2、TCC 的优点和限制 TCC 事务的优点如下： 解决了跨应用业务操作的原子性问题，在诸如组合支付、账务拆分场景非常实用。 TCC 实际上把数据库层的二阶段提交上提到了应用层来实现，对于数据库来说是一阶段提交，规避了数据库层的 2PC 性能低下问题。 TCC 事务的缺点，主要就一个： TCC 的 Try、Confirm 和 Cancel 操作功能需业务提供，开发成本高。 当然，对 TCC 事务的这个缺点是否是缺点，是一个见仁见智的事情。 4.3、一个案例理解 TCC 说实话，TCC 的理论有点让人费解。故接下来将以账务拆分为例，对 TCC 事务的流程做一个描述，希望对理解 TCC 有所帮助。 账务拆分的业务场景如下，分别位于三个不同分库的帐户 A、B、C，A 和 B 一起向 C 转帐共 80 元：分布式事务之说说 TCC 事务 4.3.1、Try：尝试执行业务。 完成所有业务检查 (一致性)：检查 A、B、C 的帐户状态是否正常，帐户 A 的余额是否不少于 30 元，帐户 B 的余额是否不少于 50 元。 预留必须业务资源 (准隔离性)：帐户 A 的冻结金额增加 30 元，帐户 B 的冻结金额增加 50 元，这样就保证不会出现其他并发进程扣减 了这两个帐户的余额而导致在后续的真正转帐操作过程中，帐户 A 和 B 的可用余额不够的情况。 4.3.2、Confirm：确认执行业务。 真正执行业务：如果 Try 阶段帐户 A、B、C 状态正常，且帐户 A、B 余额够用，则执行帐户 A 给账户 C 转账 30 元、帐户 B 给账户 C 转账 50 元的 转帐操作。 不做任何业务检查：这时已经不需要做业务检查，Try 阶段已经完成了业务检查。 只使用 Try 阶段预留的业务资源：只需要使用 Try 阶段帐户 A 和帐户 B 冻结的金额即可。 4.3.3、Cancel：取消执行业务 释放 Try 阶段预留的业务资源：如果 Try 阶段部分成功，比如帐户 A 的余额够用，且冻结相应金额成功，帐户 B 的余额不够而冻结失败，则需要对帐户 A 做 Cancel 操作，将帐户 A 被冻结的金额解冻掉。
一、分布式 将不同的业务分部在不同的地方，就构成了一个分布式的系统，系统 A 是整个分布式系统的入口，用户访问量大的时候要么是速度巨慢，可能会引起单点失败。 二、集群（Cluster） 将三个服务器（系统 A）的系统就组成了一个集群。 三、负载均衡（Load Balancer） 负载均衡的工作独立出来，放到独立的服务器上（例如 nginx）： 负载均衡的服务器虽然工作内容简单，就是拿到请求，分发请求，但是 Nginx 还是有可能挂掉，单点失败还是会出现。 可以将负载均衡也组成一个集群，但是和系统 A 的集群有两点不同： 1、这个新的集群中虽然有两个机器，但是可以用某种办法，让这个机器对外只提供一个 IP 地址，也就是用户看到的好像只有一个机器。 2、同一时刻，只让一个负载均衡的机器工作，另外一个原地待命，如果工作的那个挂掉了，待命的那个就顶上去。 四、分布式架构 4.1.集中式系统 所谓的集中式系统就是指由一台或多台主计算机组成中心节点，数据集中存储于这个中心节点中，并且整个系统的所有业务单元都集中部署在这个中心节点上，系统的所有功能均由其集中处理。 集中式系统的最大的特点就是部署结构非常简单。因此无需考虑如何对服务进行多节点的部署，也就不用考虑各节点之间的分布式协作问题。但是，由于采用单机部署，很可能带来系统大而复杂、难于维护、发生单点故障（单个点发生故障的时候会波及到整个系统或者网络，从而导致整个系统或者网络的瘫痪）、扩展性差等问题。 4.2、分布式系统 分布式系统是一个硬件或软件组件分布在不同的网络计算机上，彼此之间仅仅通过消息传递进行通信和协调的系统。简单来说就是一群独立计算机集合共同对外提供服务，但是对于系统的用户来说，就像是一台计算机在提供服务一样。分布式意味着可以采用更多的普通计算机（相对于昂贵的大型机）组成分布式集群对外提供服务。计算机越多，CPU、内存、存储资源等也就越多，能够处理的并发访问量也就越大。 各个主机之间通信和协调主要通过网络进行，所以分布式系统中的计算机在空间上几乎没有任何限制，这些计算机可能被放在不同的机柜上，也可能被部署在不同的机房中。但是，无论空间上如何分布，一个标准的分布式系统应该具有以下几个主要特征： 分布性 分布式系统中的多台计算机之间在空间位置上可以随意分布，同时，机器的分布情况也会随时变动。 对等性 分布式系统中的计算机没有主／从之分，即没有控制整个系统的主机，也没有被控制的从机，组成分布式系统的所有计算机节点都是对等的。副本（Replica）是分布式系统最常见的概念之一，指的是分布式系统对数据和服务提供的一种冗余方式。在常见的分布式系统中，为了对外提供高可用的服务，我们往往会对数据和服务进行副本处理。数据副本是指在不同节点上持久化同一份数据，当某一个节点上存储的数据丢失时，可以从副本上读取该数据，这是解决分布式系统数据丢失问题最为有效的手段。另一类副本是服务副本，指多个节点提供同样的服务，每个节点都有能力接收来自外部的请求并进行相应的处理。 并发性 在一个计算机网络中，程序运行过程的并发性操作是非常常见的行为。例如同一个分布式系统中的多个节点，可能会并发地操作一些共享的资源，如何准确并高效地协调分布式并发操作也成为了分布式系统架构与设计中最大的挑战之一。 缺乏全局时钟 在分布式系统中，很难定义两个事件究竟谁先谁后，原因就是因为分布式系统缺乏一个全局的时钟序列控制。 故障总是会发生 组成分布式系统的所有计算机，都有可能发生任何形式的故障。除非需求指标允许，在系统设计时不能放过任何异常情况。 五、分布式系统面临的问题 5.1、通信异常 分布式系统需要在各个节点之间进行网络通信，因此网络通信都会伴随着网络不可用的风险或是系统不可用都会导致最终分布式系统无法顺利完成一次网络通信。另外，即使分布式系统各节点之间的网络通信能够正常进行，其延时也会远大于单机操作，会影响消息的收发的过程，因此消息丢失和消息延迟变得非常普遍。 5.2、网络分区 当网络由于发生异常情况，导致分布式系统中部分节点之间的网络延时不断增大，最终导致组成分布式系统的所有节点中，只有部分节点之间能够进行正常通信，而另一些节点则不能，这个现象为网络分区，就是俗称的 “脑裂”。当网络分区出现时，分布式系统会出现局部小集群，在极端情况下，这些局部小集群会独立完成原本需要整个分布式才能完成的功能，这就对分布式一致性提出类非常大的挑战。 5.3、三态 分布式系统的每一次请求与响应，存在特有的三态概念，即成功、失败与超时。当出现超时现象时，网络通信的发起方是无法确定当前请求是否被成功处理的。 5.4、节点故障 节点故障则是分布式环境下另一个比较常见的问题，指的是组成分布式系统的服务器节点出现的宕机或僵死现象。 六、分布式理论 (一) - CAP 定理 CAP 原则又称 CAP 定理，指的是在一个分布式系统中，Consistency（一致性）、 Availability（可用性）、Partition tolerance（分区容错性）这三个基本需求，最多只能同时满足其中的 2 个。 6.1、初探分布式理论 6.1.1、CAP 原则简介 Consistency（一致性） 指数据在多个副本之间能够保持一致的特性（严格的一致性）。 Availability（可用性） 指系统提供的服务必须一直处于可用的状态，每次请求都能获取到非错的响应（不保证获取的数据为最新数据）。 Partition tolerance（分区容错性） 分布式系统在遇到任何网络分区故障的时候，仍然能够对外提供满足一致性和可用性的服务，除非整个网络环境都发生了故障。 在分布式系统中，不同的节点分布在不同的子网络中，由于一些特殊的原因，这些子节点之间出现了网络不通的状态，但他们的内部子网络是正常的。从而导致了整个系统的环境被切分成了若干个孤立的区域，这就是分区。 6.1.2、CAP 原则论证 如图所示，是我们证明 CAP 的基本场景，网络中有两个节点 N1 和 N2，可以简单的理解 N1 和 N2 分别是两台计算机，他们之间网络可以连通，N1 中有一个应用程序 A，和一个数据库 V，N2 也有一个应用程序 B 和一个数据库 V。现在，A 和 B 是分布式系统的两个部分，V 是分布式系统的数据存储的两个子数据库。 在满足一致性的时候，N1 和 N2 中的数据是一样的，V0=V0。 在满足可用性的时候，用户不管是请求 N1 或者 N2，都会得到立即响应。 在满足分区容错性的情况下，N1 和 N2 有任何一方宕机，或者网络不通的时候，都不会影响 N1 和 N2 彼此之间的正常运作。 如图所示，这是分布式系统正常运转的流程，用户向 N1 机器请求数据更新，程序 A 更新数据库 V0 为 V1。分布式系统将数据进行同步操作 M，将 V1 同步的 N2 中 V0，使得 N2 中的数据 V0 也更新为 V1，N2 中的数据再响应 N2 的请求。 根据 CAP 原则定义，系统的一致性、可用性和分区容错性细分如下： 一致性：N1 和 N2 的数据库 V 之间的数据是否完全一样。 可用性：N1 和 N2 的对外部的请求能否做出正常的响应。 分区容错性：N1 和 N2 之间的网络是否互通。 这是正常运作的场景，也是理想的场景。作为一个分布式系统，它和单机系统的最大区别，就在于网络。现在假设一种极端情况，N1 和 N2 之间的网络断开了，我们要支持这种网络异常。相当于要满足分区容错性，能不能同时满足一致性和可用性呢？还是说要对他们进行取舍？ 假设在 N1 和 N2 之间网络断开的时候，有用户向 N1 发送数据更新请求，那 N1 中的数据 V0 将被更新为 V1。由于网络是断开的，所以分布式系统同步操作 M，所以 N2 中的数据依旧是 V0。这个时候，有用户向 N2 发送数据读取请求，由于数据还没有进行同步，应用程序没办法立即给用户返回最新的数据 V1，怎么办呢？ 这里有两种选择： 第一：牺牲数据一致性，保证可用性。响应旧的数据 V0 给用户。 第二：牺牲可用性，保证数据一致性。阻塞等待，直到网络连接恢复，数据更新操作 M 完成之后，再给用户响应最新的数据 V1。 这个过程，证明了要满足分区容错性的分布式系统，只能在一致性和可用性两者中，选择其中一个。 6.1.3、CAP 原则权衡 通过 CAP 理论，我们知道无法同时满足一致性、可用性和分区容错性这三个特性，那要舍弃哪个呢？ CA without P 如果不要求 P（不允许分区），则 C（强一致性）和 A（可用性）是可以保证的。但其实分区不是你想不想的问题，而是始终会存在，因此 CA 的系统更多的是允许分区后各子系统依然保持 CA。 CP without A 如果不要求 A（可用），相当于每个请求都需要在 Server 之间强一致，而 P（分区）会导致同步时间无限延长，如此 CP 也是可以保证的。很多传统的数据库分布式事务都属于这种模式。 AP wihtout C 要高可用并允许分区，则需放弃一致性。一旦分区发生，节点之间可能会失去联系，为了高可用，每个节点只能用本地数据提供服务，而这样会导致全局数据的不一致性。现在众多的 NoSQL 都属于此类。 6.2、小结 对于多数大型互联网应用的场景，主机众多、部署分散。而且现在的集群规模越来越大，所以节点故障、网络故障是常态。这种应用一般要保证服务可用性达到 N 个 9，即保证 P 和 A，只有舍弃 C（退而求其次保证最终一致性）。虽然某些地方会影响客户体验，但没达到造成用户流程的严重程度。 对于涉及到钱财这样不能有一丝让步的场景，C 必须保证。网络发生故障宁可停止服务，这是保证 CA，舍弃 P。还有一种是保证 CP，舍弃 A，例如网络故障时只读不写。 孰优孰劣，没有定论，只能根据场景定夺，适合的才是最好的。 七、分布式理论 (二) - BASE 理论 BASE 理论是由 eBay 架构师提出的。BASE 是对 CAP 中一致性和可用性权衡的结果，其来源于对大规模互联网分布式系统实践的总结，是基于 CAP 定律逐步演化而来。其核心思想是即使无法做到强一致性，但每个应用都可以根据自身业务特点，才用适当的方式来使系统打到最终一致性。 7.1、CAP 的 3 选 2 伪命题 实际上，不是为了 P（分区容错性），必须在 C（一致性）和 A（可用性）之间任选其一。分区的情况很少出现，CAP 在大多时间能够同时满足 C 和 A。 对于分区存在或者探知其影响的情况下，需要提供一种预备策略做出处理： 探知分区的发生； 进入显示的分区模式，限制某些操作； 启动恢复过程，恢复数据一致性，补偿分区发生期间的错误。 7.2、BASE 理论简介 BASE 理论是 Basically Available (基本可用)，Soft State（软状态）和 Eventually Consistent（最终一致性）三个短语的缩写。 其核心思想是： 既是无法做到强一致性（Strong consistency），但每个应用都可以根据自身的业务特点，采用适当的方式来使系统达到最终一致性（Eventual consistency）。 7.3、BASE 理论的内容 7.3.1、基本可用 什么是基本可用呢？假设系统，出现了不可预知的故障，但还是能用，相比较正常的系统而言： 响应时间上的损失：正常情况下的搜索引擎 0.5 秒即返回给用户结果，而基本可用的搜索引擎可以在 2 秒作用返回结果。 功能上的损失：在一个电商网站上，正常情况下，用户可以顺利完成每一笔订单。但是到了大促期间，为了保护购物系统的稳定性，部分消费者可能会被引导到一个降级页面。 7.3.2、软状态 什么是软状态呢？相对于原子性而言，要求多个节点的数据副本都是一致的，这是一种 “硬状态”。 软状态指的是：允许系统中的数据存在中间状态，并认为该状态不影响系统的整体可用性，即允许系统在多个不同节点的数据副本存在数据延时。 7.3.3、最终一致性 上面说软状态，然后不可能一直是软状态，必须有个时间期限。在期限过后，应当保证所有副本保持数据一致性，从而达到数据的最终一致性。这个时间期限取决于网络延时、系统负载、数据复制方案设计等等因素。 而在实际工程实践中，最终一致性分为 5 种： 因果一致性（Causal consistency） 因果一致性指的是：如果节点 A 在更新完某个数据后通知了节点 B，那么节点 B 之后对该数据的访问和修改都是基于 A 更新后的值。于此同时，和节点 A 无因果关系的节点 C 的数据访问则没有这样的限制。 读己之所写（Read your writes） 读己之所写指的是：节点 A 更新一个数据后，它自身总是能访问到自身更新过的最新值，而不会看到旧值。其实也算一种因果一致性。 会话一致性（Session consistency） 会话一致性将对系统数据的访问过程框定在了一个会话当中：系统能保证在同一个有效的会话中实现 “读己之所写” 的一致性，也就是说，执行更新操作之后，客户端能够在同一个会话中始终读取到该数据项的最新值。 单调读一致性（Monotonic read consistency） 单调读一致性指的是：如果一个节点从系统中读取出一个数据项的某个值后，那么系统对于该节点后续的任何数据访问都不应该返回更旧的值。 单调写一致性（Monotonic write consistency） 单调写一致性指的是：一个系统要能够保证来自同一个节点的写操作被顺序的执行。 在实际的实践中，这 5 种系统往往会结合使用，以构建一个具有最终一致性的分布式系统。 实际上，不只是分布式系统使用最终一致性，关系型数据库在某个功能上，也是使用最终一致性的。比如备份，数据库的复制过程是需要时间的，这个复制过程中，业务读取到的值就是旧的。当然，最终还是达成了数据一致性。这也算是一个最终一致性的经典案例。