我们应该怎样选择分布式锁

在上篇文章中

分布式锁的多种实现方式

我介绍了分布式锁的几种实现方式，也有朋友提出，Redis实现分布式锁在比如机器时间回退的情况下会出问题，参考https://martin.kleppmann.com/2016/02/08/how-to-do-distributed-locking.html

接下去我想从这篇文章出发，分析如下几个问题：

上面的文章中Martin对Redlock提出了什么批评，以及Redis的作者antirez如何反驳的；

zookeeper分布式锁的详细剖析，以及它可能出现的问题；还有我们应该怎么选择分布式锁

同时在上篇文章的结尾中，我们应该如何设置锁的超时时间，在这篇文章中也会进行解答

Martin对Redlock提出了什么批评？

Martin认为Redlock有如下两个问题：

Redlock过于依赖系统时间，Martin提出了一个Redlock可能会发生的问题，假如有A、B、C、D、E五个节点，假如发生如下序列：

1、service1成功锁住了其中的3个节点A、B、C，而D和E没有锁住

2、节点C上的时钟发生了向前跳跃，导致它上面维护的锁快速过期

3、service2成功锁住了其中的3个节点C、D、E，获取锁成功

这样，service1和service2就获取到了同一把锁。

发生这个的原因是Redlock对系统的时钟有比较强的依赖，一旦系统的时间变的不确定，Redlock的安全性也就得不到保证了。

锁的超时时间，如何设置这个时间是一个两难的问题，同样假如有A、B、C、D、E五个节点，假如发生如下序列：

1、service1获取到了一个锁

2、service1执行了很长时间（可能发生GC pause）

3、在service1执行期间内，锁超时了，过期了，而service1依然在执行业务

4、service2获取了锁

5、service2很快执行完了业务，往数据库中写数据

6、service1执行完毕，但不知道自己的锁过期了，依然去往数据库中写数据

以上两个客户端，发生了写冲突，锁的互斥完全失效了

针对于锁的超时，Martin提出了应该有如下一个fencing token的机制，也就是采用数字递增的方式去解决：

1、service1获取到了锁，返回了一个初始令牌，令牌数字为1

2、service1执行了很长时间

3、在service1执行期间内，锁超时了，过期了，而service1依然在执行业务

4、service2获取了锁，返回了一个令牌，同时令牌数字递增，令牌数字递增为2

5、service2很快执行完了业务，往数据库中写数据

6、service2写数据时，判断当前的令牌是否等于传递的令牌数字，判断传入的令牌和当前的令牌值都为2，写入成功

7、service1执行完毕，当往数据库中写数据

8、service1写数据时，判断传入的令牌和当前的令牌数字不匹配，拒绝请求，写入失败

这看上去很像CAS或者乐观锁，Zookeeper也有类似的解决方案，我们接下去会说到。

总之，Martin认为Redlock不够安全，简直不伦不类（neither fish nor fowl）。

Redis的作者antirez如何反驳

针对时间跳跃的问题，antirez认为：对于手动修改时钟，这种人为原因，在生产环境上不要去做就行了。也可以使用一个不会发生跳跃的时钟程序。而且，Redlock对时钟的要求，并不需要完全精确。可能是有一定的误差，不过只要误差不超过一定范围，就对Redlock不会产生影响。这在实际环境中是完全合理的，比如即使跳跃0.5秒，可能在实际环境中，并不会产生什么坏的影响。

针对fencing token的机制，antirez认为这个顺序没有意义，既然资源服务器本身都能提供互斥的原子操作了（比如mysql的锁），为什么还需要一个分布式锁呢？即使真的需要，Redlock算法提供的随机数也能满足这需求，可以通过“Check and Set“来实现，类似于CAS的操作来实现这个需求。

antirez的解释逻辑清晰，我们大多数情况下，我们真的需要一个绝对安全的锁吗？同时认为fencing token的机制中的这个数字是类似于zk递增的，还是随机的，是没有关系的，只要能互斥就行了。另外zk就真的百分百安全吗？

zookeeper如何实现分布式锁？

zk创建的节点有4种，实现分布式锁用的是顺序临时节点，这个节点的特性是，生命周期和客户端会话（Session）绑定，即创建节点的客户端会话一旦失效，那么这个节点也会被清除，zk实现分布式锁的步骤如下：

1、客户端调用create( )方法创建名为“_locknode_/guid-lock-”的临时顺序节点（类型为EPHEMERAL_SEQUENTIAL）

