如何基于Zookeeper实现分布式锁，手把手教程来啦~

不念不忘少年蓝@ 2021-09-24 13:56 148阅读 0赞

### 文章目录 ###

*  1 前言
 *  2 基于原生的zookeeper客户端实现分布式锁
 *   *  2.1 添加maven依赖
     *  2.2 添加监听器
     *  2.3 实现分布式锁
     *  2.4 功能测试
 *  3 基于原生的ZkClient客户端实现分布式锁
 *   *  3.1 添加maven依赖
     *  3.2 添加监听器
     *  3.3 实现分布式锁
     *  3.4 功能测试
 *  4 基于Curator客户端实现分布式锁（推荐）
 *   *  4.1 添加maven依赖
     *  4.2 实现分布式锁

# 1 前言 #

随着时代的发展，现在互联网流量越来越大，多线程高并发下共享资源的安全性越来越重要，那么如何保证共享资源的安全呢？

如果单机环境可能会想到加`synchronized`关键字上锁。如果是分布式环境下，那么这种方案就显得力不从心了。那么，如何实现共享资源的安全访问呢？毫无疑问，肯定是加分布式锁。

这里，笔者以Zookeeper为例，介绍下，如何实现分布式锁。

# 2 基于原生的zookeeper客户端实现分布式锁 #

## 2.1 添加maven依赖 ##

<dependency>
 <groupId>org.apache.zookeeper</groupId>
 <artifactId>zookeeper</artifactId>
 <version>3.5.5</version>
 </dependency>

## 2.2 添加监听器 ##

当前节点如果是最小节点，就获取到了锁，如果不是最小，需要监听它的上一个节点，如果上一个节点删除了，以此类推，最后当前节点，如果是最小的节点，就获取到了锁。

【代码实现示例】

package cn.smilehappiness.distributelock.zookeeper;
 
 import org.apache.zookeeper.WatchedEvent;
 import org.apache.zookeeper.Watcher;
 
 import java.util.concurrent.CountDownLatch;
 
 /** * * 监听器，监听接节点事件 * * * @author smilehappiness * @Date 2020/8/2 21:57 */
 public class ZookeeperLockWatcher implements Watcher { 
 
 private CountDownLatch countDownLatch;
 
 public ZookeeperLockWatcher(CountDownLatch countDownLatch) { 
 this.countDownLatch = countDownLatch;
 }
 
 @Override
 public void process(WatchedEvent event) { 
 //如果是节点删除事件
 if (event.getType() == Event.EventType.NodeDeleted) { 
 //倒计数器减1
 countDownLatch.countDown();
 }
 }
 }

## 2.3 实现分布式锁 ##

**思路：** 基于有序临时节点来实现分布式锁

【代码示例】

