Redis分布式锁

收藏：Redis实现分布式锁

1.前提准备

1.1.目标

知道什么是分布式锁

知道分布式锁的几种实现方式知道Redis分布式锁原理

学会Lua脚本的编写学学如何执行Lua脚本掌握Redission的使用

1.2.基础

必须的前置知识包括：

Redis的基本命令

JDK中的线程同步方式，例如synchronize关键字，Lock 等

下列知识如果也会，更好不过了：

SpringBoot

SpringDataRedis的基本使用

zookeeper的使用

2.什么是分布式锁

在讨论分布式锁前，我们先假设一个业务场景：

2.1.业务场景

我们假设一个这样的业务场景：

在电商中，用户购买商品需要扣减商品库存，一般有两种扣减库存方式：下单减库存

优点：用户体验好，下单成功，库存直接扣除，用户支付不会出现库存不足情况

缺点：用户一直不付款，这个商品的库存就会被占用，其他人就无法购买了。支付减库存

优点：不会导致库存被恶意锁定，对商家有利

缺点：用户体验不好，用户支付时可能商品库存不足了，会导致用户交易失败

那么，我们一般为了用户体验，会采用下单减库存。但是为了解决下单减库存的缺陷，会创建一个定时任务，定时去清理超时未支付的订单。

在这个定时任务中，需要完成的业务步骤主要包括：

查询超时未支付订单，获取订单中商品信息
修改这些未支付订单的状态，为已关闭
恢复订单中商品扣减的库存

但是，如果我们给订单服务搭建一个100台服务节点的集群，那么就会在同一时刻有100个定时任务触发并执行，设想一下这样的场景：

订单服务A执行了步骤1，但还没有执行步骤B

订单服务B执行了步骤1，于是查询到了与订单服务A查询到的一样的数据订单服务A执行步骤2和3，此时订单中对应商品的库存已经恢复了

订单服务B也执行了步骤2和步骤3，此时订单中对应商品的库存再次被增加

库存被错误的恢复了多次，。

因为任务的并发执行，出现了线程安全问题，商品库存被错误的增加了多次，你能想到解决办法吗？

2.2.为什么需要分布式锁

对于线程安全问题，我们都很熟悉了，传统的解决方案就是对线程操作资源的代码加锁。如图：

理想状态下，加了锁以后，在当前订单服务执行时，其它订单服务需要等待当前订单服务完成业务后才能执行，这样就避免了线程安全问题的发生。

但是，这样真的能解决问题吗？答案时否定的，为什么呢。

2.2.1.线程锁

我们通常使用的synchronized或者Lock都是线程锁，对同一个JVM进程内的多个线程有效。因为锁的本质是内存中存放一个标记，记录获取锁的线程时谁，这个标记对每个线程都可见。

获取锁：就是判断标记中是否已经有线程存在，如果有，则获取锁失败，如果没有，在标记中记录当前线程

释放锁：就是删除标记中保存的线程，并唤醒等待队列中的其它线程

因此，锁生效的前提是：

互斥：锁的标记只有一个线程可以获取共享：标记对所有线程可见

然而我们启动的多个订单服务，就是多个JVM，内存中的锁显然是不共享的，每个JVM进程都有自己的锁，自然无法保证线程的互斥了，如图：

要解决这个问题，就必须保证各个订单服务能够共享内存中的锁标记，此时，分布式锁就闪亮登场了！

2.2.2.分布式锁

线程锁时一个多线程可见的内存标记，保证同一个任务，同一时刻只能被多线程中的某一个执行。但是这样的锁在分布式系统中，多进程环境下，就达不到预期的效果了。

分布式锁实现有多种方式，其原理都基本类似，只要满足下列要求即可：

多进程可见：多进程可见，否则就无法实现分布式效果

互斥（）：同一时刻，只能有一个进程获得锁，执行任务后释放锁可重入（可选）：同一个任务再次获取改锁不会被死锁

