Zookeeper分布式锁
原创
Zookeeper分布式锁解决方案
zookeeper客户端选型
- 原生zookeeper客户端，有watcher一次性、无超时重连机制等一系列问题
- ZkClient，解决了原生客户端一些问题，一些存量老系统中还在使用
- curator，提供了各种应用场景（封装了分布式锁，计数器等），新项目首选
分布式锁使用场景
在单体项目中jvm中的锁即可完成需要，但是微服务、分布式环境下，同一个服务可能部署在多台服务器上，多个jvm之间无法通过常用的jvm锁来完成同步操作，需要借用分布式锁来完成上锁、释放锁。例如在订单服务中，我们需要根据日期来生成订单号流水，就有可能产生相同的时间日期，从而出现重复订单号。（jdk8使用LocalDateTime线程安全，不会存在这样的问题）
zookeeper分布式锁实现原理
- zookeeper中规定，在同一时刻，不能有多个客户端创建同一个节点，我们可以利用这个特性实现分布式锁。zookeeper临时节点只在session生命周期存在，session一结束会自动销毁。
- watcher机制，在代表锁资源的节点被删除，即可以触发watcher解除阻塞重新去获取锁，这也是zookeeper分布式锁较其他分布式锁方案的一大优势。
基于临时节点方案
第一种方案实现较为简单，逻辑就是谁创建成功该节点，谁就持有锁，创建失败的自己进行阻塞，A线程先持有锁，B线程获取失败就会阻塞，同时对/lockPath设置监听，A线程执行完操作后删除节点，触发监听器，B线程此时解除阻塞，重新去获取锁。
临时节点.png
我们模仿原生jdk的lock接口设计，采用模板方法设计模式来编写分布式锁，这样的好处是扩展性强，我们可以快速切换到redis分布式锁、数据库分布式锁等实现方式。
创建Lock接口
public interface Lock {
/**
* 获取锁
*/
void getLock() throws Exception;
/**
* 释放锁
*/
void unlock() throws Exception;
}AbstractTemplateLock抽象类
public abstract class AbstractTemplateLock implements Lock {
@Override
public void getLock() {
if (tryLock()) {
System.out.println(Thread.currentThread().getName() + "获取锁成功");
} else {
//等待
waitLock();//事件监听 如果节点被删除则可以重新获取
//重新获取
getLock();
}
}
protected abstract void waitLock();
protected abstract boolean tryLock();
protected abstract void releaseLock();
@Override
public void unlock() {
releaseLock();
}
}zookeeper分布式锁逻辑
@Slf4j
public class ZkTemplateLock extends AbstractTemplateLock {
private static final String zkServers = "127.0.0.1:2181";
private static final int sessionTimeout = 8000;
private static final int connectionTimeout = 5000;
private static final String lockPath = "/lockPath";
private ZkClient client;
public ZkTemplateLock() {
client = new ZkClient(zkServers, sessionTimeout, connectionTimeout);
log.info("zk client 连接成功:{}",zkServers);
}
@Override
protected void waitLock() {
CountDownLatch latch = new CountDownLatch(1);
IZkDataListener listener = new IZkDataListener() {
@Override
public void handleDataDeleted(String dataPath) throws Exception {
System.out.println("监听到节点被删除");
latch.countDown();
}
@Override
public void handleDataChange(String dataPath, Object data) throws Exception {}
};
//完成 watcher 注册
client.subscribeDataChanges(lockPath, listener);
//阻塞自己
if (client.exists(lockPath)) {
try {
latch.await();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
//取消watcher注册
client.unsubscribeDataChanges(lockPath, listener);
}
@Override
protected boolean tryLock() {
try {
client.createEphemeral(lockPath);
System.out.println(Thread.currentThread().getName()+"获取到锁");
} catch (Exception e) {
log.error("创建失败");
return false;
}
return true;
}
@Override
public void releaseLock() {
client.delete(this.lockPath);
}
}缺点：
每次去竞争锁，都只会有一个线程拿到锁，当线程数庞大时会发生“惊群”现象，zookeeper节点可能会运行缓慢甚至宕机。这是因为其他线程没获取到锁时都会监听/lockPath节点，当A线程释放完毕，海量的线程都同时停止阻塞，去争抢锁，这种操作十分耗费资源，且性能大打折扣。
基于临时顺序节点方案
临时顺序节点与临时节点不同的是产生的节点是有序的，我们可以利用这一特点，只让当前线程监听上一序号的线程，每次获取锁的时候判断自己的序号是否为最小，最小即获取到锁，执行完毕就删除当前节点继续判断谁为最小序号的节点。
序列化节点.png
临时顺序节点.png
临时顺序节点操作源码
@Slf4j
public class ZkSequenTemplateLock extends AbstractTemplateLock {
private static final String zkServers = "127.0.0.1:2181";
private static final int sessionTimeout = 8000;
private static final int connectionTimeout = 5000;
private static final String lockPath = "/lockPath";
private String beforePath;
private String currentPath;
private ZkClient client;
public ZkSequenTemplateLock() {
client = new ZkClient(zkServers);
if (!client.exists(lockPath)) {
client.createPersistent(lockPath);
}
log.info("zk client 连接成功:{}",zkServers);
}
@Override
protected void waitLock() {
CountDownLatch latch = new CountDownLatch(1);
IZkDataListener listener = new IZkDataListener() {
@Override
public void handleDataDeleted(String dataPath) throws Exception {
System.out.println("监听到节点被删除");
latch.countDown();
}
@Override
public void handleDataChange(String dataPath, Object data) throws Exception {}
};
//给排在前面的节点增加数据删除的watcher，本质是启动另一个线程去监听上一个节点
client.subscribeDataChanges(beforePath, listener);
//阻塞自己
if (client.exists(beforePath)) {
try {
System.out.println("阻塞"+currentPath);
latch.await();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
//取消watcher注册
client.unsubscribeDataChanges(beforePath, listener);
}
@Override
protected boolean tryLock() {
if (currentPath == null) {
//创建一个临时顺序节点
currentPath = client.createEphemeralSequential(lockPath + "/", "lock-data");
System.out.println("current:" + currentPath);
}
//获得所有的子节点并排序。临时节点名称为自增长的字符串
List<String> childrens = client.getChildren(lockPath);
//排序list，按自然顺序排序
Collections.sort(childrens);
if (currentPath.equals(lockPath + "/" + childrens.get(0))) {
return true;
} else {
//如果当前节点不是排第一，则获取前面一个节点信息，赋值给beforePath
int curIndex = childrens.indexOf(currentPath.substring(lockPath.length() + 1));
beforePath = lockPath + "/" + childrens.get(curIndex - 1);
}
System.out.println("beforePath"+beforePath);
return false;
}
@Override
public void releaseLock() {
System.out.println("delete:" + currentPath);
client.delete(currentPath);
}
}Curator分布式锁工具
curator提供了以下种类的锁：
- 共享可重入锁（Shared Reentrant Lock）：全局同步锁，同一时间不会有两个客户端持有一个锁
- 共享锁：与共享可重入锁类似，但是不可重入（有时候会因为这个原因造成死锁）
- 共享可重入读写锁
- 共享信号量
- Multi Shared Lock：管理多种锁的容器实体
我们采用第一种Shared Reentrant Lock中的InterProcessMutex来完成上锁、释放锁的的操作
public class ZkLockWithCuratorTemplate implements Lock {
// zk host地址
private String host = "localhost";
// zk自增存储node
private String lockPath = "/curatorLock";
// 重试休眠时间
private static final int SLEEP_TIME_MS = 1000;
// 最大重试1000次
private static final int MAX_RETRIES = 1000;
//会话超时时间
private static final int SESSION_TIMEOUT = 30 * 1000;
//连接超时时间
private static final int CONNECTION_TIMEOUT = 3 * 1000;
//curator核心操作类
private CuratorFramework curatorFramework;
InterProcessMutex lock;
public ZkLockWithCuratorTemplate() {
curatorFramework = CuratorFrameworkFactory.builder()
.connectString(host)
.connectionTimeoutMs(CONNECTION_TIMEOUT)
.sessionTimeoutMs(SESSION_TIMEOUT)
.retryPolicy(new ExponentialBackoffRetry(SLEEP_TIME_MS, MAX_RETRIES))
.build();
curatorFramework.start();
lock = new InterProcessMutex (curatorFramework, lockPath);
}
@Override
public void getLock() throws Exception {
//5s后超时释放锁
lock.acquire(5, TimeUnit.SECONDS);
}
@Override
public void unlock() throws Exception {
lock.release();
}
}源码以及测试类地址
https://github.com/Motianshi/distribute-tool原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。
如有侵权，请联系 cloudcommunity@tencent.com 删除。
原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。
如有侵权，请联系 cloudcommunity@tencent.com 删除。
评论
登录后参与评论
推荐阅读
目录
相关产品与服务
云服务器
云服务器（Cloud Virtual Machine，CVM）提供安全可靠的弹性计算服务。 您可以实时扩展或缩减计算资源，适应变化的业务需求，并只需按实际使用的资源计费。使用 CVM 可以极大降低您的软硬件采购成本，简化 IT 运维工作。
腾讯云开发者