package cn.smilehappiness.distributelock.zookeeper;
 
 import org.apache.zookeeper.*;
 import org.apache.zookeeper.data.Stat;
 
 import java.util.List;
 import java.util.TreeSet;
 import java.util.concurrent.CountDownLatch;
 
 /** * 基于原生的zookeeper客户端API实现分布式锁锁 * 思路：基于有序临时节点来实现分布式锁 */
 public class ZookeeperLock { 
 
 //zookeeper原生客户端对象
 private ZooKeeper zooKeeper;
 
 private String zkAddress = "127.0.0.1:2181";
 
 //分布式锁的根节点的名称
 private String lockRootName = "/locks";
 
 //锁节点的名称
 private String lockName;
 
 //当前的锁节点名称
 private String currentLockName;
 
 private static final int sessionTimeout = 10000;
 
 //默认的节点的数据
 private static final byte[] bytes = new byte[0];
 
 //倒计数器
 private CountDownLatch countDownLatch = new CountDownLatch(1);
 
 /** * 构造方法 * * @param lockName */
 public ZookeeperLock(String lockName) { 
 //锁节点的名称通过构造方法初始化
 this.lockName = lockName;
 try { 
 //建立zookeeper连接
 zooKeeper = new ZooKeeper(zkAddress, sessionTimeout, (WatchedEvent event) -> { 
 if (event.getState() == Watcher.Event.KeeperState.SyncConnected) { 
 //zookeeper连上了
 countDownLatch.countDown();
 }
 });
 //等待，阻塞
 countDownLatch.await();
 
 //如果阻塞结束，说明连接zookeeper成功，我们创建一个锁的根节点 lockRootName = "/locks";
 /** * --/locks --业务区分 * --storeLock000000001 * --storeLock000000001 * --storeLock000000001 * ...... */
 Stat stat = zooKeeper.exists(lockRootName, false);
 //如果锁的根节点不存在
 if (null == stat) { 
 //根节点持久化，acl的开放的
 zooKeeper.create(lockRootName, bytes, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
 }
 } catch (Exception e) { 
 throw new RuntimeException(e);
 }
 }
 
 /** * zookeeper分布式锁：加锁 (获取分布式锁) */
 public void lock() { 
 try { 
 /** * 返回： * /locks/lockName0000001 * /locks/lockName0000002 * /locks/lockName0000003 * ......... */
 String myNode = zooKeeper.create(lockRootName + "/" + lockName, bytes, ZooDefs.Ids.OPEN_ACL_UNSAFE,
 CreateMode.EPHEMERAL_SEQUENTIAL);
 
 //拿到根节点下的所有临时有序子节点
 List<String> subNodes = zooKeeper.getChildren(lockRootName, false);
 
 //把所有子节点排序一下 (默认字典排序，0-9，a-z)
 TreeSet<String> sortNodes = new TreeSet<>();
 for (String node : subNodes) { 
 // /locks/lockName0000001
 sortNodes.add(lockRootName + "/" + node);
 }
 
 //从排好顺序的set集合中取第一个节点，它是节点编号最小的
 String minNode = sortNodes.first();
 
 System.out.println("当前的myNode=" + myNode);
 System.out.println("最小节点minNode=" + minNode);
 
 //获取一下前一个节点，获取指定节点的前一个节点
 // myNode: /locks/lockName0000003, （/locks/lockName0000002、/locks/lockName0000001）
 String preNode = sortNodes.lower(myNode);
 System.out.println("前一个节点preNode=" + preNode);
 
 //最小节点能拿到分布式锁
 if (myNode.equals(minNode)) { 
 //当前进来的这个线程所创建的myNode就是分布式锁节点
 currentLockName = myNode;
 //已经获取到分布式锁
 return;
 }
 
 //其他进来的线程没有拿到分布锁，因为它所创建的节点不是最小的，那么他就监听前一个节点的删除事件
 //一个并发线程工具类：倒计数器
 CountDownLatch countDownLatch = new CountDownLatch(1);
 //判断字符串是否为null
 if (null != preNode) { 
 //如果前一个节点不是空的，那么我就要监听前一个节点，当它删除时触发我的监听事件
 Stat stat = zooKeeper.exists(preNode, new ZookeeperLockWatcher(countDownLatch));
 
 if (null != stat) { 
 //阻塞，等待，那什么时候等待结束，就看前一个节点被监听器监听到删除事件，此时的阻塞等待结束
 countDownLatch.await();
 
 //又拿到分布式锁
 currentLockName = myNode;
 
 //倒计数对象置为null，促进垃圾回收
 countDownLatch = null;
 }
 }
 } catch (Exception e) { 
 throw new RuntimeException(e);
 }
 }
 
 /** * zookeeper分布式锁：解锁 */
 public void unLock() { 
 //解锁主要就是把当前锁的节点从zookeeper中删除
 //解锁：另外一个做法是直接关闭zookeeper客户端（问题就是：你下次还要用锁，那就需要重新再建立zookeeper连接）
 try { 
 if (currentLockName != null) { 
 //版本号 -1 表示任何版本，不需要匹配版本，不管你是什么版本，我都可以删除
 zooKeeper.delete(currentLockName, -1);
 }
 } catch (Exception e) { 
 throw new RuntimeException(e);
 }
 }
 }

## 2.4 功能测试 ##

/** * * 分布式锁测试 * * * @param * @return void * @Date 2020/8/2 22:10 */
 private void testDistributedLock() { 
 //使用原生zookeeper客户端api，底层创建zookeeper连接、创建锁的根节点
 ZookeeperLock lock = new ZookeeperLock("storeLock");
 
 try { 
 //获取分布式锁，然后下面的业务代码就会按顺序排队执行
 lock.lock();
 //TODO 拿到redis分布式锁，执行业务代码
 // ......
 
 } catch (Exception e) { 
 throw new RuntimeException(e);
 } finally { 
 //释放分布式锁
 lock.unLock();
 }
 }

# 3 基于原生的ZkClient客户端实现分布式锁 #

## 3.1 添加maven依赖 ##

<dependency>
 <groupId>com.101tec</groupId>
 <artifactId>zkclient</artifactId>
 <version>0.11</version>
 </dependency>

## 3.2 添加监听器 ##

【代码实现示例】

package cn.smilehappiness.distributelock.zkclient;
 
 import org.I0Itec.zkclient.IZkDataListener;
 
 import java.util.concurrent.CountDownLatch;
 
 /** * * 对节点的监听，当前一个节点被删除的时候，触发节点删除事件的监听 * * * @author smilehappiness * @Date 2020/8/2 21:56 */
 public class ZkClientLockWatcher implements IZkDataListener { 
 
 private CountDownLatch countDownLatch;
 
 public ZkClientLockWatcher(CountDownLatch countDownLatch) { 
 this.countDownLatch = countDownLatch;
 }
 
 @Override
 public void handleDataChange(String dataPath, Object data) throws Exception { 
 }
 
 /** * 监听节点删除事件 * * @param dataPath * @throws Exception */
 @Override
 public void handleDataDeleted(String dataPath) throws Exception { 
 //节点删除，把倒计数器减1
 countDownLatch.countDown();
 }
 }

## 3.3 实现分布式锁 ##

【代码示例】

package cn.smilehappiness.distributelock.zkclient;
 
 import org.I0Itec.zkclient.ZkClient;
 import org.apache.zookeeper.CreateMode;
 import org.apache.zookeeper.ZooDefs;
 
 import java.util.List;
 import java.util.TreeSet;
 import java.util.concurrent.CountDownLatch;
 
 /** * 基于zkClient第三方客户端实现分布式锁 * 思路：基于有序临时节点来实现分布式锁 */
 public class ZkClientLock { 
 
 //zkclient第三档客户端对象
 private ZkClient zkClient;
 
 private String zkAddress = "127.0.0.1:2181";
 
 //分布式锁的根节点的名称
 private String lockRootName = "/locks";
 
 //锁节点的名称
 private String lockName;
 
 //当前的锁节点名称
 private String currentLockName;
 
 private static final int sessionTimeOut = 10000;
 
 private static final int connectionTimeOut = 25000;
 
 //默认的节点的数据
 private static final byte[] bytes = new byte[0];
 
 //倒计数器
 private CountDownLatch countDownLatch = new CountDownLatch(1);
 
 /** * 构造方法 * * @param lockName */
 public ZkClientLock(String lockName) { 
 //锁节点的名称通过构造方法初始化
 this.lockName = lockName;
 try { 
 //建立zookeeper连接
 zkClient = new ZkClient(zkAddress, sessionTimeOut, connectionTimeOut);
 
 //创建zkclient对象完成，说明连接zookeeper成功，我们创建一个锁的根节点 lockRootName = "/locks";
 /** * --/locks --业务区分 * --storeLock000000001 * --storeLock000000001 * --storeLock000000001 * ...... */
 boolean isExists = zkClient.exists(lockRootName);
 //如果锁的根节点不存在
 if (!isExists) { 
 //根节点持久化，acl的开放的
 zkClient.create(lockRootName, bytes, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
 }
 } catch (Exception e) { 
 throw new RuntimeException(e);
 }
 }
 
 /** * zookeeper分布式锁：加锁 (获取分布式锁) */
 public void lock() { 
 try { 
 /** * 返回： * /locks/lockName0000001 * /locks/lockName0000002 * /locks/lockName0000003 * ......... */
 String myNode = zkClient.create(lockRootName + "/" + lockName, bytes, ZooDefs.Ids.OPEN_ACL_UNSAFE,
 CreateMode.EPHEMERAL_SEQUENTIAL);
 
 //拿到根节点下的所有临时有序子节点
 List<String> subNodes = zkClient.getChildren(lockRootName);
 
 //把所有子节点排序一下 (默认字典排序，0-9，a-z)
 TreeSet<String> sortNodes = new TreeSet<String>();
 for (String node : subNodes) { 
 // /locks/lockName0000001
 sortNodes.add(lockRootName + "/" + node);
 }
 
 //从排好顺序的set集合中取第一个节点，它是节点编号最小的
 String minNode = sortNodes.first();
 
 System.out.println("当前的myNode=" + myNode);
 System.out.println("最小节点minNode=" + minNode);
 
 //获取一下前一个节点，获取指定节点的前一个节点
 // myNode: /locks/lockName0000003, （/locks/lockName0000002、/locks/lockName0000001）
 String preNode = sortNodes.lower(myNode);
 System.out.println("前一个节点preNode=" + preNode);
 
 //最小节点能拿到分布式锁
 if (myNode.equals(minNode)) { 
 //当前进来的这个线程所创建的myNode就是分布式锁节点
 currentLockName = myNode;
 //已经获取到分布式锁
 return;
 }
 
 //其他进来的线程没有拿到分布锁，因为它所创建的节点不是最小的，那么他就监听前一个节点的删除事件
 //一个并发线程工具类：倒计数器
 CountDownLatch countDownLatch = new CountDownLatch(1);
 //判断字符串是否为null
 if (null != preNode) { 
 //如果前一个节点不是空的，那么我就要监听前一个节点，当它删除时触发我的监听事件
 boolean isExists = zkClient.exists(preNode);
 
 if (isExists) { 
 //监听前一个节点
 zkClient.subscribeDataChanges(preNode, new ZkClientLockWatcher(countDownLatch));
 
 countDownLatch.await();
 
 //又拿到分布式锁
 currentLockName = myNode;
 
 //倒计数对象置为null，促进垃圾回收
 countDownLatch = null;
 }
 }
 } catch (Exception e) { 
 throw new RuntimeException(e);
 }
 }
 
 /** * zookeeper分布式锁：解锁 */
 public void unLock() { 
 //解锁主要就是把当前锁的节点从zookeeper中删除
 //解锁：另外一个做法是直接关闭zookeeper客户端（问题就是：你下次还要用锁，那就需要重新再建立zookeeper连接）
 try { 
 if (currentLockName != null) { 
 //版本号 -1 表示任何版本，不需要匹配版本，不管你是什么版本，我都可以删除
 zkClient.delete(currentLockName, -1);
 }
 } catch (Exception e) { 
 throw new RuntimeException(e);
 }
 }
 }

## 3.4 功能测试 ##

/** * * 分布式锁测试 * * * @param * @return void * @Date 2020/8/2 22:30 */
 private void testDistributedLock() { 
 //使用原生zookeeper客户端api，底层创建zookeeper连接、创建锁的根节点
 ZookeeperLock lock = new ZookeeperLock("storeLock");
 
 try { 
 //获取分布式锁，然后下面的业务代码就会按顺序排队执行
 lock.lock();
 //TODO 拿到redis分布式锁，执行业务代码
 // ......
 
 } catch (Exception e) { 
 throw new RuntimeException(e);
 } finally { 
 //释放分布式锁
 lock.unLock();
 }
 }

# 4 基于Curator客户端实现分布式锁（推荐） #

由于Curator客户端给我们提供了现成的分布式互斥锁来实现分布式锁，所以我们不必自己开发。工作中为了效率还是建议使用Curator客户端。

## 4.1 添加maven依赖 ##

<dependency>
 <groupId>org.apache.curator</groupId>
 <artifactId>curator-recipes</artifactId>
 <version>4.2.0</version>
 </dependency>

## 4.2 实现分布式锁 ##

**这里采用共享可重入锁实现，代码示例如下：**

package cn.smilehappiness.distributelock.curator;
 
 import org.apache.curator.RetryPolicy;
 import org.apache.curator.framework.CuratorFramework;
 import org.apache.curator.framework.CuratorFrameworkFactory;
 import org.apache.curator.framework.recipes.locks.InterProcessLock;
 import org.apache.curator.framework.recipes.locks.InterProcessMutex;
 import org.apache.curator.retry.ExponentialBackoffRetry;
 import org.apache.curator.retry.RetryNTimes;
 import org.springframework.util.Assert;
 
 import java.util.concurrent.TimeUnit;
 
 /** * * 基于Curator客户端实现分布式锁 * * * @author smilehappiness * @Date 2020/8/2 22:28 */
 public class CuratorClientLock { 
 
 /** * 客户端连接地址 */
 private static final String ZK_ADDRESS = "127.0.0.1:2181";
 /** * 客户端根节点 */
 private static final String LOCK_NODE = "/lockRoot";
 /** * 会话超时时间 */
 private final int SESSION_TIMEOUT = 30 * 1000;
 /** * 连接超时时间 */
 private final int CONNECTION_TIMEOUT = 5 * 1000;
 /** * 创建zookeeper连接实例 */
 private static CuratorFramework client = null;
 private static CuratorFramework client2 = null;
 
 /** * 重试策略 * baseSleepTimeMs：初始的重试等待时间，单位毫秒 * maxRetries：最多重试次数 */
 private static final RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
 /** * 重试策略 * n：最多重试次数 * sleepMsBetweenRetries：重试时间间隔，单位毫秒 */
 private static final RetryPolicy retry = new RetryNTimes(3, 2000);
 
 static { 
 // 创建Curator连接对象
 connectCuratorClient();
 }
 
 /** * * 创建Curator连接对象 * * * @param * @return * @Date 2020/6/21 12:29 */
 public static void connectCuratorClient() { 
 //老版本的方式，创建zookeeper连接客户端
 client = CuratorFrameworkFactory.builder()
 .connectString(ZK_ADDRESS)
 .sessionTimeoutMs(5000)
 .connectionTimeoutMs(10000)
 .retryPolicy(retry)
 .build();
 
 //创建zookeeper连接，新版本
 client2 = CuratorFrameworkFactory.newClient(ZK_ADDRESS, retry);
 
 //启动客户端（Start the client. Most mutator methods will not work until the client is started）
 client.start();
 client2.start();
 
 System.out.println("zookeeper初始化连接成功：" + client);
 System.out.println("zookeeper初始化连接成功：" + client2);
 }
 
 public static void main(String[] args) throws Exception { 
 CuratorClientLock curatorClientLock = new CuratorClientLock();
 //测试1
 test1();
 
 //测试2
 sharedReentrantLock();
 }
 
 private static void test1() throws Exception { 
 InterProcessMutex lock = new InterProcessMutex(client, LOCK_NODE);
 try { 
 // 方式一：直接获取锁
 lock.acquire();
 
 // 方式二：获取锁，60秒如果获取不到，超时（Acquire the mutex - blocks until it's available or the given time expires）
 if (lock.acquire(60, TimeUnit.MINUTES)) { 
 
 }
 
 //TODO 获取到分布式锁后，执行业务处理
 
 } finally { 
 if (lock.isAcquiredInThisProcess()) { 
 lock.release();
 }
 }
 }
 
 public static void sharedReentrantLock() throws Exception { 
 // 创建可重入锁
 InterProcessLock lock = new InterProcessMutex(client, LOCK_NODE);
 // lock2 用于模拟其他客户端
 InterProcessLock lock2 = new InterProcessMutex(client2, LOCK_NODE);
 
 // lock 获取锁
 lock.acquire();
 try { 
 // lock 第二次获取锁
 lock.acquire();
 //TODO 获取到分布式锁后，执行业务处理
 
 try { 
 // lock2 超时获取锁, 因为锁已经被 lock 客户端占用, 所以获取失败, 需要等 lock 释放
 System.out.println(lock2.acquire(2, TimeUnit.SECONDS));
 ;
 } finally { 
 lock.release();
 }
 } finally { 
 // 重入锁获取与释放需要一一对应, 如果获取 2 次, 释放 1 次, 那么该锁依然是被占用, 如果将下面这行代码注释, 那么会发现下面的 lock2 获取锁失败
 lock.release();
 }
 
 // 在 lock 释放后, lock2 能够获取锁
 Assert.isTrue(lock2.acquire(2, TimeUnit.SECONDS));
 lock2.release();
 }
 
 }

是不是还是Curator客户端实现起来更简单呢，不用自己实现，不用监听删除事件。

好啦，本篇就总结到这里了！

**参考资料链接：**  
[https://www.jianshu.com/p/6618471f6e75][https_www.jianshu.com_p_6618471f6e75]  
[https://www.jianshu.com/p/31335efec309][https_www.jianshu.com_p_31335efec309]

> 写博客是为了记住自己容易忘记的东西，另外也是对自己工作的总结，希望尽自己的努力，做到更好，大家一起努力进步！  
>   
> 如果有什么问题，欢迎大家评论，一起探讨，代码如有问题，欢迎各位大神指正！  
>   
> 给自己的梦想添加一双翅膀，让它可以在天空中自由自在的飞翔！

[https_www.jianshu.com_p_6618471f6e75]: https://www.jianshu.com/p/6618471f6e75
[https_www.jianshu.com_p_31335efec309]: https://www.jianshu.com/p/31335efec309