阻塞锁（可选）：获取失败时，具备重试机制，尝试再次获取锁性能好（可选）：效率高，应对高并发场景

高可用：避免锁服务宕机或处理好宕机的补救措施

常见的分布式锁实现方案包括：等

3.Redis实现分布式锁

按照上面的分析，实现分布是锁要满足五点：多进程可见，互斥，可重入，阻塞，高性能，高可用等。我们来看看Redis如何满足这些需求。

3.1.版本1-基本实现

第一次尝试，我们先关注其中必须满足的2个条件：

多进程可见

互斥，锁可释放

多进程可见

首先Redis本身就是基于JVM之外的，因此满足多进程可见的要求。

互斥

互斥就是说只能有一个进程获取锁标记，这个我们可以基于Redis的setnx指令来实现。setnx是set when not exits的意思。当多次执行setnx命令时，只有第一次执行的才会成功并返回1，其它情况返回0：

多个进程来对同一个key执行setnx操作，肯定只有一个能执行成功，其它一定会失败，满足了互斥的需求。

释放锁

释放锁其实只需要把锁的key删除即可，使用del xxx指令。不过，仔细思考，如果在我们执行del之前，服务突然宕机，那么锁岂不是永远无法删除了？！

为了避免因服务宕机引起锁无法释放问题，我们可以在获取锁的时候，给锁加一个有效时间，当时间超出时，就会自动释放锁，这样就不会死锁了。

但时setnx指令没有设置时间的功能，我们要借助于set指令，然后结合set的 NX和PX参数来完成。

其中可以指定这样几个参数：

EX：过期时长，单位是秒PX：过期时长，单位是毫秒NX：等同于setnx

因此，获取和释放锁的基本流程如图：

步骤如下：

1、通过set命令设置锁

2、判断返回结果是否是OK

Nil，获取失败，结束或重试（自旋锁）
OK，获取锁成功

执行业务

释放锁，DEL 删除key即可

3、异常情况，服务宕机。超时时间EX结束，会自动释放锁

3.2.版本2-互斥性

刚才的初级版本中，会有一定的安全问题。

大家思考一下，释放锁就是用DEL语句把锁对应的key给删除，有没有这么一种可能性：

3个进程：A和B和C，在执行任务，并争抢锁，此时A获取了锁，并设置自动过期时间为10s
A开始执行业务，因为某种原因，业务阻塞，耗时超过了10秒，此时锁自动释放了
B恰好此时开始尝试获取锁，因为锁已经自动释放，成功获取锁
A此时业务执行完毕，执行释放锁逻辑（删除key），于是B的锁被释放了，而B其实还在执行业务
此时进程C尝试获取锁，也成功了，因为A把B的锁删除了。

问题出现了：B和C同时获取了锁，违反了互斥性！

如何解决这个问题呢？我们应该在删除锁之前，判断这个锁是否是自己设置的锁，如果不是（例如自己的锁已经超时释放），那么就不要删除了。

那么问题来了：如何得知当前获取锁的是不是自己呢？

我们可以在set 锁时，存入当前线程的唯一标识！删除锁前，判断下里面的值是不是与自己标识释放一致，如果不一致，说明不是自己的锁，就不要删除了。

流程如图：

3.3.版本3-重入性

接下来我们来看看分布式锁的第三个特性，重入性。

如果我们在获取锁以后，执行代码的过程中，再次尝试获取锁，执行setnx肯定会失败，因为锁已经存在了。这样有可能导致死锁，这样的锁就是不可重入的。

如何解决呢？

当然是想办法改造成可重入锁。

3.3.1.重入锁

什么叫做可重入锁呢？

可重入锁可以避免因同一线程中多次获取锁而导致死锁发生。

那么，如何实现可重入锁呢？

获取锁：首先尝试获取锁，如果获取失败，判断这个锁是否是自己的，如果是则允许再次获取，而且必须记录重复获取锁的次数。

