标签：超时 Java 32 分布式计算线程 master 服务器分布式客户端

1 这才是真正的分布式锁

技术领域，我觉得了解来龙去脉，了解本质原理，比用什么工具实现更重要：

（1）进程多线程如何互斥？

（2）一个手机上两个APP访问一个文件如何互斥？

（3）分布式环境下多个服务访问一个资源如何互斥？

归根结底，是利用一个互斥才能访问的公共资源来实现分布式锁，具体这个公共资源是redis来setnx，还是zookeeper，相反没有这么重要。

言归正传，今天把昨天文章的缘起讲一讲，并通过Google Chubby的论文阅读笔记聊一聊分布式锁。

1.1 需求缘起

58到家APP新上线了导入通讯录好友功能，测试的同学发现，连续点击导入会导入重复数据：

客户端同一个用户同时发出了多个请求，分布式环境下，多台机器上部署的多个service进行了并发操作，故插入了冗余数据。

解决思路：同一个用户同时只能有一个导入请求，需要做互斥，最简易的方案，使用setnx快速解决。

（1）同一个用户，多个service进行并发操作，service需要先去抢锁

（2）抢到锁的service，才去数据库操作

具体这个锁用setnx，还是zookeeper都不太重要，利用一个互斥方能够访问的公共资源来实现分布式锁，这才是《一分钟实现分布式锁》的重点。

1. 2 Google Chubby分布式锁阅读笔记

上一篇文章的评论中，有些朋友提到了zookeeper，会使用不够，借着Google Chubby了解下分布式锁的实现也是有必要的。

早年Google的四大基础设施，分别是GFS、MapReduce、BigTable、Chubby，其中Chubby用于提供分布式的锁服务。

1.2.1 简介

Chubby系统提供粗粒度的分布式锁服务，Chubby的使用者不需要关注复杂的同步协议，而是通过已经封装好的客户端直接调用Chubby的锁服务，就可以保证数据操作的一致性。

Chubby具有广泛的应用场景，例如：

（1）GFS选主服务器；

（2）BigTable中的表锁；

1.2.2 背景

Chubby本质上是一个分布式文件系统，存储大量小文件。每个文件就代表一个锁，并且可以保存一些应用层面的小规模数据。用户通过打开、关闭、读取文件来获取共享锁或者独占锁；并通过反向通知机制，向用户发送更新信息。

1.2.3 系统设计

1）设计目标

Chubby系统设计的目标基于以下几点：

（1）粗粒度的锁服务；

（2）高可用、高可靠；

（3）可直接存储服务信息，而无需另建服务；

（4）高扩展性；

在实现时，使用了以下特性：

（1）缓存机制：客户端缓存，避免频繁访问master；

（2）通知机制：服务器会及时通知客户端服务变化；

2）整体架构

Chubby架构并不复杂，如上图分为两个重要组件：

（1）Chubby库：客户端通过调用Chubby库，申请锁服务，并获取相关信息，同时通过租约保持与服务器的连接；

（2）Chubby服务器组：一个服务器组一般由五台服务器组成（至少3台），其中一台master，服务维护与客户端的所有通信；其他服务器不断和主服务器通信，获取用户操作。

1.2.4 系统实现

1）文件系统

Chubby文件系统类似于简单的unix文件系统，但它不支持文件移动操作与硬连接。文件系统由许多Node组成，每个Node代表一个文件，或者一个目录。文件系统使用Berkeley DB来保存每个Node的数据。文件系统提供的API很少：创建文件系统、文件操作、目录操作等简易操作。

2）基于ICE的Chubby通信机制

一种基于ICE的RPC异步机制，核心就是异步，部分组件负责发送，部分组件负责接收。

3）客户端与master的通信

（1）长连接保持连接，连接有效期内，客户端句柄、锁服务、缓存数据均一直有效；

（2）定时双向keep alive；

（3）出错回调是客户端与服务器通信的重点。

下面将说明正常、客户端租约过期、主服务器租约过期、主服务器出错等情况。

（1）正常情况

keep alive是周期性发送的一种消息，它有两方面功能：延长租约有效期，携带事件信息告诉客户端更新。正常情况下，租约会由keep alive一直不断延长。