2、客户端调用getChildren(“_locknode_”)方法来获取所有已经创建的子节点，同时在这个节点上注册上子节点变更通知的Watcher。

3、客户端获取到所有子节点path之后，如果发现自己在步骤1中创建的节点是所有节点中序号最小的，那么就认为这个客户端获得了锁。

4、如果在步骤3中发现自己并非是所有子节点中最小的，说明自己还没有获取到锁，就开始等待，直到下次子节点变更通知的时候，再进行子节点的获取，判断是否获取锁。

释放锁的过程相对比较很简单，就是删除自己创建的那个子节点即可。

zookeeper实现分布式锁相对于Redis有什么优势？

临时顺序节点，这个是相对于Redis确实是一个优势，能在需要的时候自动释放锁，如果创建znode的那个节点崩溃了，也能绝对保证释放，这是znode的一个特性。这看起来很完美，没有Redlock的过期时间问题。

zookeeper实现的分布式锁到底真的安全吗？

zookeeper是通过session维护和客户端的通讯的，也是通过session监测某一个客户端是否崩溃了，这个session依赖的是心跳来维护和客户端的连接，如果长时间收不到客户端的心跳（session过期时间），那么就认为这个客户端过期了，创建的znode节点也会自动删除。

zookeeper可能发生的羊群效应以及如何避免？

上面这个分布式锁的实现中，大体能够满足了一般的分布式集群竞争锁的需求，比如10台机器以内。

但是我们仔细阅读上面的步骤4，服务器会发送大量的时间通知，原因是：“发现自己并非是所有子节点中最小的”，大多数的判断结果都是，自己并非是序号最小的节点，从而继续等待下一次通知，如果在集群规模比较大的情况下，看上去不怎么合理，会造成很大的性能影响。

更好的实现应该如下：http://zookeeper.apache.org/doc/r3.4.9/recipes.html#sc_recipes_Locks

让我们来分析它实现的步骤：

1、客户端调用create()方法创建名为"_locknode_/lock-"的节点，节点类型EPHEMERAL_SEQUENTIAL

2、客户端调用getChildren( )获取已经创建的节点，不注册任何的watch

3、如果发现自己在步骤1创建的节点序号最小，说明获取到了锁

4、如果在步骤3中发现自己不是节点中最小的，说明自己还没有获取到锁，此时需要找到比自己小的节点，然后调用exist( )方法，同时注册事件监听

5、如果exists( ) 返回false，跳转到步骤2，否则，收到节点被移除的通知后，进入步骤2

我们应该怎么选择分布式锁

我们应该区分，分布式锁的用途和业务场景，如果从安全性的角度上考虑，我们要保证绝对的一致性，建议使用zookeeper，同时还要考虑，是否还要使用数据库的锁。

如果我们只是为了协调各个服务，防止重复处理，锁偶尔失效也可以接受，可以使用Redis。

在文章的最后，贴一个之前写的一个通过后台守护线程，解决使用Redis锁时如何不设置超时时间，让程序自动检测的办法，供大家参考，整体思路就是通过守护线程来维护和程序的心跳：

import java.net.InetAddress;

import java.net.NetworkInterface;

import java.util.ArrayList;

import java.util.List;

