分布式锁三 ZK分布式锁
概述
基于zookeeper临时有序节点可以实现的分布式锁,大致思想即为：每个客户端对某个方法加锁时，在zookeeper上的与该方法对应的指定节点的目录下，生成一个唯一的瞬时有序节点。 判断是否获取锁的方式很简单，只需要判断有序节点中序号最小的一个。 当释放锁的时候，只需将这个瞬时节点删除即可。
ZK排他锁
排他锁的核心是如何保证当前有且仅有一个事务获得锁，并且锁被释放后，所有正在等待获取锁的事务都能够被通知到.
获取锁
- ZooKeeper 是通过创建数据节点（ZNode）来表示一个锁。假设我们有一个
/exclusive_lock临时节点，在需要创建排他锁时，所有的客户端都会试图调用create()接口在/exclusive_lock下创建一个临时节点/lock。zk会保证在所有的客户端中，最终只有一个能成功创建，那么就可以认为该客户端获取了锁。同时，所有没有获取到锁的客户端就需要在/exclusive_lock节点上创建一个Watcher监听，以便随时监听到/lock节点变更情况。
释放锁
在定义锁的时候，我们提到了
/exclusive_lock是一个临时节点，因此在两种情况下，可能释放锁- 当前获取锁的客户端机器宕机，那么ZooKeeper上的这个临时节点就会被移除
- 正常执行完业务逻辑后，客户端就会主动将自己创建的临时节点删除
无论哪种情况移除了
/lock节点，zk都会通知所有在/exclusive_lock节点上注册了子节点变更Watcher监听的客户端。这些客户端在接收到通知后，再次重新发起分布式锁获取，即重复“获取锁”过程
ZK共享锁
同排他锁一样，同样是通过ZooKeeper上的数据节点来表示一个锁，是一个类似于/shared_lock/[Hostname]-请求类型-序号的临时顺序节点，如： /shared_lock/192.168.0.1-R-00000001，那么 这个节点就代表了一个共享锁.(R-读 W-写)
获取锁
- 在需要获取共享锁时，所有客户端都会到/shared_lock这个节点下面创建一个临时顺序节点，如果当前是读请求，那么就创建例如 /shared_lock/192.168.0.1-R-00000001的节点。如果是写请求，那么就创建例如 /shared_lock/192.168.0.1-W-00000001的节点。
判断读写顺序
- 根据共享锁的定义，不同的事务都可以同时对同一个数据对象进行读操作，而更新操作必须在当前没有任何事务读写操作的情况下进行。基于这个原则来通过ZooKeeper的节点来确定分布式读写顺序，大概可以分为如下4个步骤：
创建完节点后，获取/shared_lock节点下的所有子节点，并对该节点注册了子节点变更的Wathcher监听
确定自己的节点序号在所有子节点中的顺序
对于读请求： 如果没有比自己序号小的子节点，或是所有比自己序号小的子节点都是读请求，那么表明 自己已经成功获取到了共享锁，同时开始执行读取逻辑；如果比自己序号小的子节点中有写请求，那么就需要进入等待； 对于写请求： 如果自己不是序号最小的子节点，那么就需要进入等待。
接收到Watcher的通知后，重复步骤1。
释放锁
- 释放锁的逻辑和排他锁是一致.
实现思路
zookeeper 可以根据有序节点+watch实现，实现思路：
- 为每个线程生成一个有序的临时节点，为确保有序性，在排序一次全部节点，获取全部节点，每个线程判断自己是否最小，如果是的话，获得锁，执行操作，操作完删除自身节点。如果不是第一个的节点则监听它的前一个节点，当它的前一个节点被删除时，则它会获得锁，以此类推。
由于自己实现麻烦，且apache提供了zk开源客户端Curator包解决此方案。Curator框架在zookeeper原生API接口上进行了包装，解决了很多ZooKeeper客户端非常底层的细节开发。提供ZooKeeper各种应用场景(recipe， 比如：分布式锁服务、集群领导选举、共享计数器、缓存机制、分布式队列等)的抽象封装，实现了Fluent风格的API接口,是最好用，最流行的zookeeper的客户端。
代码片段
- 引入依赖
<dependency>
<groupId>org.apache.curator</groupId>
<artifactId>curator-recipes</artifactId>
<version>4.0.1</version>
<exclusions>
<exclusion>
<groupId>org.apache.zookeeper</groupId>
<artifactId>zookeeper</artifactId>
</exclusion>
</exclusions>
</dependency>
<!-- curator中有zookeeper依赖，排除掉，引用版本和下载的客户端版本一致的jar -->
<dependency>
<groupId>org.apache.zookeeper</groupId>
<artifactId>zookeeper</artifactId>
<version>3.4.14</version>
</dependency>
- DistributedRedisLock 类
public class DistributedRedisLock {
public static Logger log = LoggerFactory.getLogger(DistributedRedisLock.class);
/**
* 可重入排它锁
**/
private InterProcessMutex interProcessMutex;
/**
* 竞争资源标志
**/
private String lockName;
/**
* 根节点
**/
private String root = "/distributed/lock/";
private static CuratorFramework curatorFramework;
/**
* 多ZK_URL 逗号分隔
**/
private static String ZK_URL = "127.0.0.1:2181";
static {
curatorFramework = CuratorFrameworkFactory.newClient(ZK_URL, new ExponentialBackoffRetry(3000, 3));
curatorFramework.start();
}
/**
* 实例化
*
* @param lockName
*/
public DistributedRedisLock(String lockName) {
try {
this.lockName = lockName;
interProcessMutex = new InterProcessMutex(curatorFramework, root + lockName);
} catch (Exception e) {
log.error("initial InterProcessMutex exception=" + e);
}
}
/**
* 获取锁
*/
public boolean acquire() {
int flag = 0;
try {
//重试2次，每次最大等待2s，也就是最大等待4s
while (!interProcessMutex.acquire(2, TimeUnit.SECONDS)) {
flag++;
if (flag > 1) { //重试两次
break;
}
}
} catch (Exception e) {
log.error("distributed lock acquire exception=" + e);
}
if (flag > 1) {
log.debug("Thread:" + Thread.currentThread().getId() + " acquire distributed lock busy");
return Boolean.FALSE;
} else {
log.debug("Thread:" + Thread.currentThread().getId() + " acquire distributed lock success");
return Boolean.TRUE;
}
}
/**
* 释放锁
*/
public void release() {
try {
if (interProcessMutex != null && interProcessMutex.isAcquiredInThisProcess()) {
interProcessMutex.release();
curatorFramework.delete().inBackground().forPath(root + lockName);
log.debug("Thread:" + Thread.currentThread().getId() + " release distributed lock success");
}
} catch (Exception e) {
log.debug("Thread:" + Thread.currentThread().getId() + " release distributed lock exception=" + e);
}
}
}
- ZkLockTemplate 类和之前类似
public class ZkLockTemplateImpl implements ZkLockTemplate {
private Logger LOG = LoggerFactory.getLogger(ZkLockTemplateImpl.class);
@Override
public <T> T execute(String lockName, CallBack<T> callBack) {
DistributedRedisLock lock = new DistributedRedisLock(lockName);
boolean acquire = false;
try {
acquire = lock.acquire();
if (acquire){
LOG.debug("ZkLockTemplateImpl,获取到锁");
return callBack.invoke();
}
LOG.info("ZkLockTemplateImpl,未获取到锁");
throw new AppException("1011", lockName + "业务正在处理中");
} finally {
if (acquire) {
lock.release();
}
}
}
}
测试
- 先来一个不加分布式锁的模拟
public class DemoTest {
public static void say(CountDownLatch down) {
try {
down.await();
} catch (InterruptedException e) {
e.printStackTrace();
}
SimpleDateFormat sdf = new SimpleDateFormat("HH:mm:ss|SSS");
String orderNo = sdf.format(new Date());
System.out.println("生成的订单号是：" + orderNo);
}
public static void main(String[] args) {
CountDownLatch down = new CountDownLatch(1);
for (int i = 0; i < 10; i++) {
new Thread(()-> {
DemoTest.say(down);
}).start();
}
down.countDown();
}
}
- 结果：
生成的订单号是：19:39:41|168
生成的订单号是：19:39:41|169
生成的订单号是：19:39:41|168
生成的订单号是：19:39:41|168
生成的订单号是：19:39:41|169
生成的订单号是：19:39:41|168
生成的订单号是：19:39:41|169
生成的订单号是：19:39:41|170
生成的订单号是：19:39:41|170
生成的订单号是：19:39:41|170
可以看出时间是会乱的
- zk分布式锁改造版本
public class DemoTest {
public static void say(CountDownLatch down) {
try {
down.await();
} catch (InterruptedException e) {
e.printStackTrace();
}
SimpleDateFormat sdf = new SimpleDateFormat("HH:mm:ss|SSS");
String orderNo = sdf.format(new Date());
System.out.println("生成的订单号是：" + orderNo);
}
public static void main(String[] args) {
ZkLockTemplate template = new ZkLockTemplateImpl();
CountDownLatch down = new CountDownLatch(1);
for (int i = 0; i < 10; i++) {
new Thread(()-> {
template.execute("createOrder",()->{
DemoTest.say(down);
return null;
});
}).start();
}
down.countDown();
}
}
- 结果：
生成的订单号是：19:42:13|592
生成的订单号是：19:42:13|655
生成的订单号是：19:42:13|670
生成的订单号是：19:42:13|670
生成的订单号是：19:42:13|686
生成的订单号是：19:42:13|702
生成的订单号是：19:42:13|717
生成的订单号是：19:42:13|717
生成的订单号是：19:42:13|733
生成的订单号是：19:42:13|733
改造后的打印按照时间顺序打印