潜在回调事件包括：文件内容修改、子节点增删改、master出错等。

（2）客户端租约过期

客户端没有收到master的keep alive，租约随之过期，将会进入一个“危险状态”。由于此时不能确定master是否已经终止，客户端必须主动让cache失效，同时，进入一个寻找新的master的阶段。

这个阶段中，客户端会轮询Chubby Cell中非master的其他服务器节点，当客户端收到一个肯定的答复时，他会向新的master发送keep alive信息，告之自己处于“危险状态”，并和新的master建立session，然后把cache中的handler发送给master刷新。

一段时间后，例如45s，新的session仍然不能建立，客户端立马认为session失效，将其终止。当然这段时间内，不能更改cache信息，以求保证数据的一致性。

（3）master租约过期

master一段时间没有收到客户端的keep alive，则其进入一段等待期，此期间内仍没有响应，则master认为客户端失效。失效后，master会把客户端获得的锁，机器打开的临时文件清理掉，并通知各副本，以保持一致性。

（4）主服务器出错

master出错，需要内部进行重新选举，各副本只响应客户端的读取命令，而忽略其他命令。新上任的master会进行以下几步操作：

a，选择新的编号，不再接受旧master的消息；

b，只处理master位置相关消息，不处理session相关消息；

c，等待处理“危险状态”的客户端keep alive；

d，响应客户端的keep alive，建立新的session，同时拒绝其他session相关操作；同事向客户端返回keep alive，警告客户端master fail-over，客户端必须更新handle和lock；

e，等待客户端的session确认keep alive，或者让session过期；

f，再次响应客户端所有操作；

g，一段时间后，检查是否有临时文件，以及是否存在一些lock没有handle；如果临时文件或者lock没有对应的handle，则清除临时文件，释放lock，当然这些操作都需要保持数据的一致性。

4）服务器间的一致性操作

这块考虑的问题是：当master收到客户端请求时（主要是写），如何将操作同步，以保证数据的一致性。

（1）节点数目

一般来说，服务器节点数为5，如果临时有节点被拿走，可预期不久的将来就会加进来。

（2）关于复制

服务器接受客户端请求时，master会将请求复制到所有成员，并在消息中添加最新被提交的请求序号。member收到这个请求后，获取master处被提交的请求序号，然后执行这个序列之前的所有请求，并把其记录到内存的日志里。如果请求没有被master接受，就不能执行。

各member会向master发送消息，master收到>=3个以上的消息，才能够进行确认，发送commit给各member，执行请求，并返回客户端。

如果某个member出现暂时的故障，没有收到部分消息也无碍，在收到来自master的新请求后，主动从master处获得已执行的，自己却还没有完成的日志，并进行执行。

最终，所有成员都会获得一致性的数据，并且，在系统正常工作状态中，至少有3个服务器保持一致并且是最新的数据状态。

5）Chubby系统锁机制

客户端和服务器除了要保存lease对象外，服务器和客户端还需要保存另一张表，用于描述已经加锁的文件及相关信息。由于Chubby系统所使用锁是建议性而非强制性的，这代表着如果有多个锁请求，后达的请求会进入锁等待队列，直到锁被释放。

1.2.5 Chubby使用例子（重点）

1）选master

（1）每个server都试图创建/打开同一个文件，并在该文件中记录自己的服务信息，任何时刻都只有一个服务器能够获得该文件的控制权；

（2）首先创建该文件的server成为主，并写入自己的信息；

（3）后续打开该文件的server成为从，并读取主的信息；

2）进程监控

（1）各个进程都把自己的状态写入指定目录下的临时文件里；

（2）监控进程通过阅读该目录下的文件信息来获得进程状态；

（3）各个进程随时有可能死亡，因此指定目录的数据状态会发生变化；

（4）通过事件机制通知监控进程，读取相关内容，获取最新状态，达到监控目的；

6）总结

Google Chubby提供粗粒度锁服务，它的本质是一个松耦合分布式文件系统；开发者不需要关注复杂的同步协议，直接调用库来取得锁服务，并保证了数据的一致性。

最后要说明的是，最终Chubby系统代码共13700多行，其中ice自动生成6400行，手动编写约8000行，这就是Google牛逼的地方：强大的工程能力，快速稳定的实现，然后用来解决各种业务问题。

