Zookeeper分布式锁实现

        zookeeper最初设计的初衷就是为了保证分布式系统的一致性。本文将讲解如何利用zookeeper的临时顺序结点，实现分布式锁。

目录

1. 理论分析

        1.1 结点类型

        1.2 监听器

        1.3 实现原理

2. 手写实现简易zookeeper分布式锁

        1.1 依赖

         1.2 常量定义

        1.3 实现zookeeper分布式锁

        1.4 使用方式

3. 引入Curator框架实现zookeeper分布式锁

        2.1 框架依赖

        2.2 使用方式

---

1. 理论分析

        zookeeper 和Linux一样，采用目录树的方式管理结点，目录层级间以 / 区分

        每个数据节点在 ZooKeeper 中被称为 znode，它是 ZooKeeper 中数据的最小单元。由于ZooKeeper 主要用于协调服务，出于性能和一致性考虑，每个节点的存放数据上限为1M

     

        1.1 结点类型

        znode有四种类型：

        1.持久化结点  (PERSISTENT）: 创建节点后一直存在

        2. 持久化有序结点（PERSISTENT_SEQUENTIAL）：在持久化结点的基础上，zookeeper会自动根据创建顺序，在结点名称后面加上一串序号

        3. 临时结点（EPHEMERAL）：在zookeeper与客户端失去连接后自动删除

        4. 临时有序结点（EPHEMERAL_SEQUENTIAL）：在临时结点的基础上，zookeeper会自动根据创建顺序，在结点名称后面加上一串序号

        1.2 监听器

                Watcher 监听机制是 Zookeeper 中非常重要的特性。结点可以绑定监听事件，当监听事件发生的时候,Zookeeper会向客户端发送通知事件，执行监听器的回调方法。

        

        1.3 实现原理

        我们首先新建一个"/locks"的持久化结点,用来管理表示锁的子节点。（实际场景使用可以根据不同锁对象划分成更细致的持久化结点,比如"/locks/bilibili/comment/publish"）

        当用户尝试获取锁的时候，在"locks"结点下新建一个临时有序结点,例如"seq-00001"

        新建结点成功后，系统进行检查，建立的结点是否是当前所有子节点中序号最小的一个

        如果是最小的一个，说明用户是当前锁的持有者，往下执行业务逻辑，执行完成后摧毁临时结点

        ​​​​​​​

        如果不是最小的一个，为了避免不断地自旋检查空耗性能，一般采用注册监听器的方式减少性能消耗：监听前一个结点的摧毁事件。如果用户持有的结点前面还有其他结点，说明用户不是持有的人，不能执行业务逻辑，应当阻塞等待；直到用户前一个结点被摧毁，说明轮到用户持有锁了，可以继续往下执行业务逻辑。

        

2. 手写实现简易zookeeper分布式锁         1.1 依赖 <dependency> <groupId>junit</groupId> <artifactId>junit</artifactId> <version>RELEASE</version> <scope>test</scope> </dependency> <!--日志--> <dependency> <groupId>org.apache.logging.log4j</groupId> <artifactId>log4j-core</artifactId> <version>2.8.2</version> </dependency> <!--zookeeper--> <dependency> <groupId>org.apache.zookeeper</groupId> <artifactId>zookeeper</artifactId> <version>3.5.6</version> </dependency> <dependency> <groupId>org.junit.jupiter</groupId> <artifactId>junit-jupiter</artifactId> <version>RELEASE</version> <scope>compile</scope> </dependency>          1.2 常量定义 public interface ZkConstants { //连接地址 String connectString = "127.0.0.1:2181"; // 连接超时时间 int sessionTimeout = 2000; }         1.3 实现zookeeper分布式锁 public class DistributedLock { // zk客户端连接 private ZooKeeper zkClient; // 连接成功等待 private CountDownLatch connectLatch = new CountDownLatch(1); // 前一个结点（锁） private String waitPath; // 结点删除等待 private CountDownLatch waitLatch = new CountDownLatch(1); // 当前创建的结点（锁） private String createNode; /** * 构造方法:初始化客户端连接 * * @throws IOException * @throws InterruptedException * @throws KeeperException */ public DistributedLock() throws IOException, InterruptedException, KeeperException { //获取连接 zkClient = new ZooKeeper(ZkConstants.connectString, ZkConstants.sessionTimeout, new Watcher() { @Override public void process(WatchedEvent watchedEvent) { //连接成功，释放countDownLatch if (watchedEvent.getState() == Event.KeeperState.SyncConnected) { connectLatch.countDown(); } //前一个结点删除 if (watchedEvent.getType() == Event.EventType.NodeDeleted && watchedEvent.getPath().equals(waitPath)) { //解锁下一个结点 waitLatch.countDown(); } } }); //等待zk正常连接后，再往下执行 connectLatch.await(); //判断根节点/locks是否存在 Stat exists = zkClient.exists("/locks", false); if (exists == null) { //创建根节点 -- 持久结点 zkClient.create("/locks", "locks".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); } } /** * 加锁 * * @throws InterruptedException * @throws KeeperException */ public void zkLock() throws InterruptedException, KeeperException { //创建对应的临时带序号结点 createNode = zkClient.create("/locks/" + "seq-", null, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL); //判断创建节点是否是序号最小的结点 List<String> children = zkClient.getChildren("/locks", false); if (children.size() == 1) { return; } else { //排序结点以得到当前创建结点的序号（等待锁的序位） Collections.sort(children); //获取生成的临时结点序号 String thisNode = createNode.substring("/locks/".length()); //获得排序 int index = children.indexOf(thisNode); if (index == -1) { System.out.println("数据异常"); } else if (index == 0) { //最小序号结点，直接获取锁 return; } else { //监听序号前一个结点 waitPath = "/locks/" + children.get(index - 1); //true代表使用创建zkClient时初始化的监听器 zkClient.getData(waitPath, true, null); waitLatch.await(); } } } /** * 解锁 * * @throws InterruptedException * @throws KeeperException */ public void zkUnLock() throws InterruptedException, KeeperException { //删除临时带序号结点 zkClient.delete(createNode, -1); } }         1.4 使用方式 public class DistributedLockTest { public static void main(String[] args) throws IOException, InterruptedException, KeeperException { ExecutorService executorService = Executors.newFixedThreadPool(2); DistributedLock lock1 = new DistributedLock(); DistributedLock lock2 = new DistributedLock(); //多线程获取锁1 CompletableFuture.supplyAsync(() -> { try { lock1.zkLock(); System.out.println("线程" + Thread.currentThread().getName() + "获取到锁......"); Thread.sleep(5000); lock1.zkUnLock(); System.out.println("线程" + Thread.currentThread().getName() + "释放锁......"); } catch (InterruptedException e) { throw new RuntimeException(e); } catch (KeeperException e) { throw new RuntimeException(e); } return true; }, executorService); //多线程获取锁2 CompletableFuture.supplyAsync(() -> { try { lock2.zkLock(); System.out.println("线程" + Thread.currentThread().getName() + "获取到锁......"); Thread.sleep(5000); lock2.zkUnLock(); System.out.println("线程" + Thread.currentThread().getName() + "释放锁......"); } catch (InterruptedException e) { throw new RuntimeException(e); } catch (KeeperException e) { throw new RuntimeException(e); } return true; }, executorService); executorService.shutdown(); } }

3. 引入Curator框架实现zookeeper分布式锁

        实际生产环境下，自然不可能手写这么多代码处理分布式锁，且不提很多地方的代码可复用，CountDownLatch反复处理带来的代码复杂性高，并且一些可重入锁、异常处理等逻辑上文也并没有完善。

        生产场景中被广泛使用的zookeeper分布式锁的框架便是Curator

        2.1 框架依赖 /.. 省略 ../ <!--Curator--> <dependency> <groupId>org.apache.curator</groupId> <artifactId>curator-framework</artifactId> <version>4.3.0</version> </dependency> <dependency> <groupId>org.apache.curator</groupId> <artifactId>curator-recipes</artifactId> <version>4.3.0</version> </dependency> <dependency> <groupId>org.apache.curator</groupId> <artifactId>curator-client</artifactId> <version>4.3.0</version> </dependency>

        2.2 使用方式 public class CuratorLockTest { public static void main(String[] args) { //创建分布式锁1 InterProcessLock lock1 = new InterProcessMutex(getCuratorFramework(), "/locks"); InterProcessLock lock2 = new InterProcessMutex(getCuratorFramework(), "/locks"); ExecutorService executorService = Executors.newFixedThreadPool(2); executorService.execute(() -> { try { lock1.acquire(); System.out.println("线程1获取到锁"); //curator支持可重入锁 lock1.acquire(); System.out.println("线程1 再次获取到锁"); Thread.sleep(5000); lock1.release(); System.out.println("线程1 释放锁"); lock1.release(); System.out.println("线程1 再次释放锁"); } catch (Exception e) { throw new RuntimeException(e); } }); executorService.execute(() -> { try { lock2.acquire(); System.out.println("线程2获取到锁"); lock2.acquire(); System.out.println("线程2 再次获取到锁"); Thread.sleep(5000); lock2.release(); System.out.println("线程2 释放锁"); lock2.release(); System.out.println("线程2 再次释放锁"); } catch (Exception e) { throw new RuntimeException(e); } } ); executorService.shutdown(); } public static CuratorFramework getCuratorFramework() { //4秒超时，重试3次 ExponentialBackoffRetry exponentialBackoffRetry = new ExponentialBackoffRetry(4000, 3); CuratorFramework client = CuratorFrameworkFactory.builder().connectString(ZkConstants.connectString) .connectionTimeoutMs(ZkConstants.sessionTimeout) .sessionTimeoutMs(ZkConstants.sessionTimeout) .retryPolicy(exponentialBackoffRetry) .build(); client.start(); System.out.println("zookeeper 启动成功..."); return client; } }

        希望能对大家理解zookeeper分布式锁有所帮助