释放锁：释放锁不能直接删除了，因为锁是可重入的，如果锁进入了多次，在最内层直接删除锁，导致外部的业务在没有锁的情况下执行，会有安全问题。因此必须获取锁时累计重入的次数，释放时则减去重入次数，如果减到0，则可以删除锁.

因此，存储在锁中的信息就必须包含：key、线程标识、重入次数。不能再使用简单的key-value结构，这里推荐使用hash结构：

key：lock hashKey：线程信息hashValue：重入次数，默认1

3.4.2.流程图

需要用到的一些Redis命令包括：

EXISTS key：判断一个Key是否存在

HEXISTS key field：判断一个hash的field是否存在HSET key field value ：给一个hash的field设置一个值

HINCRBY key field increment：给一个hash的field值增加指定数值EXPIRE key seconds：给一个key设置过期时间

DEL key：删除指定key

具体流程如图：

下面我们假设锁的key为" 获取锁的步骤：

1、判断lock是否存在

"，hashKey是当前线程的id："

"，锁自动释放时间假设为20

存在，说明有人获取锁了，下面判断是不是自己的锁判断当前线程id作为hashKey是否存在：

不存在，说明锁已经有了，且不是自己获取的，锁获取失败，end 存在，说明是自己获取的锁，重入次数+1：

2、不存在，说明可以获取锁，

3、设置锁自动释放时间，释放锁的步骤：

1、判断当前线程id作为hashKey是否存在：

不存在，说明锁已经失效，不用管了

，去到步骤3

存在，说明锁还在，重入次数减1：

2、判断重入次数是否为0：

为0，说明锁全部释放，删除key：

，获取新的重入次数

大于0，说明锁还在使用，重置有效时间：

3.4.Lua脚本

上面探讨的Redis锁实现方案都忽略了一个非常重要的问题：原子性问题。无论是获取锁，还是释放锁的过程，都是有多行Redis指令来完成的，如果不能保证这些Redis命令执行的原子性，则整个过程都是不安全的。

而Redis中支持以Lua脚本来运行多行命令，并且保证整个脚本运行的原子性。接下来，我们分几块来学习Lua脚本的使用：

Redis中如何执行Lua脚本

Lua脚本的基本语法

编写上述分布式锁对应的Lua脚本

3.4.1.Redis中如何执行Lua脚本

与操作Lua相关的命令如下：

其中我们会用到的几个：

直接执行一段脚本，参数包括：

script：脚本内容，或者脚本地址

numkeys：脚本中用到的key的数量，接下来的numkeys个参数会作为key参数，剩下的作为arg参数

key：作为key的参数，会被存入脚本环境中的KEYS数组，角标从1开始

arg：其它参数，会被存入脚本环境中的ARGV数组，角标从1开始

示例：

，其中：

：就是脚本的内容，直接返回字符串，没有别的命令

效果：

将一段脚本编译并缓存起来，生成一个SHA1值并返回，作为脚本字典的key，方便下次使用。参数script就是脚本内容或地址。

以之前案例中的的脚本为例：

此处返回的就是脚本缓存后得到的sha1值。

在脚本字典中，每一个这样的sha1值，对应一段解析好的脚本：

与EVAL类似，执行一段脚本，区别是通过脚本的sha1值，去脚本缓存中查找，然后执行，参数：

sha1：就是脚本对应的sha1值

我们用刚刚缓存的脚本为例：

3.4.2.Lua脚本的基本语法

Lua的详细语法大家可以参考网站上的一些教学，例如： [Lua菜鸟教程]{.underline}，任何语言都是从基本的如：变量、数据类型、循环、逻辑判断、运算、数组等入手。相信熟悉java的你应该可以快速上手Lua。

我们的分布式锁脚本中，主要用到的是对Redis指令的调用，还有些变量声明等。因此我们从这几块入手，看一些简单命令即可：

变量声明

声明一个局部变量，用local关键字即可：

这样的逻辑判断，再加上一

打印结果
条件控制
循环语句：

注意，使用break可以跳出循环。