public class RedisLock {

//启动时间

private static long SYSTEM_START_TIME = System.currentTimeMillis();

//机器网卡mac地址

private static String MAC;

//进程pid

private static String PID;

// 守护线程val前缀，MAC + "_" + PID + "_" + SYSTEM_START_TIME + "_"

private static String KEEPLIVE_VAL_PRE;

//守护线程

private Thread keepliveThread;

//守护线程key前缀

private String keepliveInfoKeyPre;

//守护线程key nodeKeepliveInfoKeyPre + mac + pid + systemStartTime

private String keepliveInfoKey;

//守护线程定时监测间隔时间

private long loopKeepliveInterval;

//守护线程key失效时间

private int keepliveInfoExpire;

/**

* 构造方法

* @param jedis

* @param nodeKeepLiveInfoKeyPre 前缀

* @param loopKeepliveInterval 守护线程sleep时间

*/

public RedisLock(String nodeKeepLiveInfoKeyPre, long loopKeepliveInterval) {

init();

this.keepliveInfoKeyPre = nodeKeepLiveInfoKeyPre;

this.keepliveInfoKey = getKeepliveKey(MAC, PID, String.valueOf(SYSTEM_START_TIME));

//需要保证时间上keepliveInfoExpire大于loopKeepliveInterval

this.loopKeepliveInterval = loopKeepliveInterval;

this.keepliveInfoExpire = (int) (loopKeepliveInterval) / 1000 * 2;

initKeepLive();

}

/**

* 初始化方法

*/

public void init(){

//通过RuntimeMXBean获取进程PID

String name = ManagementFactory.getRuntimeMXBean().getName();

PID = name.split("@")[0];

try {

//获取本地MAC地址

InetAddress ia = InetAddress.getLocalHost();

byte[] macBytes = NetworkInterface.getByInetAddress(ia).getHardwareAddress();

StringBuffer sb = new StringBuffer("");

for (int i = 0; i

//字节转换为整数

int temp = macBytes[i] & 0xff;

String str = Integer.toHexString(temp);

if (str.length() == 1) {

sb.append("0" + str);

} else {

sb.append(str);

}

}

MAC = sb.toString().toUpperCase();

} catch (Exception e) {

e.printStackTrace();

}

//根据上面的结果，生成keeplive值前缀

KEEPLIVE_VAL_PRE = MAC + "_" + PID + "_" + SYSTEM_START_TIME + "_";

}

/**

* 初始化守护线程

*/

private void initKeepLive() {

keepliveThread = new Thread(() -> {

Jedis jedis = RedisUtils.getPoll();

String keepliveVal = null;

while (true) {

try {

keepliveVal = getKeepliveVal();

//如果 key 已经存在， SETEX 命令将覆写旧值

jedis.setex(keepliveInfoKey, keepliveInfoExpire, keepliveVal);

Thread.sleep(loopKeepliveInterval);

} catch (Exception e) {

e.printStackTrace();

}

}

}, "lock-keeplive-thread");

keepliveThread.setDaemon(true);

keepliveThread.start();

}

/**

* 加锁

* @param lockKey

*/

public boolean lock(String lockKey) {

Jedis jedis = RedisUtils.getPoll();

//加锁

if (1 == jedis.setnx(lockKey, getLockVal())) {

return true;

}

String lockVal = jedis.get(lockKey);

if(lockVal!=null && lockVal.equals(getLockVal())){

//可重入性

return true;

}

//拿到这把锁对应的守护线程的key

String nodeInfoKey = getKeepliveKey(lockVal);

String keepliveVal = jedis.get(nodeInfoKey);

if(keepliveVal == null){

//加锁

unlock(lockKey, lockVal);

if (1 == jedis.setnx(lockKey, getLockVal())) {

return true;

}

}

return false;

}

/**

* 解锁

*/

public void unlock(String lockKey) {

String lockValue = getLockVal();

unlock(lockKey, lockValue);

}

private void unlock(String lockKey, String lockValue){

Jedis jedis = RedisUtils.getPoll();

final StringBuilder luaScript = new StringBuilder("");

luaScript.append("\nlocal v = redis.call('GET', KEYS[1]);");

luaScript.append("\nlocal r= 0;");

luaScript.append("\nif v == ARGV[1] then");

luaScript.append("\nr = redis.call('DEL',KEYS[1]);");

luaScript.append("\nend");

luaScript.append("\nreturn r");

final List keys = new ArrayList();

keys.add(lockKey);

final List args = new ArrayList();

args.add(lockValue);

jedis.eval(luaScript.toString(), keys, args);

}

/**

* LOCK时锁的val

* @return

*/

private String getLockVal() {

return MAC + "_" + PID + "_" + SYSTEM_START_TIME;

}

/**

* 根据pid、mac、时间戳，获取守护线程的key

* @return

*/

private String getKeepliveKey(String mac, String pid, String systemStartTime) {

String nodeKeepLiveInfoKey = keepliveInfoKeyPre + mac + pid + systemStartTime;

return nodeKeepLiveInfoKey;

}

/**

* 根据Keeplive的val获取守护线程的key

* @return

*/

private String getKeepliveKey(String nodeLockInfo) {

String[] meta = nodeLockInfo.split("_");

return getKeepliveKey(meta[0], meta[1], meta[2]);

}

/**

* 生成Keeplive的val

* @param keepliveValPre

* @return

*/

private String getKeepliveVal() {

return KEEPLIVE_VAL_PRE + String.valueOf(System.currentTimeMillis());

}

}