ZooKeeper实现分布式锁的三种方案（有实例），不能错过！ |
您所在的位置：网站首页 › 分布式锁的概念 › ZooKeeper实现分布式锁的三种方案（有实例），不能错过！ |
ZooKeeper实现分布式锁的三种方案（有实例），不能错过！
|
zookeeper客户端选型
1、原生zookeeper客户端，有watcher一次性、无超时重连机制等一系列问题 2、ZkClient，解决了原生客户端一些问题，一些存量老系统中还在使用 3、curator，提供了各种应用场景（封装了分布式锁，计数器等），新项目首选 分布式锁使用场景在单体项目中jvm中的锁即可完成需要，但是微服务、分布式环境下，同一个服务可能部署在多台服务器上，多个jvm之间无法通过常用的jvm锁来完成同步操作，需要借用分布式锁来完成上锁、释放锁。例如在订单服务中，我们需要根据日期来生成订单号流水，就有可能产生相同的时间日期，从而出现重复订单号。 zookeeper分布式锁实现原理1、zookeeper中规定，在同一时刻，不能有多个客户端创建同一个节点，我们可以利用这个特性实现分布式锁。zookeeper临时节点只在session生命周期存在，session一结束会自动销毁。 2、watcher机制，在代表锁资源的节点被删除，即可以触发watcher解除阻塞重新去获取锁，这也是zookeeper分布式锁较其他分布式锁方案的一大优势。 基于临时节点方案第一种方案实现较为简单，逻辑就是谁创建成功该节点，谁就持有锁，创建失败的自己进行阻塞，A线程先持有锁，B线程获取失败就会阻塞，同时对/lockPath设置监听，A线程执行完操作后删除节点，触发监听器，B线程此时解除阻塞，重新去获取锁。
我们模仿原生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线程释放完毕，海量的线程都同时停止阻塞，去争抢锁，这种操作十分耗费资源，且性能大打折扣。 基于临时顺序节点方案临时顺序节点与临时节点不同的是产生的节点是有序的，我们可以利用这一特点，只让当前线程监听上一序号的线程，每次获取锁的时候判断自己的序号是否为最小，最小即获取到锁，执行完毕就删除当前节点继续判断谁为最小序号的节点。
临时顺序节点操作源码 @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 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提供了以下种类的锁： 1、共享可重入锁（Shared Reentrant Lock）：全局同步锁，同一时间不会有两个客户端持有一个锁 2、共享锁：与共享可重入锁类似，但是不可重入（有时候会因为这个原因造成死锁） 3、共享可重入读写锁 4、共享信号量 5、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(); } } |
临时顺序节点 
【本文地址】
今日新闻 |
推荐新闻 |