2 Redis实现分布式锁

2.2 分布式锁超时问题的处理(只是参考，推荐使用redission框架和ZK做分布式锁)

2.2.1 redis分布式锁的基本实现

redis加锁命令：

SETNX resource_name my_random_value PX 30000

这个命令的作用是在只有这个key不存在的时候才会设置这个key的值（NX选项的作用），超时时间设为30000毫秒（PX选项的作用）这个key的值设为“my_random_value”。这个值必须在所有获取锁请求的客户端里保持唯一。

SETNX 值保持唯一的是为了确保安全的释放锁，避免误删其他客户端得到的锁。举个例子，一个客户端拿到了锁，被某个操作阻塞了很长时间，过了超时时间后自动释放了这个锁，然后这个客户端之后又尝试删除这个其实已经被其他客户端拿到的锁。所以单纯的用DEL指令有可能造成一个客户端删除了其他客户端的锁，通过校验这个值保证每个客户端都用一个随机字符串’签名’了，这样每个锁就只能被获得锁的客户端删除了。

既然释放锁时既需要校验这个值又需要删除锁，那么就需要保证原子性，redis支持原子地执行一个lua脚本，所以我们通过lua脚本实现原子操作。代码如下：

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

2.2.2 业务逻辑执行时间超出锁的超时限制导致两个客户端同时执行（而不是叫同时获取锁的错误写法）的问题

发现问题

但在最近查线上日志的时候偶然发现，有一个业务场景下，分布式锁偶尔会失效，导致有多个线程同时执行了相同的代码。

（这里理解，当一个线程锁超时后，锁会被自动释放，另一个线程可以获取锁，造成多个线程同时执行此段代码，关键就在这个“同时”问题）

我们经过初步排查，定位到是因为在这段代码中间调用了第三方的接口导致。

因为业务代码耗时过长，超过了锁的超时时间，造成锁自动失效，然后另外一个线程意外的持有了锁。于是就出现了多个线程同时执行代码问题。

问题剖析

如果在加锁和释放锁之间的逻辑执行得太长，以至于超出了锁的超时限制，就会出现问题。因为这时候第一个线程持有的锁过期了，临界区的逻辑还没有执行完，(疑问：锁超时了，是不是释放锁呢，既然释放锁了，怎么是二个线程都持有这把锁呢)，这个时候第二个线程就提前重新持有了这把锁，导致临界区代码不能得到严格的串行执行。

不难发现正常情况下锁操作完后都会被手动释放，常见的解决方案是调大锁的超时时间，之后若再出现超时带来的并发问题，人工介入修正数据。这也不是一个完美的方案，因为但业务逻辑执行时间是不可控的，所以还是可能出现超时，当前线程的逻辑没有执行完，其它线程乘虚而入。并且如果锁超时时间设置过长，当持有锁的客户端宕机，释放锁就得依靠redis的超时时间，这将导致业务在一个超时时间周期内不可用。

基本上，如果在执行计算期间发现锁快要超时了，客户端可以给redis服务实例发送一个Lua脚本让redis服务端延长锁的时间，只要这个锁的key还存在而且值还等于客户端设置的那个值。客户端应当只有在失效时间内无法延长锁时再去重新获取锁（基本上这个和获取锁的算法是差不多的）。

启动另外一个线程去检查的问题，这个key是否超时，在某个时间还没释放。

当锁超时时间快到期且逻辑未执行完，延长锁超时时间的伪代码：

if redis.call("get",KEYS[1]) == ARGV[1] then
          redis.call("set",KEYS[1],ex=3000)  
else
          getDLock();//重新获取锁

2.2.3 锁超时限制的解决方案

问题既然已经出现了，那么接下来我们就应该考虑解决方案了。

我们也曾经想过，是否可以通过合理地设置LockTime(锁超时时间)来解决这个问题?

但LockTime的设置原本就很不容易。LockTime设置过小，锁自动超时的概率就会增加，锁异常失效的概率也就会增加，而LockTime设置过大，万一服务出现异常无法正常释放锁，那么出现这种异常锁的时间也就越长。我们只能通过经验去配置，一个可以接受的值，基本上是这个服务历史上的平均耗时再增加一定的buff。