大家能否利用上述语法编写一个猜数字的小游戏？

提示：可以用来读取一个用户输入的数字

代码示例：

Lua调用Redis指令

当我们再Redis中允许Lua脚本时，有一个内置变量redis，并且具备两个函数：

redis.call("命令名称", 参数1，参数2 ...) : 执行指定的redis命令，执行遇到错误会直接返回错误

redis.pcall("命令名称", 参数1，参数2 ...) : 执行指定的redis命令，执行遇到错误会错误以Lua表的形式返回

例如：

这行Lua脚本的含义就是执行Redis命令：

不过，我们编写脚本时并不希望把set后面的key和value写死，而是可以由调用脚本的人来指定，把key 和value作为参数传入脚本中执行。

还记得redis中的EVAL命令吗？

EVAL执行脚本时可以接受参数，key和arg，并且会用两个内置变量（数组格式）来接受用户传入的key和

arg参数：

KEYS：用来存放key参数

ARGV：用来存放除Key以外的参数

我们在脚本中，可以从数组中根据角标（Lua中数组角标时从1开始）取出用户传入的参数，像这样：

而后，我们在执行脚本时可以动态指定key及需要存放的value值：

3.4.3.编写分布式锁的Lua脚本

接下来，我们就可以将上面的分布式锁思路用Lua脚本来实现了。

1）普通互斥锁

先看版本2的实现：

获取锁：直接使用客户端的set nx ex 命令即可，无需脚本

释放锁：因为要判断锁中的标识是否时自己的，因此需要脚本，如下：

参数的含义说明：

KEYS[1]：就是锁的 key，比如 "lock" ARGV[1]：就是线程的唯一标识，可以时随机字符串

2）可重入锁

首先是获取锁：

然后是释放锁：

3.5.Redis客户端调用Lua

脚本编写完成，还需要通过客户端来调用lua脚本，封装一个获取锁和释放锁的工具。首先我们创建一个工程：

填写信息：

选择依赖：

在配置文件中引入Redis的地址信息：

3.5.1.锁接口

首先定义一个锁接口，定义锁中的方法：

3.5.2.实现类

我们通过Spring提供的RedisTemplate来操作lua脚本，脚本：

中提供了一个方法，用来执行Lua

包含3个参数：

RedisScript<T> script ：封装了Lua脚本的对象

List<K> keys ：脚本中的key的值

Object ... args ：脚本中的参数的值

因此，要执行Lua脚本，我们需要先把脚本封装到象：

通过RedisScript中的静态方法：

对象中，有两种方式来构建对

这个方法接受两个参数：

返回值类型

需要把脚本内容写到代码中，作为参数传递，不够优雅。

自己创建DefaultRedisScript

另一种方式，就是自己去创建

的实现类

的对象：

可以把脚本文件写到classpath下的某个位置，然后通过加载这个文件来获取脚本内容，并设置给实例。

此处我们选择方式二，方便后期对脚本文件的维护。首先在classpath中编写两个Lua脚本文件：然后编写一个新的RedisLock实现：ReentrantRedisLock，利用静态代码块来加载脚本并初始化：

其中，加载脚本文件的代码如下：

然后实现RedisLock接口，实现其中的抽象方法，完整代码如下：

/**

* 存入的线程信息的前缀，防止与其它JVM中线程信息冲突

*/

private final String ID_PREFIX = UUID.randomUUID().toString();

public ReentrantRedisLock(StringRedisTemplate redisTemplate, String key) { this.redisTemplate = redisTemplate;

this.key = key;

}

private static final DefaultRedisScript<Long> LOCK_SCRIPT; private static final DefaultRedisScript<Object> UNLOCK_SCRIPT; static {

// 加载释放锁的脚本

LOCK_SCRIPT = new DefaultRedisScript<>(); LOCK_SCRIPT.setScriptSource(new ResourceScriptSource(new

ClassPathResource("lock.lua"))); LOCK_SCRIPT.setResultType(Long.class);

// 加载释放锁的脚本

UNLOCK_SCRIPT = new DefaultRedisScript<>(); UNLOCK_SCRIPT.setScriptSource(new ResourceScriptSource(new

ClassPathResource("unlock.lua")));

}

