Zookeeper-分布式锁案例
Posted 逆十字
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了Zookeeper-分布式锁案例相关的知识，希望对你有一定的参考价值。
Zookeeper分布式锁案例
什么叫做分布式锁呢？
比如说"进程 1"在使用该资源的时候，会先去获得锁，"进程 1"获得锁以后会对该资源保持独占，这样其他进程就无法访问该资源，"进程 1"用完该资源以后就将锁释放掉，让其 他进程来获得锁，那么通过这个锁机制，我们就能保证了分布式系统中多个进程能够有序的 访问该临界资源。那么我们把这个分布式环境下的这个锁叫作分布式锁。
原生Zookeeper实现分布式锁案例
分布式锁实现
package com.atguigu.case2;
import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;
import java.io.IOException;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;
/**
* @author: fxl
* @Description:
* @Data:Create in 2022-02-05
* @Modified By:
*/
public class DistributedLock
private final String connectString = "hadoop102:2181,hadoop103:2181,hadoop104:2181";
private final int sessionTimeout = 2000;
private final ZooKeeper zk;
private final CountDownLatch connectLatch = new CountDownLatch(1);
private final CountDownLatch waitLatch = new CountDownLatch(1);
private String waitPath;
private String currentMode;
public DistributedLock() throws IOException, InterruptedException, KeeperException
//获取连接
zk = new ZooKeeper(connectString, sessionTimeout, new Watcher()
@Override
public void process(WatchedEvent watchedEvent)
// connnectLatch 如果连接上zk 可以释放
if (watchedEvent.getState() == Event.KeeperState.SyncConnected)
connectLatch.countDown();
//waitLatch 需要释放
if (watchedEvent.getType() == Event.EventType.NodeDeleted && watchedEvent.getPath().equals(waitPath))
waitLatch.countDown();
);
// 等待zk正常连接后，往下走程序
connectLatch.await();
//判断根节点/locks是否存在
Stat stat = zk.exists("/locks", false);
if (stat == null)
//创建根节点
zk.create("/locks", "locks".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
//对zk加锁
public void zklock() throws InterruptedException, KeeperException
//创建对应的临时带序号的节点
currentMode = zk.create("/locks/" + "seq-", null, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
//判断创建的节点是否是最小的序号节点，如果是获取到锁，如果不是，监听前一个节点
List<String> children = zk.getChildren("/locks", false);
//如果只有一个节点，直接获取锁，如果有多个节点，就需要判断那个节点的值最小
if (children.size() == 1)
return;
else
Collections.sort(children);
//获取对应的节点名称 seq-00000000
String thisNode = currentMode.substring("/locks/".length());
//通过seq-00000000获取该节点在children的位置
int index = children.indexOf(thisNode);
if (index == -1)
System.out.println("数据异常");
else if (index == 0)
//就一个节点获取锁了
return;
else
//需要监听前一个节点的变化
waitPath = "/locks/" + children.get(index - 1);
zk.getData(waitPath, true, new Stat());
//等待监听
waitLatch.await();
return ;
;
//解锁
public void unzklock() throws InterruptedException, KeeperException
//删除节点
zk.delete(currentMode, -1);
分布式锁测试
创建两个线程
package com.atguigu.case2;
import org.apache.zookeeper.KeeperException;
import java.io.IOException;
/**
* @author: fxl
* @Description:
* @Data:Create in 2022-02-05
* @Modified By:
*/
public class DistributedLockTest
public static void main(String[] args) throws IOException, InterruptedException, KeeperException
final DistributedLock lock1 = new DistributedLock();
final DistributedLock lock2 = new DistributedLock();
new Thread(new Runnable()
@Override
public void run()
try
lock1.zklock();
System.out.println("线程1,获取到锁");
Thread.sleep(5 * 1000);
lock1.unzklock();
System.out.println("线程1 释放锁");
catch (InterruptedException e)
e.printStackTrace();
catch (KeeperException e)
e.printStackTrace();
).start();
new Thread(new Runnable()
@Override
public void run()
try
lock2.zklock();
System.out.println("线程2,获取到锁");
Thread.sleep(5 * 1000);
lock2.unzklock();
System.out.println("线程2 释放锁");
catch (InterruptedException e)
e.printStackTrace();
catch (KeeperException e)
e.printStackTrace();
).start();
观察控制台变化
线程 1 获取锁
线程 1 释放锁
线程 2 获取锁
线程 2 释放锁
Curator 框架实现分布式锁案例
原生的 Java API 开发存在的问题
会话连接是异步的，需要自己去处理。比如使用 CountDownLatch Watch 需要重复注册，不然就不能生效 开发的复杂性还是比较高的 不支持多节点删除和创建。需要自己去递归
Curator 是一个专门解决分布式锁的框架，解决了原生 JavaAPI 开发分布式遇到的问题
详情请查看官方文档：https://curator.apache.org/index.html
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>代码实现
package com.atguigu.case3;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.locks.InterProcessMultiLock;
import org.apache.curator.framework.recipes.locks.InterProcessMutex;
import org.apache.curator.retry.ExponentialBackoffRetry;
/**
* @author: fxl
* @Description:
* @Data:Create in 2022-02-05
* @Modified By:
*/
public class CuratorLockTest
public static void main(String[] args)
//创建分布式锁1
InterProcessMutex lock1 = new InterProcessMutex(getCuratorFramework(), "/locks");
//创建分布式锁2
InterProcessMutex lock2 = new InterProcessMutex(getCuratorFramework(), "/locks");
new Thread(new Runnable()
@Override
public void run()
try
lock1.acquire();
System.out.println("线程1 获取到锁");
lock1.acquire();
System.out.println("线程1 再次获取到锁");
Thread.sleep(5 * 1000);
lock1.release();
System.out.println("线程1 释放锁");
lock1.release();
System.out.println("线程1 再次释放锁");
catch (Exception e)
e.printStackTrace();
).start();
new Thread(new Runnable()
@Override
public void run()
try
lock2.acquire();
System.out.println("线程2 获取到锁");
lock2.acquire();
System.out.println("线程2 再次获取到锁");
Thread.sleep(5 * 1000);
lock2.release();
System.out.println("线程2 释放锁");
lock2.release();
System.out.println("线程2 再次释放锁");
catch (Exception e)
e.printStackTrace();
).start();
private static CuratorFramework getCuratorFramework()
ExponentialBackoffRetry policy = new ExponentialBackoffRetry(3000, 3);
CuratorFramework client = CuratorFrameworkFactory.builder().connectString("hadoop102:2181,hadoop103:2181,hadoop104:2181")
.connectionTimeoutMs(2000)
.sessionTimeoutMs(2000)
.retryPolicy(policy).build();
//启动客户端
client.start();
System.out.println("zookeeper 启动成功");
return client;
观察控制台变化
线程 1 获取锁
线程 1 再次获取锁
线程 1 释放锁
线程 1 再次释放锁
线程 2 获取锁
线程 2 再次获取锁
线程 2 释放锁
线程 2 再次释放锁
以上是关于Zookeeper-分布式锁案例的主要内容，如果未能解决你的问题，请参考以下文章
Day377.服务器动态上下线监听案例&ZooKeeper分布式锁案例&企业面试真题 -Zookeeper
分布式锁三大技术方案实战——基于zookeeper方式实现分布式锁
Zookeeper -- Zookeeper JavaAPI相关操作（Curator介绍Curator API 常用操作（节点的CRUD，Watch事件监听）分布式锁模拟12306售票案例）