查看: 2203|回复: 4
用Redis实现分布式锁
认证徽章论坛徽章:277
分布式锁是一个在很多环境中非常有用的原语，它是不同进程互斥操作共享资源的唯一方法。有很多的开发库和博客描述如何使用Redis实现DLM（Distributed Lock Manager），但是每个开发库使用不同的方式，而且相比更复杂的设计与实现，很多库使用一些简单低可靠的方式来实现。
这篇文章尝试提供更标准的算法来使用Redis实现分布式锁。我们提出一种算法，叫做Relock，它实现了我们认为比vanilla单一实例方式更安全的DLM（分布式锁管理）。我们希望社区分析它并提供反馈，以做为更加复杂或替代设计的一个实现。 实现
在说具体算法之前，下面有一些具体的实现可供参考.
(Ruby实现).-
(PHP 实现).-
(Go 实现).-
(Java 实现).
认证徽章论坛徽章:277
安全和活跃性保证
从有效分布式锁的最小保证粒度来说，我们的模型里面只用了3个属性，具体如下：
1. 属性安全: 互斥行.在任何时候，只有一个客户端可以获得锁.
2. 活跃属性A: 死锁自由. 即使一个客户端已经拥用了已损坏或已被分割资源的锁，但它也有可能请求其他的锁.
3. 活跃属性B:容错. 只要大部分Redis节点可用, 客户端就可以获得和释放锁.
为何基于容错的实现还不够
要理解我们所做的改进，就要先分析下当前基于Redis的分布式锁的做法。
使用Redis锁住资源的最简单的方法是创建一对key-value值。利用Redis的超时机制，key被创建为有一定的生存期，因此它最终会被释放。而当客户端想要释放时，直接删除key就行了。
认证徽章论坛徽章:277
一般来说这工作得很好，但有个问题: 这是系统的一个单点。如果Redis主节点挂了呢？当然，我们可以加个子节点，主节点出问题时可以切换过来。不过很可惜，这种方案不可行，因为Redis的主-从复制是异步的，我们无法用其实现互斥的安全特性。
这明显是该模型的一种竞态条件：
1）客户端A在主节点获得了一个锁。
2）主节点挂了，而到从节点的写同步还没完成。
3）从节点被提升为主节点。
4）客户端B获得和A相同的锁。
注意，锁安全性被破坏了！
有时候，在某些情况下这反而工作得很好，例如在出错时，多个客户端可以获得同一个锁。如果这正好是你想要的，那就可以使用主-从复制的方案。否则，我们建议使用这篇文章中描述的方法。
认证徽章论坛徽章:277
单实例的正确实现方案
在尝试解决上文描述的单实例方案的缺陷之前，先让我们确保针对这种简单的情况，怎么做才是无误的，因为这种方案对某些程序而言也是可以接受的，而且这也是我们即将描述的分布式方案的基础。
为了获取锁，方法是这样的：SET resource_name my_random_value NX PX 30000复制代码这条指令将设置key的值，仅当其不存在时生效(NX选项), 且设置其生存期为30000毫秒(PX选项)。和key关联的value值是&my_random_value&。这个值在所有客户端和所有加锁请求中是必须是唯一的。
使用随机值主要是为了能够安全地释放锁，这要同时结合这么个处理逻辑：删除key值当且仅当其已存在并且其value值是我们所期待的。看看以下lua代码：if redis.call(&get&,KEYS[1]) == ARGV[1] then
& & return redis.call(&del&,KEYS[1])
else
& & return 0
end
复制代码
认证徽章论坛徽章:277
这么做很重要，可以避免误删其他客户端创建的锁。例如某个客户端获得了一个锁，但它的处理时长超过了锁的有效时长，之后它删除了这个锁，而此时这个锁可能又被其他客户端给获得了。仅仅做删除是不够安全的，很可能会把其他客户端的锁给删了。结合上面的代码，每个锁都有个唯一的随机值，因此仅当这个值依旧是客户端所设置的值时，才会去删除它。
那么应该怎样生成这个随机值呢？我们使用的是从/dev/urandom读取的20个字节，但你也可以找个更简单的方法，只要能满足任务就行。例如，可以使用/dev/urandom初始化RC4算法，然后用其产生随机数流。更简单的方法是组合unix时间戳和客户端ID, 这并不安全，但对很多环境而言也够用了。
我们所说的key的时间，是指”锁的有效时长“. 它代表两种情况，一种是指锁的自动释放时长，另一种是指在另一个客户端获取锁之前某个客户端占用这个锁的时长，这被限制在从锁获取后开始的一段时间窗口内。
现在我们已经有好的办法获取和释放锁了。在单实例非分布式系统中，只要保证节点没挂掉，这个方法就是安全的。那么让我们把这个概念扩展到分布式的系统中吧，那里可没有这种保证。
itpub.net All Right Reserved. 北京皓辰网域网络信息技术有限公司版权所有　　　　
　北京市公安局海淀分局网监中心备案编号：　广播电视节目制作经营许可证：编号（京）字第1149号Java实现基于Redis的分布式锁 - 莫名的拉风的博客 - ITeye博客
博客分类：
Java实现基于Redis的分布式锁
单JVM内同步好办， 直接用JDK提供的锁就可以了，但是跨进程同步靠这个肯定是不可能的，这种情况下肯定要借助第三方，我这里实现用Redis，当然还有很多其他的实现方式。其实基于Redis实现的原理还算比较简单的，在看代码之前建议大家先去看看原理，我就不翻译了，免得变味了，看懂了之后看代码应该就容易理解了。
时间统一问题：各个客户端加锁时需要获取时间，而这个时间都不应当从本地获取，因为各个客户端的时间并不是一致的，因此需要提供一个TimeServer提供获取时间的服务，下面源码中用到的关于时间服务的三个类(包括TimeServer、TimeClient和Time Client Exception)会在另一篇博客《》给源码.
我这里不实现JDK的java.util.concurrent.locks.Lock接口，而是自定义一个，因为JDK的有个newCondition()方法我这里暂时没实现。这个Lock提供了5个lock方法的变体，可以自行选择使用哪一个来获取锁，我的想法是
最好用带超时返回的那几个方法，因为不这样的话，假如redis挂了，线程永远都在那死循环了(关于这里，应该还可以进一步优化，如果redis挂了，Jedis的操作肯定会抛异常之类的，可以定义个机制让redis挂了的时候通知使用这个lock的用户，或者说是线程)。
package cc.lixiaohui.
import java.util.concurrent.TimeU
public interface Lock extends Releasable{
* 阻塞性的获取锁, 不响应中断
void lock();
* 阻塞性的获取锁, 响应中断
* @throws InterruptedException
void lockInterruptibly() throws InterruptedE
* 尝试获取锁, 获取不到立即返回, 不阻塞
boolean tryLock();
* 超时自动返回的阻塞性的获取锁, 不响应中断
* @param time
* @param unit
* @return {@code true} 若成功获取到锁, {@code false} 若在指定时间内未获取到锁
boolean tryLock(long time, TimeUnit unit);
* 超时自动返回的阻塞性的获取锁, 响应中断
* @param time
* @param unit
* @return {@code true} 若成功获取到锁, {@code false} 若在指定时间内未获取到锁
* @throws InterruptedException 在尝试获取锁的当前线程被中断
boolean tryLockInterruptibly(long time, TimeUnit unit) throws InterruptedE
void unlock();
Releasable.java :
package cc.lixiaohui.
* 代表持有资源的对象, 例如
* &li& 基于jedis的锁自然持有与redis server的连接 &/li&
* &li& 基于时间统一的的锁自然持有与time server的连接&/li&
* 因此锁应该实现该接口, 并在{@link Releasable#resease() release} 方法中释放相关的连接
* @author lixiaohui
public interface Releasable {
* 释放持有的所有资源
void release();
看Lock的抽象实现：
package cc.lixiaohui.
import java.util.concurrent.TimeU
* 锁的骨架实现, 真正的获取锁的步骤由子类去实现.
* @author lixiaohui
public abstract class AbstractLock implements Lock {
* 这里需不需要保证可见性值得讨论, 因为是分布式的锁,
* 1.同一个jvm的多个线程使用不同的锁对象其实也是可以的, 这种情况下不需要保证可见性
* 2.同一个jvm的多个线程使用同一个锁对象, 那可见性就必须要保证了.
protected vol
* 当前jvm内持有该锁的线程(if have one)
private Thread exclusiveOwnerT
public void lock() {
lock(false, 0, null, false);
} catch (InterruptedException e) {
// TODO ignore
public void lockInterruptibly() throws InterruptedException {
lock(false, 0, null, true);
public boolean tryLock(long time, TimeUnit unit) {
return lock(true, time, unit, false);
} catch (InterruptedException e) {
// TODO ignore
public boolean tryLockInterruptibly(long time, TimeUnit unit) throws InterruptedException {
return lock(true, time, unit, true);
public void unlock() {
// TODO 检查当前线程是否持有锁
if (Thread.currentThread() != getExclusiveOwnerThread()) {
throw new IllegalMonitorStateException("current thread does not hold the lock");
unlock0();
setExclusiveOwnerThread(null);
protected void setExclusiveOwnerThread(Thread thread) {
exclusiveOwnerThread =
protected final Thread getExclusiveOwnerThread() {
return exclusiveOwnerT
protected abstract void unlock0();
* 阻塞式获取锁的实现
* @param useTimeout
* @param time
* @param unit
* @param interrupt 是否响应中断
* @throws InterruptedException
protected abstract boolean lock(boolean useTimeout, long time, TimeUnit unit, boolean interrupt) throws InterruptedE
基于Redis的最终实现(not reentrant)，关键的获取锁，释放锁的代码在这个类的lock方法和unlock0方法里，大家可以只看这两个方法然后完全自己写一个：
package cc.lixiaohui.
import java.io.IOE
import java.net.SocketA
import java.util.concurrent.TimeU
import redis.clients.jedis.J
import cc.lixiaohui.lock.time.nio.client.TimeC
* 基于Redis的SETNX操作实现的分布式锁
* 获取锁时最好用lock(long time, TimeUnit unit), 以免网路问题而导致线程一直阻塞
* &a href="http://redis.io/commands/setnx"&SETNX操作参考资料&/a&
* @author lixiaohui
public class RedisBasedDistributedLock extends AbstractLock {
private TimeClient timeC
// 锁的名字
protected String lockK
// 锁的有效时长(毫秒)
protected long lockE
public RedisBasedDistributedLock(Jedis jedis, String lockKey, long lockExpires, SocketAddress timeServerAddr) throws IOException {
this.jedis =
this.lockKey = lockK
this.lockExpires = lockE
timeClient = new TimeClient(timeServerAddr);
// 阻塞式获取锁的实现
protected boolean lock(boolean useTimeout, long time, TimeUnit unit, boolean interrupt) throws InterruptedException{
if (interrupt) {
checkInterruption();
// 超时控制 的时间可以从本地获取, 因为这个和锁超时没有关系, 只是一段时间区间的控制
long start = localTimeMillis();
long timeout = unit.toMillis(time); // if !useTimeout, then it's useless
while (useTimeout ? isTimeout(start, timeout) : true) {
if (interrupt) {
checkInterruption();
long lockExpireTime = serverTimeMillis() + lockExpires + 1;//锁超时时间
String stringOfLockExpireTime = String.valueOf(lockExpireTime);
if (jedis.setnx(lockKey, stringOfLockExpireTime) == 1) { // 获取到锁
// TODO 成功获取到锁, 设置相关标识
setExclusiveOwnerThread(Thread.currentThread());
String value = jedis.get(lockKey);
if (value != null && isTimeExpired(value)) { // lock is expired
// 假设多个线程(非单jvm)同时走到这里
String oldValue = jedis.getSet(lockKey, stringOfLockExpireTime); // getset is atomic
// 但是走到这里时每个线程拿到的oldValue肯定不可能一样(因为getset是原子性的)
// 加入拿到的oldValue依然是expired的，那么就说明拿到锁了
if (oldValue != null && isTimeExpired(oldValue)) {
// TODO 成功获取到锁, 设置相关标识
setExclusiveOwnerThread(Thread.currentThread());
// TODO lock is not expired, enter next loop retrying
public boolean tryLock() {
long lockExpireTime = serverTimeMillis() + lockExpires + 1;//锁超时时间
String stringOfLockExpireTime = String.valueOf(lockExpireTime);
if (jedis.setnx(lockKey, stringOfLockExpireTime) == 1) { // 获取到锁
// TODO 成功获取到锁, 设置相关标识
setExclusiveOwnerThread(Thread.currentThread());
String value = jedis.get(lockKey);
if (value != null && isTimeExpired(value)) { // lock is expired
// 假设多个线程(非单jvm)同时走到这里
String oldValue = jedis.getSet(lockKey, stringOfLockExpireTime); // getset is atomic
// 但是走到这里时每个线程拿到的oldValue肯定不可能一样(因为getset是原子性的)
// 假如拿到的oldValue依然是expired的，那么就说明拿到锁了
if (oldValue != null && isTimeExpired(oldValue)) {
// TODO 成功获取到锁, 设置相关标识
setExclusiveOwnerThread(Thread.currentThread());
// TODO lock is not expired, enter next loop retrying
* Queries if this lock is held by any thread.
* @return {@code true} if any thread holds this lock and
{@code false} otherwise
public boolean isLocked() {
if (locked) {
String value = jedis.get(lockKey);
// TODO 这里其实是有问题的, 想:当get方法返回value后, 假设这个value已经是过期的了,
// 而就在这瞬间, 另一个节点set了value, 这时锁是被别的线程(节点持有), 而接下来的判断
// 是检测不出这种情况的.不过这个问题应该不会导致其它的问题出现, 因为这个方法的目的本来就
// 不是同步控制, 它只是一种锁状态的报告.
return !isTimeExpired(value);
protected void unlock0() {
// TODO 判断锁是否过期
String value = jedis.get(lockKey);
if (!isTimeExpired(value)) {
doUnlock();
public void release() {
jedis.close();
timeClient.close();
private void doUnlock() {
jedis.del(lockKey);
private void checkInterruption() throws InterruptedException {
if(Thread.currentThread().isInterrupted()) {
throw new InterruptedException();
private boolean isTimeExpired(String value) {
// 这里拿服务器的时间来比较
return Long.parseLong(value) & serverTimeMillis();
private boolean isTimeout(long start, long timeout) {
// 这里拿本地的时间来比较
return start + timeout & System.currentTimeMillis();
private long serverTimeMillis(){
return timeClient.currentTimeMillis();
private long localTimeMillis() {
return System.currentTimeMillis();
如果将来还换一种实现方式（比如memcached，数据库之类的），到时直接继承AbstractLock并实现lock(boolean useTimeout, long time, TimeUnit unit, boolean interrupt)， unlock0()方法即可(所谓抽象嘛)
模拟全局ID增长器，设计一个IDGenerator类，该类负责生成全局递增ID，其代码如下：
package cc.lixiaohui.lock.
import java.math.BigI
import java.util.concurrent.TimeU
import cc.lixiaohui.lock.L
import cc.lixiaohui.lock.R
* 模拟分布式环境中的ID生成
* @author lixiaohui
public class IDGenerator implements Releasable{
private static BigInteger id = BigInteger.valueOf(0);
private final L
private static final BigInteger INCREMENT = BigInteger.valueOf(1);
public IDGenerator(Lock lock) {
this.lock =
public String getAndIncrement() {
if (lock.tryLock(3, TimeUnit.SECONDS)) {
// TODO 这里获取到锁, 访问临界区资源
System.out.println(Thread.currentThread().getName() + " get lock");
return getAndIncrement0();
} finally {
lock.unlock();
//return getAndIncrement0();
public void release() {
lock.release();
private String getAndIncrement0() {
String s = id.toString();
id = id.add(INCREMENT);
测试主逻辑：同一个JVM内开两个线程死循环地（循环之间无间隔，有的话测试就没意义了）获取ID（我这里并不是死循环而是跑20s），获取到ID存到同一个Set里面，在存之前先检查该ID在set中是否存在，如果已存在，则让两个线程都停止。如果程序能正常跑完20s，那么说明这个分布式锁还算可以满足要求，如此测试的效果应该和不同JVM(也就是真正的分布式环境中)测试的效果是一样的，下面是测试类的代码：
package cc.lixiaohui.DistributedLock.DistributedL
import java.net.InetSocketA
import java.net.SocketA
import java.util.HashS
import java.util.S
import org.junit.T
import redis.clients.jedis.J
import cc.lixiaohui.lock.L
import cc.lixiaohui.lock.RedisBasedDistributedLockV0_0;
import cc.lixiaohui.lock.RedisBasedDistributedL
import cc.lixiaohui.lock.example.IDG
public class IDGeneratorTest {
private static Set&String& generatedIds = new HashSet&String&();
private static final String LOCK_KEY = "lock.lock";
private static final long LOCK_EXPIRE = 5 * 1000;
public void testV1_0() throws Exception {
SocketAddress addr = new InetSocketAddress("localhost", 9999);
Jedis jedis1 = new Jedis("localhost", 6379);
Lock lock1 = new RedisBasedDistributedLock(jedis1, LOCK_KEY, LOCK_EXPIRE, addr);
IDGenerator g1 = new IDGenerator(lock1);
IDConsumeTask consume1 = new IDConsumeTask(g1, "consume1");
Jedis jedis2 = new Jedis("localhost", 6379);
Lock lock2 = new RedisBasedDistributedLock(jedis2, LOCK_KEY, LOCK_EXPIRE, addr);
IDGenerator g2 = new IDGenerator(lock2);
IDConsumeTask consume2 = new IDConsumeTask(g2, "consume2");
Thread t1 = new Thread(consume1);
Thread t2 = new Thread(consume2);
t1.start();
t2.start();
Thread.sleep(20 * 1000); //让两个线程跑20秒
IDConsumeTask.stopAll();
t1.join();
t2.join();
static String time() {
return String.valueOf(System.currentTimeMillis() / 1000);
static class IDConsumeTask implements Runnable {
private IDGenerator idG
private static v
public IDConsumeTask(IDGenerator idGenerator, String name) {
this.idGenerator = idG
this.name =
public static void stopAll() {
public void run() {
System.out.println(time() + ": consume " + name + " start ");
while (!stop) {
String id = idGenerator.getAndIncrement();
if (id != null) {
if(generatedIds.contains(id)) {
System.out.println(time() + ": duplicate id generated, id = " + id);
generatedIds.add(id);
System.out.println(time() + ": consume " + name + " add id = " + id);
// 释放资源
idGenerator.release();
System.out.println(time() + ": consume " + name + " done ");
说明一点，我这里停止两个线程的方式并不是很好，我是为了方便才这么做的，因为只是测试，最好不要这么做。
跑20s打印的东西太多，前面打印的被clear了，只有差不多跑完的时候才有，下面截图。说明了这个锁能正常工作：
当IDGererator没有加锁(即IDGererator的getAndIncrement方法内部获取id时不上锁)时,测试是不通过的，非常大的概率中途就会停止，下面是不加锁时的测试结果:
这个1秒都不到： 这个也1秒都不到：
在中改进使其可重入
如有问题麻烦指正
浏览 13257
请问一下楼主，if (Thread.currentThread() != getExclusiveOwnerThread()） 应该不需要吧，昨晚测试每秒50个线程，持续500秒，报了throw new IllegalMonitorStateException("current thread does not hold the lock");&& 这个错，是的，你可以去 //DistributedLock 看看可重入的锁， private Thread exclusiveOwnerT这个字段已去掉，博客没更新
莫名的拉风 写道xiaoy81 写道设置超时时间时候不能去本地时间，每个服务器获取的时间不一样，这样可能造成判断key是否超时错误，不知道这样理解对不博主？想了一下，的确是这样，疏忽了，应该写个获取时间的服务，有空再改博主对redis实现分布式读写锁有什么思路吗，请教一下你可以去看看这篇 /blog/2328067 看完后你自己应该会有思路，只是可能这种读写锁代价太大，不适用
xiaoy81 写道设置超时时间时候不能去本地时间，每个服务器获取的时间不一样，这样可能造成判断key是否超时错误，不知道这样理解对不博主？想了一下，的确是这样，疏忽了，应该写个获取时间的服务，有空再改博主对redis实现分布式读写锁有什么思路吗，请教一下
设置超时时间时候不能去本地时间，每个服务器获取的时间不一样，这样可能造成判断key是否超时错误，不知道这样理解对不博主？想了一下，的确是这样，疏忽了，应该写个获取时间的服务，有空再改
莫名的拉风
浏览: 85633 次
来自: 广州
minisnoopy2u 写道您好，ClassLoader,
您好，ClassLoader, targetClass, pr ...
写的不错，感谢分享
import cc.lixiaohui.util.Random ...
漂泊一剑客 写道public MonitorHandler r ...2609人阅读
redis（9）
分布式+高并发（23）
redis系列文章目录
分布式锁介绍
注：本文讲的是分布式互斥锁
在很多互联网产品应用中，有些场景需要加锁处理，比如：秒杀，全局递增ID，楼层生成,还有一些分布式任务调度等等。大部分的解决方案是基于DB实现的，Redis为单进程单线程模式，采用队列模式将并发访问变成串行访问，且多客户端对Redis的连接并不存在竞争关系。其次Redis提供一些命令SETNX，GETSET，可以方便实现分布式锁机制。
本着批判性原则，现在网上可以随手搜索到很多基于Redis的分布式锁实现，但可能是时间原因吧，那些实现方式不是很完美，比如这篇文章，阅读数量挺大的，但是其内容比较老旧，而且也不是很好的解决方案，因为他这种实现机制牵扯到全局时钟一致。在分布式环境中要想实现全局时钟完全同步是非常困难的，个人感觉也没有必要。
完美的解决方案（并非真的完美）
基于set nx ex实现
jedisCluster语法： jedisCluster.set(key, "本机ID", "nx", "ex", 3);
方法声明： set(final String key, final String value, final String nxxx, final String expx,
final long time)
其实这也不是我想出来的，是redis官网提出的解决方案
现在就来看看redis官方推荐的方案
官网将这种算法称为Redlock
安全和可靠性保证
在描述我们的设计之前，我们想先提出三个属性，这三个属性在我们看来，是实现高效分布式锁的基础。
安全属性：互斥，不管任何时候，只有一个客户端能持有同一个锁。
效率属性A：不会死锁，最终一定会得到锁，就算一个持有锁的客户端宕掉或者发生网络分区。
效率属性B：容错，只要大多数Redis节点正常工作，客户端应该都能获取和释放锁。
为什么基于故障切换的方案不够好
为了理解我们想要优化的到底是什么，我们先看下当前大多数基于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持有锁的信息）
当然，在某些特殊场景下，前面提到的这个方案则完全没有问题，比如在宕机期间，多个客户端允许同时都持有锁，如果你可以容忍这个问题的话，那用这个基于复制的方案就完全没有问题，否则的话我们还是建议你采用这篇文章里接下来要描述的方案。
采用单实例的正确实现
在讲述如何用其他方案突破单实例方案的限制之前，让我们先看下是否有什么办法可以修复这个简单场景的问题，因为这个方案其实如果可以忍受竞争条件的话是有望可行的，而且单实例来实现分布式锁是我们后面要讲的算法的基础。 要获得锁，要用下面这个命令： SET resource_name my_random_value NX PX 30000这个命令的作用是在只有这个key不存在的时候才会设置这个key的值（NX选项的作用），超时时间设为30000毫秒（PX选项的作用） 这个key的值设为“my_random_value”。这个值必须在所有获取锁请求的客户端里保持唯一。 基本上这个随机值就是用来保证能安全地释放锁，我们可以用下面这个Lua脚本来告诉Redis：删除这个key当且仅当这个key存在而且值是我期望的那个值。
if redis.call("get",KEYS[1]) == ARGV[1] then
return redis.call("del",KEYS[1])
笔者注： 使用lua脚本可以使得原本非原子操作变成原子的
这个很重要，因为这可以避免误删其他客户端得到的锁，举个例子，一个客户端拿到了锁，被某个操作阻塞了很长时间，过了超时时间后自动释放了这个锁，然后这个客户端之后又尝试删除这个其实已经被其他客户端拿到的锁。所以单纯的用DEL指令有可能造成一个客户端删除了其他客户端的锁，用上面这个脚本可以保证每个客户单都用一个随机字符串’签名’了，这样每个锁就只能被获得锁的客户端删除了。
这个随机字符串应该用什么生成呢？我假设这是从/dev/urandom生成的20字节大小的字符串，但是其实你可以有效率更高的方案来保证这个字符串足够唯一。比如你可以用RC4加密算法来从/dev/urandom生成一个伪随机流。还有更简单的方案，比如用毫秒的unix时间戳加上客户端id，这个也许不够安全，但是也许在大多数环境下已经够用了。
笔者注：锁持有者在处理完工作以后要主动释放锁；另外锁超时时间保证了在锁未被正常释放的时候不会产生死锁
key值的超时时间，也叫做”锁有效时间”。这个是锁的自动释放时间，也是一个客户端在其他客户端能抢占锁之前可以执行任务的时间，这个时间从获取锁的时间点开始计算。 所以现在我们有很好的获取和释放锁的方式，在一个非分布式的、单点的、保证永不宕机的环境下这个方式没有任何问题，接下来我们看看无法保证这些条件的分布式环境下我们该怎么做。
Redlock算法
在分布式版本的算法里我们假设我们有N个Redis master节点，这些节点都是完全独立的，我们不用任何复制或者其他隐含的分布式协调算法。我们已经描述了如何在单节点环境下安全地获取和释放锁。因此我们理所当然地应当用这个方法在每个单节点里来获取和释放锁。在我们的例子里面我们把N设成5，这个数字是一个相对比较合理的数值，因此我们需要在不同的计算机或者虚拟机上运行5个master节点来保证他们大多数情况下都不会同时宕机。一个客户端需要做如下操作来获取锁：
获取当前时间（单位是毫秒）。
轮流用相同的key和随机值在N个节点上请求锁，在这一步里，客户端在每个master上请求锁时，要求redis超时时间比锁的释放时间小的多。比如如果锁自动释放时间是10秒钟，那每个节点锁请求的超时时间可能是5-50毫秒的范围，这个可以防止一个客户端在某个宕掉的master节点上阻塞过长时间，如果一个master节点不可用了，我们应该尽快尝试下一个master节点。
客户端计算第二步中获取锁所花的时间，只有当客户端在大多数master节点上成功获取了锁（在这里是3个），而且总共消耗的时间不超过锁释放时间，这个锁就认为是获取成功了。
如果锁获取成功了，那现在锁自动释放时间就是最初的锁释放时间减去之前获取锁所消耗的时间。
如果锁获取失败了，不管是因为获取成功的锁不超过一半（N/2+1)还是因为总消耗时间超过了锁释放时间，客户端都会到每个master节点上释放锁，即便是那些他认为没有获取成功的锁。
这个算法是否是异步的？
这个算法是基于一个假设：虽然不存在可以跨进程的同步时钟，但是不同进程时间都是以差不多相同的速度前进，这个假设不一定完全准确，但是和自动释放锁的时间长度相比不同进程时间前进速度差异基本是可以忽略不计的。这个假设就好比真实世界里的计算机：每个计算机都有本地时钟，但是我们可以说大部分情况下不同计算机之间的时间差是很小的。 现在我们需要更细化我们的锁互斥规则，只有当客户端能在T时间内完成所做的工作才能保证锁是有效的（详见算法的第3步），T的计算规则是锁失效时间T1减去一个用来补偿不同进程间时钟差异的delta值（一般只有几毫秒而已） 如果想了解更多基于有限时钟差异的类似系统，可以参考这篇有趣的文章：《Leases: an efficient fault-tolerant mechanism for distributed file cache consistency.》
失败的重试
当一个客户端获取锁失败时，这个客户端应该在一个随机延时后进行重试，之所以采用随机延时是为了避免不同客户端同时重试导致谁都无法拿到锁的情况出现。同样的道理客户端越快尝试在大多数Redis节点获取锁，出现多个客户端同时竞争锁和重试的时间窗口越小，可能性就越低，所以最完美的情况下，客户端应该用多路传输的方式同时向所有Redis节点发送SET命令。 这里非常有必要强调一下客户端如果没有在多数节点获取到锁，一定要尽快在获取锁成功的节点上释放锁，这样就没必要等到key超时后才能重新获取这个锁（但是如果网络分区的情况发生而且客户端无法连接到Redis节点时，会损失等待key超时这段时间的系统可用性）
释放锁比较简单，因为只需要在所有节点都释放锁就行，不管之前有没有在该节点获取锁成功。
安全性的论证
这个算法到底是不是安全的呢？我们可以观察不同场景下的情况来理解这个算法为什么是安全的。 开始之前，让我们假设客户端可以在大多数节点都获取到锁，这样所有的节点都会包含一个有相同存活时间的key。但是需要注意的是，这个key是在不同时间点设置的，所以这些key也会在不同的时间超时，但是我们假设最坏情况下第一个key是在T1时间设置的（客户端连接到第一个服务器时的时间），最后一个key是在T2时间设置的（客户端收到最后一个服务器返回结果的时间），从T2时间开始，我们可以确认最早超时的key至少也会存在的时间为MIN_VALIDITY=TTL-(T2-T1)-CLOCK_DRIFT，TTL是锁超时时间、（T2-T1）是最晚获取到的锁的耗时，CLOCK_DRIFT是不同进程间时钟差异，这个是用来补偿前面的(T2-T1）。其他的key都会在这个时间点之后才会超时，所以我们可以确定这些key在这个时间点之前至少都是同时存在的。
在大多数节点的key都set了的时间段内，其他客户端无法抢占这个锁，因为在N/2+1个客户端的key已经存在的情况下不可能再在N/2+1个客户端上获取锁成功，所以如果一个锁获取成功了，就不可能同时重新获取这个锁成功（不然就违反了分布式锁互斥原则），然后我们也要确保多个客户端同时尝试获取锁时不会都同时成功。 如果一个客户端获取大多数节点锁的耗时接近甚至超过锁的最大有效时间时（就是我们为SET操作设置的TTL值），那么系统会认为这个锁是无效的同时会释放这些节点上的锁，所以我们仅仅需要考虑获取大多数节点锁的耗时小于有效时间的情况。在这种情况下，根据我们前面的证明，在MIN_VALIDITY时间内，没有客户端能重新获取锁成功，所以多个客户端都能同时成功获取锁的结果，只会发生在多数节点获取锁的时间都大大超过TTL时间的情况下，实际上这种情况下这些锁都会失效 。 我们非常期待和欢迎有人能提供这个算法安全性的公式化证明，或者发现任何bug。
这个系统的性能主要基于以下三个主要特征：
锁自动释放的特征（超时后会自动释放），一定时间后某个锁都能被再次获取。
客户端通常会在不再需要锁或者任务执行完成之后主动释放锁，这样我们就不用等到超时时间会再去获取这个锁。
当一个客户端需要重试获取锁时，这个客户端会等待一段时间，等待的时间相对来说会比我们重新获取大多数锁的时间要长一些，这样可以降低不同客户端竞争锁资源时发生死锁的概率。
然而，我们在网络分区时要损失TTL的可用性时间，所以如果网络分区持续发生，这个不可用会一直持续。这种情况在每次一个客户端获取到了锁并在释放锁之前被网络分区了时都会出现。
基本来说，如果持续的网络分区发生的话，系统也会在持续不可用。
性能、故障恢复和fsync
很多使用Redis做锁服务器的用户在获取锁和释放锁时不止要求低延时，同时要求高吞吐量，也即单位时间内可以获取和释放的锁数量。为了达到这个要求，一定会使用多路传输来和N个服务器进行通信以降低延时（或者也可以用假多路传输，也就是把socket设置成非阻塞模式，发送所有命令，然后再去读取返回的命令，假设说客户端和不同Redis服务节点的网络往返延时相差不大的话）。
然后如果我们想让系统可以自动故障恢复的话，我们还需要考虑一下信息持久化的问题。
为了更好的描述问题，我们先假设我们Redis都是配置成非持久化的，某个客户端拿到了总共5个节点中的3个锁，这三个已经获取到锁的节点中随后重启了，这样一来我们又有3个节点可以获取锁了（重启的那个加上另外两个），这样一来其他客户端又可以获得这个锁了，这样就违反了我们之前说的锁互斥原则了。
如果我们启用AOF持久化功能，情况会好很多。举例来说，我们可以发送SHUTDOWN命令来升级一个Redis服务器然后重启之，因为Redis超时时效是语义层面实现的，所以在服务器关掉期间时超时时间还是算在内的，我们所有要求还是满足了的。然后这个是基于我们做的是一次正常的shutdown，但是如果是断电这种意外停机呢？如果Redis是默认地配置成每秒在磁盘上执行一次fsync同步文件到磁盘操作，那就可能在一次重启后我们锁的key就丢失了。理论上如果我们想要在所有服务重启的情况下都确保锁的安全性，我们需要在持久化设置里设置成永远执行fsync操作，但是这个反过来又会造成性能远不如其他同级别的传统用来实现分布式锁的系统。 然后问题其实并不像我们第一眼看起来那么糟糕，基本上只要一个服务节点在宕机重启后不去参与现在所有仍在使用的锁，这样正在使用的锁集合在这个服务节点重启时，算法的安全性就可以维持，因为这样就可以保证正在使用的锁都被所有没重启的节点持有。 为了满足这个条件，我们只要让一个宕机重启后的实例，至少在我们使用的最大TTL时间内处于不可用状态，超过这个时间之后，所有在这期间活跃的锁都会自动释放掉。 使用延时重启的策略基本上可以在不适用任何Redis持久化特性情况下保证安全性，然后要注意这个也必然会影响到系统的可用性。举个例子，如果系统里大多数节点都宕机了，那在TTL时间内整个系统都处于全局不可用状态（全局不可用的意思就是在获取不到任何锁）。
扩展锁来使得算法更可靠
如果客户端做的工作都是由一些小的步骤组成，那么就有可能使用更小的默认锁有效时间，而且扩展这个算法来实现一个锁扩展机制。基本上，客户端如果在执行计算期间发现锁快要超时了，客户端可以给所有服务实例发送一个Lua脚本让服务端延长锁的时间，只要这个锁的key还存在而且值还等于客户端获取时的那个值。 客户端应当只有在失效时间内无法延长锁时再去重新获取锁（基本上这个和获取锁的算法是差不多的） 然而这个并不会对从本质上改变这个算法，所以最大的重新获取锁数量应该被设置成合理的大小，不然性能必然会受到影响。
大多数情况下只要使用set nx ex然后主动解锁就可以了。不过这样要考虑到其局限性，上面已经讲到。
任何方案都有其存在的价值以及其适应场景，几乎没有一劳永逸的完美的解决方法，不要一概而论，选择合适的就好。
&&相关文章推荐
* 以上用户言论只代表其个人观点，不代表CSDN网站的观点或立场
访问：177817次
积分：2700
积分：2700
排名：第13024名
原创：87篇
转载：35篇
评论：40条
文章：10篇
阅读：32035
文章：14篇
阅读：27536
java高级工程师，5年工作经验，一直专注于java领域的学习研究。
对java分布式高并发等有深入研究.
曾就职于 京东、乐视、网易 等公司.
联系方式：
(6)(1)(2)(10)(32)(20)(1)(2)(4)(13)(11)(5)(14)(2)