// 锁释放时间

private String releaseTime;

@Override

public boolean tryLock(long releaseTime) {

// 记录释放时间

this.releaseTime = String.valueOf(releaseTime);

// 执行脚本

Long result = redisTemplate.execute( LOCK_SCRIPT,

Collections.singletonList(key),

ID_PREFIX + Thread.currentThread().getId(), this.releaseTime);

// 判断结果

return result != null && result.intValue() == 1;

}

@Override

public void unlock() {

// 执行脚本

redisTemplate.execute(

UNLOCK_SCRIPT,

Collections.singletonList(key),

ID_PREFIX + Thread.currentThread().getId(), this.releaseTime);

}

}

3.5.3.获取锁的工厂

定义一个工厂，用来生成锁对象：

3.5.4.测试

我们定义一个定时任务，模拟清理订单的任务：

接下来，我们给任务加锁：

将启动项复制2份（或多分），测试锁是否能生效：

修改第二个启动项的端口，避免冲突

同时启动2个启动项，查看日志：第一个服务：

第二个服务：

可以看到：

在13:39:50秒时，8081服务获取锁失败，而8082服务获取锁成功在13:40:00秒时，8082服务获取锁失败，而8081服务获取锁成功

3.6.Redis锁的其它特性

在一开始介绍分布式锁时，我们聊到分布式锁要满足的一些特性：

多进程可见：多进程可见，否则就无法实现分布式效果

互斥：同一时刻，只能有一个进程获得锁，执行任务后释放锁可重入（可选）：同一个任务再次获取改锁不会被死锁

阻塞锁（可选）：获取失败时，具备重试机制，尝试再次获取锁性能好（可选）：效率高，应对高并发场景

高可用：避免锁服务宕机或处理好宕机的补救措施

目前在Redis中我们已经实现了：多进程可见

互斥

可重入

剩下的几个特性也并非不能满足，例如：

1）阻塞

我们现在的代码中获取锁失败就立即结束，可以修改代码为失败后不断重试，直到某个指定的超时时间后才结束。

// 订阅频道，等待锁被释放通知

countdownlauch while(true){

// 获取锁，如果超过一定时间，break;

}

pubsub 发布订阅，

2）性能好

Redis一向以出色的读写并发能力著称，因此这一点没有问题

3）高可用

单点的redis无法保证高可用，因此一般我们都会给redis搭建主从集群。但是，主从集群无法保证分布式锁的高可用特性。

在Redis官网上，也对这种单点故障做了说明：

因此，Redis的作者又给出了一种新的算法来解决整个高可用问题，即Redlock算法，摘抄了算法的介绍如下：

不过，这种方式并不能完全保证锁的安全性，因为我们给锁设置了自动释放时间，因此某些极端特例下，依然会导致锁的失败，例如下面的情况：

如果Client 1 在持有锁的时候，发生了一次很长时间的FGC 超过了锁的过期时间。锁就被释放了。这个时候Client 2 又获得了一把锁，提交数据。

这个时候Client 1 从FGC 中苏醒过来了，又一次提交数据。冲突发生了

还有一种情况也是因为锁的超时释放问题，例如：

Client 1 从A、B、D、E五个节点中，获取了A、B、C三个节点获取到锁，我们认为他持有了锁这个时候，由于B 的系统时间比别的系统走得快，B就会先于其他两个节点优先释放锁。

Clinet 2 可以从B、D、E三个节点获取到锁。在整个分布式系统就造成两个Client 同时持有锁了。

不过，这种因为时钟偏移造成的问题，我们可以通过延续超时时间、调整系统时间减少时间偏移等方式来解决。Redis作者也对超时问题给出了自己的意见：