既然这条路走不通了，那么还有其他路可以走么？

当然还是有的，我们可以先给锁设置一个LockTime，然后启动一个守护线程，让守护线程在一段时间后，重新去设置这个锁的LockTime。这种做法，自己手动来实现锁超时间延长，我们可以使用Redission框架来实现锁超时间延长。推荐使用redission来实现

看起来很简单是不是？

但在实际操作中，我们要注意以下几点：

1、和释放锁的情况一致，我们需要先判断锁的对象是否没有变。否则会造成无论谁持有锁，守护线程都会去重新设置锁的LockTime。不应该续的不能瞎续。
2、守护线程要在合理的时间再去重新设置锁的LockTime，否则会造成资源的浪费。不能动不动就去续。
3、如果持有锁的线程已经处理完业务了，那么守护线程也应该被销毁。不能主人都挂了，守护者还在那里继续浪费资源。

代码实现

我们首先先生成一个内部类去实现Runnable，作为守护线程的参数。

public class SurvivalClamProcessor implements Runnable {
 
    private static final int REDIS_EXPIRE_SUCCESS = 1;
 
    SurvivalClamProcessor(String field, String key, String value, int lockTime) {
        this.field = field;
        this.key = key;
        this.value = value;
        this.lockTime = lockTime;
        this.signal = Boolean.TRUE;
    }
 
    private String field;
 
    private String key;
 
    private String value;
 
    private int lockTime;
 
    //线程关闭的标记
    private volatile Boolean signal;
 
    void stop() {
        this.signal = Boolean.FALSE;
    }
 
    @Override
    public void run() {
        int waitTime = lockTime * 1000 * 2 / 3;
        while (signal) {
            try {
                Thread.sleep(waitTime);
                if (cacheUtils.expandLockTime(field, key, value, lockTime) == REDIS_EXPIRE_SUCCESS) {
                    if (logger.isInfoEnabled()) {
                        logger.info("expandLockTime 成功，本次等待{}ms，将重置锁超时时间重置为{}s,其中field为{},key为{}", waitTime, lockTime, field, key);
                    }
                } else {
                    if (logger.isInfoEnabled()) {
                        logger.info("expandLockTime 失败，将导致SurvivalClamConsumer中断");
                    }
                    this.stop();
                }
            } catch (InterruptedException e) {
                if (logger.isInfoEnabled()) {
                    logger.info("SurvivalClamProcessor 处理线程被强制中断");
                }
            } catch (Exception e) {
                logger.error("SurvivalClamProcessor run error", e);
            }
        }
        if (logger.isInfoEnabled()) {
            logger.info("SurvivalClamProcessor 处理线程已停止");
        }
    }
}

其中expandLockTime是通过Lua脚本实现的。延长锁超时的脚本语句和释放锁的Lua脚本类似。

String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('expire', KEYS[1],ARGV[2]) else return '0' end";

在以上代码中，我们将waitTime设置为Math.max(1, lockTime * 2 / 3)，即守护线程许需要等待waitTime后才可以去重新设置锁的超时时间，避免了资源的浪费。

同时在expandLockTime时候也去判断了当前持有锁的对象是否一致，避免了胡乱重置锁超时时间的情况。

然后我们在获得锁的代码之后，添加如下代码：

SurvivalClamProcessor survivalClamProcessor 
    = new SurvivalClamProcessor(lockField, lockKey, randomValue, lockTime);
Thread survivalThread = new Thread(survivalClamProcessor);
survivalThread.setDaemon(Boolean.TRUE);
survivalThread.start();
Object returnObject = joinPoint.proceed(args);
survivalClamProcessor.stop();
survivalThread.interrupt();
return returnObject;

这段代码会先初始化守护线程的内部参数，然后通过start函数启动线程，最后在业务执行完之后，设置守护线程的关闭标记，最后通过interrupt()去中断sleep状态，保证线程及时销毁。

2.2.3 redis的单点故障主从切换带来的两个客户端同时执行（而不是叫同时获取锁的错误写法）的问题

