郑爽，也能看懂的Zookeeper分布式锁原理
介绍
许多场景中，数据一致性是一个比较重要的话题，在单机环境中，我们可以通过Java提供的并发API来解决；而在分布式环境(会遇到网络故障、消息重复、消息丢失等各种问题)下要复杂得多，比如电商的库存扣减，秒杀活动，集群定时任务执行等需要进程互斥的场景。本文主要探讨如何利用Zookeeper来实现分布式锁，对比了一些其他方案分布式锁的优缺点。至于使用何种，要因自己的业务场景去决定，没有绝对的方案。
分布式锁是什么？
分布式锁是控制分布式系统之间同步访问共享资源的一种方式。
实现分布式锁注意点：
- 锁的可重入性(递归调用不应该被阻塞、避免死锁)
- 锁的超时(避免死锁、死循环等意外情况)
- 锁的阻塞(保证原子性等)
- 锁的特性支持(阻塞锁、可重入锁、公平锁、联锁、信号量、读写锁)
- 分布式锁的开销(分布式锁一般能不用就不用，有些场景可以用乐观锁代替)
- 加锁的粒度(控制加锁的粒度，可以优化系统的性能)
- 加锁的方式
基于数据库表唯一索引 最简单的方式就是直接创建一张锁表，当我们要锁住某个方法或资源时，我们就在该表中增加一条记录，想要释放锁的时候就删除这条记录。给某字段添加唯一性约束，如果有多个请求同时提交到数据库的话，数据库会保证只有一个操作可以成功，那么我们就可以认为操作成功的那个线程获得了该方法的锁，可以执行方法体内容。
但会引入数据库单点、无失效时间、不阻塞、不可重入等问题。
基于数据库排他锁 如果使用的是MySql的InnoDB引擎，在查询语句后面增加for update，数据库会在查询过程中(须通过唯一索引查询)给数据库表增加排他锁，我们可以认为获得排它锁的线程即可获得分布式锁，通过 connection.commit() 操作来释放锁。
会引入数据库单点、不可重入、无法保证一定使用行锁、排他锁，所以有可能长时间不提交导致占用数据库连接等问题。
优缺点
- 优点：
- 缺点
操作数据库需要一定的开销，有一定的性能问题
使用数据库的行级锁并不一定靠谱，尤其是当我们的锁表并不大的时候
缓存
相比较于基于数据库实现分布式锁的方案来说，基于缓存来实现在性能方面会表现的更好一点，目前有很多成熟的缓存产品，包括Redis、memcached、tair等。
基于 redis 的 setnx()、expire() 方法做分布式锁 setnx 的含义就是 SET if Not Exists，其主要有两个参数 setnx(key, value)。该方法是原子的，如果 key 不存在，则设置当前 key 成功，返回 1；如果当前 key 已经存在，则设置当前 key 失败，返回 0。
expire 设置过期时间，要注意的是 setnx 命令不能设置 key 的超时时间，只能通过 expire() 来对 key 设置。
redis有个命令可以实现setnx 和 expire 同样效果的原子性指令，博主这边解释并未提及到。命令：基于 Redlock 做分布式锁 Redlock 是 Redis 的作者 antirez 给出集群模式的 Redis 分布式锁，它基于 N 个完全独立的 Redis 节点（通常情况下 N 可以设置成 5）
set k1 v1 ex 10 nx 。当然为了实现以上博主说的两个命令原子性操作，也可以使用lua脚本完成。
基于 redisson 做分布式锁 redisson 是 redis 官方的分布式锁组件
优缺点
- 优点
- 缺点
通过超时时间来控制锁的失效时间并不是十分的靠谱
Zookeeper
大致思想 每个客户端对某个方法加锁时，在 Zookeeper 上与该方法对应的指定节点目录下，生成一个唯一的临时有序节点（zk有自动生成有序节点的功能）。判断是否获取锁的方式很简单，只需要判断有序节点中序号最小的一个。当释放锁的时候，只需将这个临时节点删除即可。同时，其可以避免服务宕机导致的锁无法释放，而产生的死锁问题
Zookeeper 实现分布式锁的核心原理 1. 排他锁 排他锁，又称写锁或独占锁。如果事务T1对数据对象O1加上了排他锁，那么在整个加锁期间，只允许事务T1对O1进行读取或更新操作，其他事务都不能对这个数据对象进行任何操作，直到T1释放了排他锁。
排他锁核心是保证当前有且仅有一个事务获得锁，并且锁释放之后，所有正在等待获取锁的事务都能够被通知到。
Zookeeper 的强一致性特性，能够很好地保证在分布式高并发情况下，创建节点能够保证全局唯一性，可以利用Zookeeper这个特性，实现排他锁。
就是读写互斥、写写互斥、读读互斥实现原理，有3个核心步骤：定义锁、获取锁、释放锁
- 定义锁
- 获取锁
- 释放锁
正常执行完业务逻辑，客户端主动删除自己创建的临时节点。
2. 共享锁 共享锁，又称读锁。如果事务T1对数据对象O1加上了共享锁，那么当前事务只能对O1进行读取操作，其他事务也只能对这个数据对象加共享锁，直到该数据对象上的所有共享锁都被释放。
共享锁与排他锁的区别在于，加了排他锁之后，数据对象只对当前事务可见，而加了共享锁之后，数据对象对所有事务都可见。
总结，读读共享，读写互斥。都是读请求不锁资源，资源共享。有读有写就锁资源。实现原理也是3个核心步骤：
- 定义锁
- 获取锁
判断获得共享锁的逻辑为：
- 创建完节点后，获取 /lockpath 节点下的所有子节点，并对该节点注册子节点变更的Watcher监听
- 确定自己的节点序号
- 对于读请求，如果比发现有比自己序号小的写请求就等待继续获取锁，没有则获取共享锁，操作资源。 对于写请求，如果发现有比自己序号小的读/写请求就等待获取锁，没有则获取锁。
- 接收到Watcher通知后，重复步骤1
- 释放锁
3. 羊群效应的产生 在实现共享锁的 “判断读写顺序” 的第1个步骤是：创建完节点后，获取 /lockpath 节点下的所有子节点，并对该节点注册子节点变更的Watcher监听。这样的话，任何一次客户端移除共享锁之后，Zookeeper将会发送子节点变更的Watcher通知给所有机器，系统中将有大量的 “Watcher通知” 和 “子节点列表获取” 这个操作重复执行，然后所有节点再判断自己是否是序号最小的节点(写请求)或者判断比自己序号小的子节点是否都是读请求(读请求)，从而继续等待下一次通知。
然而，这些重复操作很多都是 “无用的”，实际上每个锁竞争者只需要关注序号比自己小的那个节点是否存在即可。
当集群规模比较大时，这些 “无用的” 操作不仅会对Zookeeper造成巨大的性能影响和网络冲击，更为严重的是，如果同一时间有多个客户端释放了共享锁，Zookeeper服务器就会在短时间内向其余客户端发送大量的事件通知–这就是所谓的 “羊群效应“。
改进后的分布式锁实现：
- 客户端调用 create 方法创建一个类似于 /lockpath/[hostname]-请求类型-序号 的临时顺序节点
- 客户端调用 getChildren 方法获取所有已经创建的子节点列表(这里不注册任何Watcher)
- 读请求：向比自己序号小的最后一个写请求节点注册Watcher监听，写请求：向比自己序号小的最后一个节点注册Watcher监听 （有无获取锁的逻辑同上）
- 等待Watcher监听，继续进入步骤2
- Shared Reentrant Lock 可重入锁
- Shared Lock 共享不可重入锁
- Shared Reentrant Read Write Lock 可重入读写锁
- Shared Semaphore 信号量
- Multi Shared Lock 多锁
关于错误处理：还是强烈推荐使用ConnectionStateListener处理连接状态的改变。当连接LOST时你不再拥有锁。1. 可重入锁 Shared Reentrant Lock，全局可重入锁，所有客户端都可以请求，同一个客户端在拥有锁的同时，可以多次获取，不会被阻塞。它是由类 InterProcessMutex 来实现，它的主要方法：
//构造方法
public InterProcessMutex(CuratorFramework client, String path)
public InterProcessMutex(CuratorFramework client, String path, LockInternalsDriver driver)
// 通过acquire获得锁,并提供超时机制：
public void acquire() throws Exception
public boolean acquire(long time, TimeUnit unit) throws Exception
// 撤销锁
public void makeRevocable(RevocationListener listener)
public void makeRevocable(final RevocationListener listener, Executor executor) 定义一个 FakeLimitedResource 类来模拟共享资源，该资源一次只能被一个线程使用，直到使用结束，下一个线程才能使用，否则会抛出异常。
public class FakeLimitedResource {
private final AtomicBoolean inUse = new AtomicBoolean(false);
// 模拟只能单线程操作的资源
public void use() throws InterruptedException {
if (!inUse.compareAndSet(false, true)) {
// 在正确使用锁的情况下，此异常不可能抛出
throw new IllegalStateException("Needs to be used by one client at a time");
}
try {
Thread.sleep((long) (100 * Math.random()));
} finally {
inUse.set(false);
}
}
}下面的代码将创建 N 个线程来模拟分布式系统中的节点，系统将通过 InterProcessMutex 来控制对资源的同步使用。
每个节点都将发起10次请求，完成 请求锁--访问资源--再次请求锁--释放锁--释放锁 的过程。
客户端通过 acquire 请求锁，通过 release 释放锁，获得几把锁就要释放几把锁。
这个共享资源一次只能被一个线程使用，如果控制同步失败，将抛异常。
public class SharedReentrantLockTest {
private static final String lockPath = "/testZK/sharedreentrantlock";
private static final Integer clientNums = 5;
final static FakeLimitedResource resource = new FakeLimitedResource();
// 共享的资源
private static CountDownLatch countDownLatch = new CountDownLatch(clientNums);
public static void main(String[] args) throws InterruptedException {
for (int i = 0;
i < clientNums;
i++) {
String clientName = "client#" + i;
new Thread(new Runnable() {
@Override
public void run() {
CuratorFramework client = ZKUtils.getClient();
client.start();
Random random = new Random();
try {
final InterProcessMutex lock = new InterProcessMutex(client, lockPath);
// 每个客户端请求10次共享资源
for (int j = 0;
j < 10;
j++) {
if (!lock.acquire(10, TimeUnit.SECONDS)) {
throw new IllegalStateException(j + ". " + clientName + " 不能得到互斥锁");
}
try {
System.out.println(j + ". " + clientName + " 已获取到互斥锁");
resource.use();
// 使用资源
if (!lock.acquire(10, TimeUnit.SECONDS)) {
throw new IllegalStateException(j + ". " + clientName + " 不能再次得到互斥锁");
}
System.out.println(j + ". " + clientName + " 已再次获取到互斥锁");
lock.release();
// 申请几次锁就要释放几次锁
} finally {
System.out.println(j + ". " + clientName + " 释放互斥锁");
lock.release();
// 总是在finally中释放
}
Thread.sleep(random.nextInt(100));
}
} catch (Throwable e) {
System.out.println(e.getMessage());
} finally {
CloseableUtils.closeQuietly(client);
System.out.println(clientName + " 客户端关闭！");
countDownLatch.countDown();
}
}
}).start();
}
countDownLatch.await();
System.out.println("结束！");
}
}控制台打印日志，可以看到对资源的同步访问控制成功，并且锁是可重入的
0. client#3 已获取到互斥锁
0. client#3 已再次获取到互斥锁
0. client#3 释放互斥锁
0. client#1 已获取到互斥锁
0. client#1 已再次获取到互斥锁
0. client#1 释放互斥锁
0. client#2 已获取到互斥锁
0. client#2 已再次获取到互斥锁
0. client#2 释放互斥锁
0. client#0 已获取到互斥锁
0. client#0 已再次获取到互斥锁
0. client#0 释放互斥锁
0. client#4 已获取到互斥锁
0. client#4 已再次获取到互斥锁
0. client#4 释放互斥锁
1. client#1 已获取到互斥锁
1. client#1 已再次获取到互斥锁
1. client#1 释放互斥锁
2. client#1 已获取到互斥锁
2. client#1 已再次获取到互斥锁
2. client#1 释放互斥锁
1. client#4 已获取到互斥锁
1. client#4 已再次获取到互斥锁
1. client#4 释放互斥锁
1. client#3 已获取到互斥锁
1. client#3 已再次获取到互斥锁
1. client#3 释放互斥锁
1. client#2 已获取到互斥锁
1. client#2 已再次获取到互斥锁
1. client#2 释放互斥锁
2. client#4 已获取到互斥锁
2. client#4 已再次获取到互斥锁
2. client#4 释放互斥锁
....
....
client#2 客户端关闭！
9. client#0 已获取到互斥锁
9. client#0 已再次获取到互斥锁
9. client#0 释放互斥锁
9. client#3 已获取到互斥锁
9. client#3 已再次获取到互斥锁
9. client#3 释放互斥锁
client#0 客户端关闭！
8. client#4 已获取到互斥锁
8. client#4 已再次获取到互斥锁
8. client#4 释放互斥锁
9. client#4 已获取到互斥锁
9. client#4 已再次获取到互斥锁
9. client#4 释放互斥锁
client#3 客户端关闭！
client#4 客户端关闭！
结束！同时在程序运行期间查看Zookeeper节点树，可以发现每一次请求的锁实际上对应一个临时顺序节点
[zk: localhost:2181(CONNECTED) 42] ls /testZK/sharedreentrantlock
[leases, _c_208d461b-716d-43ea-ac94-1d2be1206db3-lock-0000001659, locks, _c_64b19dba-3efa-46a6-9344-19a52e9e424f-lock-0000001658, _c_cee02916-d7d5-4186-8867-f921210b8815-lock-0000001657]2. 不可重入锁 Shared Lock 与 Shared Reentrant Lock 相似，但是不可重入，这个不可重入锁由类 InterProcessSemaphoreMutex 来实现，使用方法和上面的类似。
将上面程序中的 InterProcessMutex 换成不可重入锁 InterProcessSemaphoreMutex，如果再运行上面的代码，结果就会发现线程被阻塞在第二个 acquire 上，直到超时，也就是此锁不是可重入的。控制台输出日志如下：
0. client#2 已获取到互斥锁
0. client#1 不能得到互斥锁
0. client#4 不能得到互斥锁
0. client#0 不能得到互斥锁
0. client#3 不能得到互斥锁
client#1 客户端关闭！
client#4 客户端关闭！
client#3 客户端关闭！
client#0 客户端关闭！
0. client#2 释放互斥锁
0. client#2 不能再次得到互斥锁
client#2 客户端关闭！
结束！把第二个获取锁的代码注释，程序才能正常执行
0. client#1 已获取到互斥锁
0. client#1 释放互斥锁
0. client#2 已获取到互斥锁
0. client#2 释放互斥锁
0. client#0 已获取到互斥锁
0. client#0 释放互斥锁
0. client#4 已获取到互斥锁
0. client#4 释放互斥锁
0. client#3 已获取到互斥锁
0. client#3 释放互斥锁
1. client#1 已获取到互斥锁
1. client#1 释放互斥锁
1. client#2 已获取到互斥锁
1. client#2 释放互斥锁
....
....
9. client#4 已获取到互斥锁
9. client#4 释放互斥锁
9. client#0 已获取到互斥锁
client#2 客户端关闭！
9. client#0 释放互斥锁
9. client#1 已获取到互斥锁
client#0 客户端关闭！
client#4 客户端关闭！
9. client#1 释放互斥锁
9. client#3 已获取到互斥锁
client#1 客户端关闭！
9. client#3 释放互斥锁
client#3 客户端关闭！
结束！3. 可重入读写锁 Shared Reentrant Read Write Lock，可重入读写锁，一个读写锁管理一对相关锁，一个负责读操作，另外一个负责写操作；读操作在写锁没被使用时可同时由多个进程使用，而写锁在使用时不允许读(阻塞)；此锁是可重入的；一个拥有写锁的线程可重入读锁，但是读锁却不能进入写锁，这也意味着写锁可以降级成读锁， 比如 请求写锁 --->读锁 ---->释放写锁；从读锁升级成写锁是不行的。
可重入读写锁主要由两个类实现：InterProcessReadWriteLock、InterProcessMutex，使用时首先创建一个 InterProcessReadWriteLock 实例，然后再根据你的需求得到读锁或者写锁，读写锁的类型是 InterProcessMutex。
可以理解为我们上面分析的共享锁
public static void main(String[] args) throws InterruptedException {
for (int i = 0;
i < clientNums;
i++) {
final String clientName = "client#" + i;
new Thread(new Runnable() {
@Override
public void run() {
CuratorFramework client = ZKUtils.getClient();
client.start();
final InterProcessReadWriteLock lock = new InterProcessReadWriteLock(client, lockPath);
final InterProcessMutex readLock = lock.readLock();
final InterProcessMutex writeLock = lock.writeLock();
try {
// 注意只能先得到写锁再得到读锁，不能反过来！！！
if (!writeLock.acquire(10, TimeUnit.SECONDS)) {
throw new IllegalStateException(clientName + " 不能得到写锁");
}
System.out.println(clientName + " 已得到写锁");
if (!readLock.acquire(10, TimeUnit.SECONDS)) {
throw new IllegalStateException(clientName + " 不能得到读锁");
}
System.out.println(clientName + " 已得到读锁");
try {
resource.use();
// 使用资源
} finally {
System.out.println(clientName + " 释放读写锁");
readLock.release();
writeLock.release();
}
} catch (Exception e) {
System.out.println(e.getMessage());
} finally {
CloseableUtils.closeQuietly(client);
countDownLatch.countDown();
}
}
}).start();
}
countDownLatch.await();
System.out.println("结束！");
}
}控制台打印日志
client#1 已得到写锁
client#1 已得到读锁
client#1 释放读写锁
client#2 已得到写锁
client#2 已得到读锁
client#2 释放读写锁
client#0 已得到写锁
client#0 已得到读锁
client#0 释放读写锁
client#4 已得到写锁
client#4 已得到读锁
client#4 释放读写锁
client#3 已得到写锁
client#3 已得到读锁
client#3 释放读写锁
结束！4. 信号量 Shared Semaphore，一个计数的信号量类似JDK的 Semaphore，JDK中 Semaphore 维护的一组许可(permits)，而Cubator中称之为租约(Lease)。
有两种方式可以决定 semaphore 的最大租约数，在构造方法初始化的时候完成。第一种方式是由用户给定maxLeases 决定，第二种方式使用 SharedCountReader 类。
public InterProcessSemaphoreV2(CuratorFramework client, String path, int maxLeases)
public InterProcessSemaphoreV2(CuratorFramework client, String path, SharedCountReader count)信号量主要实现类有
InterProcessSemaphoreV2 - 信号量实现类
Lease - 租约(单个信号)
SharedCountReader - 计数器，用于计算最大租约数量你可以请求多个租约，通过调用acquire方法完成。如果 Semaphore 当前的租约不够，则请求线程会被阻塞，同时还提供了超时的重载方法。
public Lease acquire() throws Exception
public Collection acquire(int qty) throws Exception
public Lease acquire(long time, TimeUnit unit) throws Exception
public Collection acquire(int qty, long time, TimeUnit unit) throws Exception 调用 acquire 会返回一个租约对象，客户端必须在 finally 中 close 这些租约对象，否则这些租约会丢失掉。但是，如果客户端session由于某种原因比如crash丢掉，那么这些客户端持有的租约会自动close，这样其它客户端可以继续使用这些租约。租约还可以通过下面的几个方法返还，可以返回一个，也可以返回多个。
public void returnLease(Lease lease)
public void returnAll(Collection leases) 一个Demo程序如下
public class SharedSemaphoreTest {
private static final int MAX_LEASE = 10;
private static final String PATH = "/testZK/semaphore";
private static final FakeLimitedResource resource = new FakeLimitedResource();
public static void main(String[] args) throws Exception {
CuratorFramework client = ZKUtils.getClient();
client.start();
InterProcessSemaphoreV2 semaphore = new InterProcessSemaphoreV2(client, PATH, MAX_LEASE);
Collection leases = semaphore.acquire(5);
System.out.println("获取租约数量：" + leases.size());
Lease lease = semaphore.acquire();
System.out.println("获取单个租约");
resource.use();
// 使用资源
// 再次申请获取5个leases，此时leases数量只剩4个，不够，将超时
Collection leases2 = semaphore.acquire(5, 10, TimeUnit.SECONDS);
System.out.println("获取租约，如果超时将为null：" + leases2);
System.out.println("释放租约");
semaphore.returnLease(lease);
// 再次申请获取5个，这次刚好够
leases2 = semaphore.acquire(5, 10, TimeUnit.SECONDS);
System.out.println("获取租约，如果超时将为null：" + leases2);
System.out.println("释放集合中的所有租约");
semaphore.returnAll(leases);
semaphore.returnAll(leases2);
client.close();
System.out.println("结束!");
}
} 控制台打印日志
获取租约数量：5
获取单个租约
获取租约，如果超时将为null：null
释放租约
获取租约，如果超时将为null：[org.apache.curator.framework.recipes.locks.InterProcessSemaphoreV2$3@3108bc, org.apache.curator.framework.recipes.locks.InterProcessSemaphoreV2$3@370736d9, org.apache.curator.framework.recipes.locks.InterProcessSemaphoreV2$3@5f9d02cb, org.apache.curator.framework.recipes.locks.InterProcessSemaphoreV2$3@63753b6d, org.apache.curator.framework.recipes.locks.InterProcessSemaphoreV2$3@6b09bb57]
释放集合中的所有租约
结束!上面所讲的4种锁都是公平锁(fair)。从ZooKeeper的角度看，每个客户端都按照请求的顺序获得锁，相当公平。5. 多锁 Multi Shared Lock 是一个锁的容器。当调用 acquire，所有的锁都会被 acquire，如果请求失败，所有的锁都会被 release。同样调用 release 时所有的锁都被 release(失败被忽略)。基本上，它就是组锁的代表，在它上面的请求释放操作都会传递给它包含的所有的锁。
主要涉及两个类：
InterProcessMultiLock - 对所对象实现类
InterProcessLock - 分布式锁接口类它的构造函数需要包含锁的集合，或者一组 ZooKeeper 的 path，用法和 Shared Lock 相
public InterProcessMultiLock(CuratorFramework client, List paths)
public InterProcessMultiLock(List locks) 一个Demo程序如下
public class MultiSharedLockTest {
private static final String lockPath1 = "/testZK/MSLock1";
private static final String lockPath2 = "/testZK/MSLock2";
private static final FakeLimitedResource resource = new FakeLimitedResource();
public static void main(String[] args) throws Exception {
CuratorFramework client = ZKUtils.getClient();
client.start();
InterProcessLock lock1 = new InterProcessMutex(client, lockPath1);
// 可重入锁
InterProcessLock lock2 = new InterProcessSemaphoreMutex(client, lockPath2);
// 不可重入锁
// 组锁，多锁
InterProcessMultiLock lock = new InterProcessMultiLock(Arrays.asList(lock1, lock2));
if (!lock.acquire(10, TimeUnit.SECONDS)) {
throw new IllegalStateException("不能获取多锁");
}
System.out.println("已获取多锁");
System.out.println("是否有第一个锁: " + lock1.isAcquiredInThisProcess());
System.out.println("是否有第二个锁: " + lock2.isAcquiredInThisProcess());
try {
resource.use();
// 资源操作
} finally {
System.out.println("释放多个锁");
lock.release();
// 释放多锁
}
System.out.println("是否有第一个锁: " + lock1.isAcquiredInThisProcess());
System.out.println("是否有第二个锁: " + lock2.isAcquiredInThisProcess());
client.close();
System.out.println("结束!");
}
}读完是否有疑问？不妨留个言一起探讨，探讨！！ 整理自互联网，未知出处，文章略有改动。如有侵权请及时联系哦！
推荐阅读
- 数据分析也能Freestyle | 不一样的Smartbi Insight
- 现在，我终于敢承认我的爱豆是郑爽了
- 华语电影大师王家卫（台词艺术也是我“装逼”的资本）
- A股也没这么精彩，深度解读新三板十大并购案
- 柔性关系
- Redis|17 为什么 CPU 结构也会影响 Redis的性能（）
- 我是孩子妈，也是妈妈的孩子
- 科技类|“元宇宙”既是机遇也是挑战
- 编程语言|Node 之父斥责 Oracle（你们也不用，那请交出 JavaScript 商标！）
- 郑州也有动漫展你知道吗（）