简单来说就是在获取锁成功后，监视锁的失效时间，如果即将到期，可以再次去申请续约，延长锁的有效期。

我们可以采用看门狗（watch dog)解决锁超时问题，/开启一个任务，这个任务在获取锁之后10秒后，重新向redis发起请求，重置有效期，重新执行expire

3.7.Redission

如果按照Redlock算法来实现分布式锁，加上各种安全控制，代码会比较复杂。而开源的Redission框架就帮我们实现了各种基于Redis的分布式锁，包括Redlock锁。

Redisson是一个在Redis的基础上实现的Java驻内存数据网格（In-Memory Data Grid）。它不仅提供了一系列的分布式的Java常用对象，还提供了许多分布式服务。其中包括( BitSet , Set , Multimap , SortedSet , Map , List , Queue , BlockingQueue , Deque , BlockingDeque , Semaphore , Lock , AtomicLong , CountDownLatch , Publish / Subscribe , Bloom filter , Remote service , Spring cache , Executor service , Live Object service , Scheduler service ) Redisson提供了使用Redis的最简单和最便捷的方法。Redisson的宗旨是促进使用者对Redis的关注分离（Separation of Concern），从而让使用者能够将精力更集中地放在处理业务逻辑上。

[官网地址]{.underline}： [https://redisson.or]{.underline}g/

[GitHub地址]{.underline}： [https://]{.underline}gi[thub.com/redisson/redisson]{.underline}

看看Redission能实现的功能：

3.7.1.快速入门

1）依赖

使用起来非常方便，首先引入依赖：

2）配置

然后通过Java配置的方式，设置Redis的地址，构建RedissionClient客户端：

3）常用API介绍

RedissClient中定义了常见的锁：

获取锁对象后，可以通过方法获取锁：

有3个重载的方法，可以控制锁是否需要重试来获取：

三个参数：获取锁，设置锁等待时间

、释放时间

，时间单位。

如果获取锁失败后，会在

然获取失败，则认为获取锁失败；

减去获取锁用时的剩余时间段内继续尝试获取锁，如果依

获取锁后，如果超过 leaseTime 未释放，为避免死锁会自动释放。

两个参数：获取锁，设置锁等待时间 time 、时间单位

。释放时间

按照默认的30s

空参：获取锁，

默认0s，即获取锁失败不重试，

默认30s

任务执行完毕，使用方法释放锁：

4）完整案例

使用Redission来代替我们之前自定义锁的测试案例：

代码如下：

3.7.2.Redisson实现细节

首先看空参获取lock的方法：

这里的核心有两部分：

一个是获取锁的方法：tryLockInnerAsync

一个是自动续期（看门狗）的方法：scheduleExpirationRenewal

1）获取锁

首先看tryLockInnerAsync，这个方法是获取锁的方法：

这里的核心就是这段Lua脚本，看看与我们写的是不是基本类似呢，区别是最后返回了这个key的剩余有效期。

2）锁的自动续期

锁如果在执行任务时自动过期，就会引起各种问题，因此我们需要在锁过期前自动申请续期，这个被称为watch dog，看门狗。

刷新时间的代码：

刷新过期时间的代码：

3）带阻塞的获取锁

阻塞获取锁，会在获取失败以后重试，不过会设置失败超时时间。

waitTime：获取锁重试的最大超时时间，默认为0

leaseTime：释放锁的最大时间，默认时30秒unit：时间单位

代码如下：

});

}

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(); 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);

}

}

获取锁失败，会通过Redis的pubsub功能订阅一个频道，如果释放锁会通知自己，然后再重试获取锁。

4）释放锁

释放锁代码基本一致：

下面跟到unlockAsync方法：

然后关键是释放锁的代码：

代码基本一致，就是再最后释放成功后，通过

知锁已经释放，那些再等待的其它线程，就可以获取锁了。

发布了一条消息，通

3.8.总结

总结来看，Redis实现分布式锁，具备下列优缺点：

优点：实现简单，性能好，并发能力强，如果对并发能力有要求，推荐使用

