原创 怎么写分布式锁框架？零基础新手教程

分布式锁常常通过数据库乐观锁、基于Redis的分布式锁和基于Zookeeper的分布式锁三种方式来实现。本文是一个零基础新手教程，将教大家怎么写分布式锁框架。有兴趣的小伙伴就赶紧看下去吧！

一、核心代码如下：

import org.springframework.beans.factory.annotation.Autowired;
　　

import org.springframework.context.annotation.Bean;
　　

import org.springframework.data.redis.core.StringRedisTemplate;
　　

import org.springframework.data.redis.core.script.DefaultRedisScript;
　　

import org.springframework.stereotype.Component;
　　

import java.util.Arrays;
　　

import java.util.concurrent.TimeUnit;
　　

@Component
　　

public class RedisLock {
　　

@Autowired
　　

private StringRedisTemplate template;
　　

@Autowired
　　

private DefaultRedisScript<Long> redisScript;
　　

private static final Long RELEASE_SUCCESS = 1L;
　　

private long timeout = 3000;
　　

public boolean lock(String key, String value) {
　　

//执行set命令
　　

Boolean absent = template.opsForValue().setIfAbsent(key, value, timeout, TimeUnit.MILLISECONDS);//1
　　

//其实没必要判NULL，这里是为了程序的严谨而加的逻辑
　　

if (absent == null) {
　　

return false;
　　

}
　　

//是否成功获取锁
　　

return true;
　　

}
　　

public boolean unlock(String key, String value) {
　　

//使用Lua脚本：先判断是否是自己设置的锁，再执行删除
　　

Long result = template.execute(redisScript, Arrays.asList(key,value));
　　

//返回最终结果
　　

return RELEASE_SUCCESS.equals(result);
　　

}
　　

public void setTimeout(long timeout) {
　　

this.timeout = timeout;
　　

}
　　

@Bean
　　

public DefaultRedisScript<Long> defaultRedisScript() {
　　

DefaultRedisScript<Long> defaultRedisScript = new DefaultRedisScript<>();
　　

defaultRedisScript.setResultType(Long.class);
　　

defaultRedisScript.setScriptText("if redis.call('get', KEYS[1]) == KEYS[2] then return redis.call('del', KEYS[1]) else return 0 end");
　　

return defaultRedisScript;
　　

}
　　

}
　　

执行上面的setIfAbsent()方法就只会导致两种结果：

1. 当前没有锁（key不存在），那么就进行加锁操作，并对锁设置个有效期，同时value表示加锁的客户端。

2. 已有锁存在，不做任何操作。
　　

在任意时刻，该代码都能保证只有一个客户端能持有锁，并且每一个分布式锁都加了过期时间，保证不会出现死锁，容错性暂时不考虑的话，加锁和解锁通过key保证了对多个客户端而言都是同一把锁，value的作用则是保证对同一把锁的加锁和解锁操作都是同一个客户端。
　　

二、为什么上述方案不够好
　　

为了理解我们想要提高的到底是什么，我们先看下当前大多数基于Redis的分布式锁三方库的现状。 用Redis来实现分布式锁最简单的方式就是在实例里创建一个键值，创建出来的键值一般都是有一个超时时间的（这个是Redis自带的超时特性），所以每个锁最终都会释放（参见前文属性2）。而当一个客户端想要释放锁时，它只需要删除这个键值即可。 表面来看，这个方法似乎很管用，但是这里存在一个问题：在我们的系统架构里存在一个单点故障，如果Redis的master节点宕机了怎么办呢？有人可能会说：加一个slave节点！在master宕机时用slave就行了！但是其实这个方案明显是不可行的，因为这种方案无法保证第1个安全互斥属性，因为Redis的复制是异步的。 总的来说，这个方案里有一个明显的竞争条件（race condition），举例来说：
　　

客户端A在master节点拿到了锁。
　　

master节点在把A创建的key写入slave之前宕机了。
　　

slave变成了master节点
　　

B也得到了和A还持有的相同的锁（因为原来的slave里还没有A持有锁的信息）
　　

当然，在某些特殊场景下，前面提到的这个方案则完全没有问题，比如在宕机期间，多个客户端允许同时都持有锁，如果你可以容忍这个问题的话，那用这个基于复制的方案就完全没有问题，否则的话我还是建议你对上述方案进行改进。比如，考虑使用Redlock算法。
　　

三、Redlock算法
　　

在分布式版本的算法里我们假设我们有N个Redis master节点，这些节点都是完全独立的，我们不用任何复制或者其他隐含的分布式协调算法。我们已经描述了如何在单节点环境下安全地获取和释放锁。因此我们理所当然地应当用这个方法在每个单节点里来获取和释放锁。在我们的例子里面我们把N设成5，这个数字是一个相对比较合理的数值，因此我们需要在不同的计算机或者虚拟机上运行5个master节点来保证他们大多数情况下都不会同时宕机。一个客户端需要做如下操作来获取锁：
　　

获取当前时间（单位是毫秒）。
　　

轮流用相同的key和随机值在N个节点上请求锁，在这一步里，客户端在每个master上请求锁时，会有一个和总的锁释放时间相比小的多的超时时间。比如如果锁自动释放时间是10秒钟，那每个节点锁请求的超时时间可能是5-50毫秒的范围，这个可以防止一个客户端在某个宕掉的master节点上阻塞过长时间，如果一个master节点不可用了，我们应该尽快尝试下一个master节点。

客户端计算第二步中获取锁所花的时间，只有当客户端在大多数master节点上成功获取了锁（在这里是3个），而且总共消耗的时间不超过锁释放时间，这个锁就认为是获取成功了。
　　

如果锁获取成功了，那现在锁自动释放时间就是最初的锁释放时间减去之前获取锁所消耗的时间。
　　

如果锁获取失败了，不管是因为获取成功的锁不超过一半（N/2+1)还是因为总消耗时间超过了锁释放时间，客户端都会到每个master节点上释放锁，即便是那些他认为没有获取成功的锁。

看完零基础新手教程，相信大家都明白了“怎么写分布式锁框架”。觉得本篇文章有用的小伙伴，不妨分享一下让更多人看见吧！