生产中redis一般是主从模式，主节点挂掉时，从节点会取而代之，客户端上却并没有明显感知。原先第一个客户端在主节点中申请成功了一把锁，但是这把锁还没有来得及同步到从节点，主节点突然挂掉了。然后从节点变成了主节点，这个新的节点内部没有这个锁，所以当另一个客户端过来请求加锁时，立即就批准了。这样就会导致系统中同样一把锁被两个客户端同时持有，不安全性由此产生。

不过这种不安全也仅仅是在主从发生 failover 的情况下才会产生，而且持续时间极短，业务系统多数情况下可以容忍。

3 Zookeeper实现分布式锁

3.1 CAP定理：

一个分布式系统不可能在满足分区容错性（P）的情况下同时满足一致性（C）和可用性（A）。在此ZooKeeper保证的是CP，ZooKeeper不能保证每次服务请求的可用性，在极端环境下，ZooKeeper可能会丢弃一些请求，消费者程序需要重新请求才能获得结果。另外在进行leader选举时集群都是不可用，所以说，ZooKeeper不能保证服务可用性。

3.2 BASE理论

BASE理论是基本可用，软状态，最终一致性三个短语的缩写。BASE理论是对CAP中一致性和可用性（CA）权衡的结果，其来源于对大规模互联网系统分布式实践的总结，是基于CAP定理逐步演化而来的，它大大降低了我们对系统的要求。

基本可用：基本可用是指分布式系统在出现不可预知故障的时候，允许损失部分可用性。但是，这绝不等价于系统不可用。比如正常情况下，一个在线搜索引擎需要在0.5秒之内返回给用户相应的查询结果，但由于出现故障，查询结果的响应时间增加了1~2秒。
软状态：软状态指允许系统中的数据存在中间状态，并认为该中间状态的存在不会影响系统的整体可用性，即允许系统在不同节点的数据副本之间进行数据同步的过程存在延时。
最终一致性：最终一致性强调的是系统中所有的数据副本，在经过一段时间的同步后，最终能够达到一个一致的状态。因此，最终一致性的本质是需要系统保证最终数据能够达到一致，而不需要实时保证系统数据的强一致性。

3.3 ZooKeeper特点

顺序一致性：同一客户端发起的事务请求，最终将会严格地按照顺序被应用到 ZooKeeper 中去。
原子性：所有事务请求的处理结果在整个集群中所有机器上的应用情况是一致的，也就是说，要么整个集群中所有的机器都成功应用了某一个事务，要么都没有应用。
单一系统映像：无论客户端连到哪一个 ZooKeeper 服务器上，其看到的服务端数据模型都是一致的。
可靠性：一旦一次更改请求被应用，更改的结果就会被持久化，直到被下一次更改覆盖。

3.4 ZAB协议：

ZAB协议包括两种基本的模式：崩溃恢复和消息广播。当整个 Zookeeper 集群刚刚启动或者Leader服务器宕机、重启或者网络故障导致不存在过半的服务器与 Leader 服务器保持正常通信时，所有服务器进入崩溃恢复模式，首先选举产生新的 Leader 服务器，然后集群中 Follower 服务器开始与新的 Leader 服务器进行数据同步。当集群中超过半数机器与该 Leader 服务器完成数据同步之后，退出恢复模式进入消息广播模式，Leader 服务器开始接收客户端的事务请求生成事物提案（超过半数同意）来进行事务请求处理。

选举算法和流程：FastLeaderElection(默认提供的选举算法)

目前有5台服务器，每台服务器均没有数据，它们的编号分别是1,2,3,4,5,按编号依次启动，它们的选择举过程如下：

服务器1启动，给自己投票，然后发投票信息，由于其它机器还没有启动所以它收不到反馈信息，服务器1的状态一直属于Looking。
服务器2启动，给自己投票，同时与之前启动的服务器1交换结果，由于服务器2的编号大所以服务器2胜出，但此时投票数没有大于半数，所以两个服务器的状态依然是LOOKING。
服务器3启动，给自己投票，同时与之前启动的服务器1,2交换信息，由于服务器3的编号最大所以服务器3胜出，此时投票数正好大于半数，所以服务器3成为leader，服务器1,2成为follower。
服务器4启动，给自己投票，同时与之前启动的服务器1,2,3交换信息，尽管服务器4的编号大，但之前服务器3已经胜出，所以服务器4只能成为follower。
服务器5启动，后面的逻辑同服务器4成为follower。