缺点：可靠性有争议，极端情况会出现锁失效问题，如果对安全要求较高，不建议使用

zookeeper实现分布式锁

Zookeeper是一种提供配置管理、分布式协同以及命名的中心化服务。

zk的模型是这样的：zk包含一系列的节点，叫做znode，就好像文件系统一样每个znode表示一个目录，然后znode有一些特性：

有序节点：假如当前有一个父节点为，我们可以在这个父节点下面创建子节点；

zookeeper提供了一个可选的有序特性，例如我们可以创建子节点"/lock/node-"并且指明有序，那么zookeeper在生成子节点时会根据当前的子节点数量自动添加整数序号

也就是说，如果是第一个创建的子节点，那么生成的子节点为，下一个节

点则为，依次类推。

临时节点：客户端可以建立一个临时节点，在会话结束或者会话超时后，zookeeper会自动删除该节点。

事件监听：在读取数据时，我们可以同时对节点设置事件监听，当节点数据或结构变化时，

zookeeper会通知客户端。当前zookeeper有如下四种事件：

节点创建节点删除

节点数据修改子节点变更

基于以上的一些zk的特性，我们很容易得出使用zk实现分布式锁的落地方案：

使用zk的临时节点和有序节点，每个线程获取锁就是在zk创建一个临时有序的节点，比如在/lock/ 目录下。
创建节点成功后，获取/lock目录下的所有临时节点，再判断当前线程创建的节点是否是所有的节点的序号最小的节点
如果当前线程创建的节点是所有节点序号最小的节点，则认为获取锁成功。
如果当前线程创建的节点不是所有节点序号最小的节点，则对节点序号的前一个节点添加一个事件监听。

比如当前线程获取到的节点序号为

,则对

,然后所有的节点列表为

这个节点添加一个事件监听器。

如果锁释放了，会唤醒下一个序号的节点，然后重新执行第3步，判断是否自己的节点序号是最小。

比如释放了，

监听到时间，此时节点集合为

,则为最

小序号节点，获取到锁。

Curator是一个zookeeper的开源客户端，也提供了分布式锁的实现。来看看锁的一些特性Zookeeper是否满足：

互斥：因为只有一个最小节点，满足互斥特性

锁释放：使用Zookeeper可以有效的解决锁无法释放的问题，因为在创建锁的时候，客户端会在ZK 中创建一个临时节点，一旦客户端获取到锁之后突然挂掉（Session连接断开），那么这个临时节点就会自动删除掉。其他客户端就可以再次获得锁。

阻塞锁：使用Zookeeper可以实现阻塞的锁，客户端可以通过在ZK中创建顺序节点，并且在节点上绑定监听器，一旦节点有变化，Zookeeper会通知客户端，客户端可以检查自己创建的节点是不是当前所有节点中序号最小的，如果是，那么自己就获取到锁，便可以执行业务逻辑了。

可重入：使用Zookeeper也可以有效的解决不可重入的问题，客户端在创建节点的时候，把当前客户端的主机信息和线程信息直接写入到节点中，下次想要获取锁的时候和当前最小的节点中的数据比对一下就可以了。如果和自己的信息一样，那么自己直接获取到锁，如果不一样就再创建一个临时的顺序节点，参与排队。

高可用：使用Zookeeper可以有效的解决单点问题，ZK是集群部署的，只要集群中有半数以上的机器存活，就可以对外提供服务。

高性能：Zookeeper集群是满足强一致性的，因此就会牺牲一定的性能，与Redis相比略显不足

总结：

优点：使用非常简单，不用操心释放问题、阻塞获取问题缺点：性能比Redis稍差一些

总结

分布式锁释放方式多种多样，每种方式都有自己的优缺点，我们应该根据业务的具体需求，先择合适的实现。

Redis实现：实现比较简单，性能最高，但是可靠性难以维护

Zookeeper实现：实现最简单，可靠性最高，性能比redis略低