ZooKeeper分布式锁与代码实现

女爷i 2022-06-02 23:50 127阅读 0赞

**1.zk的核心机制之一：分布式锁**

**       分布式锁能够在一组进程之间提供互斥机制，使得在任何时候只有一个进程可以持有锁。分布式锁可以用于在大型分布式系统中实现领导者选举，在任何时间点，持有锁的那个进程就是系统的领导者。注意：不要将zookeeper自己的领导者选举和使用ZooKeeper基本操作实现的一般领导者选举服务混为一谈。事实上，zookeepr自己的领导者选举机制是不对外公开的。**

![20171212162657386][] ![Image 1][]

2.锁的具体实现

![Image 1][] ![20171212162726867][]

代码实现一个分布式锁。

\- 客户端A

public class DistributedClient \{

// 超时时间

private static final int SESSION\_TIMEOUT = 5000;

// zookeeper server列表

private String hosts = "localhost:4180,localhost:4181,localhost:4182";

private String groupNode = "locks";

private String subNode = "sub";

private ZooKeeper zk;

// 当前client创建的子节点

private String thisPath;

// 当前client等待的子节点

private String waitPath;

private CountDownLatch latch = new CountDownLatch(1);

/\*\*

\* 连接zookeeper

\*/

public void connectZookeeper() throws Exception \{

zk = new ZooKeeper(hosts, SESSION\_TIMEOUT, new Watcher() \{

public void process(WatchedEvent event) \{

try \{

// 连接建立时, 打开latch, 唤醒wait在该latch上的线程

if (event.getState() == KeeperState.SyncConnected) \{

latch.countDown();

// 发生了waitPath的删除事件

if (event.getType() == EventType.NodeDeleted && event.getPath().equals(waitPath)) \{

doSomething();

\} catch (Exception e) \{

e.printStackTrace();

\});

// 等待连接建立

latch.await();

// 创建子节点

thisPath = zk.create("/" + groupNode + "/" + subNode, null, Ids.OPEN\_ACL\_UNSAFE,

CreateMode.EPHEMERAL\_SEQUENTIAL);

// wait一小会, 让结果更清晰一些

Thread.sleep(10);

// 注意, 没有必要监听"/locks"的子节点的变化情况

List<String> childrenNodes = zk.getChildren("/" + groupNode, false);

// 列表中只有一个子节点, 那肯定就是thisPath, 说明client获得锁

if (childrenNodes.size() == 1) \{

doSomething();

\} else \{

String thisNode = thisPath.substring(("/" + groupNode + "/").length());

// 排序

Collections.sort(childrenNodes);

int index = childrenNodes.indexOf(thisNode);

if (index == -1) \{

// never happened

\} else if (index == 0) \{

// inddx == 0, 说明thisNode在列表中最小, 当前client获得锁

**   doSomething();**

\} else \{

// 获得排名比thisPath前1位的节点

this.waitPath = "/" + groupNode + "/" + childrenNodes.get(index - 1);

// 在waitPath上注册监听器, 当waitPath被删除时, zookeeper会回调监听器的process方法

zk.getData(waitPath, true, new Stat());

private void doSomething() throws Exception \{

try \{

System.out.println("gain lock: " + thisPath);

Thread.sleep(2000);

// do something

\} finally \{

System.out.println("finished: " + thisPath);

// 将thisPath删除, 监听thisPath的client将获得通知

// 相当于释放锁

zk.delete(this.thisPath, -1);

public static void main(String\[\] args) throws Exception \{

for (int i = 0; i < 10; i++) \{

new Thread() \{

public void run() \{

try \{

DistributedClient dl = new DistributedClient();

dl.connectZookeeper();

\} catch (Exception e) \{

e.printStackTrace();

\}.start();

Thread.sleep(Long.MAX\_VALUE);

\- 分布式多进程模式实现：

public class DistributedClientMy \{

// 超时时间

private static final int SESSION\_TIMEOUT = 5000;

// zookeeper server列表

private String hosts = "spark01:2181,spark02:2181,spark03:2181";

private String groupNode = "locks";

private String subNode = "sub";

private boolean haveLock = false;

private ZooKeeper zk;

// 当前client创建的子节点

private volatile String thisPath;

/\*\*

\* 连接zookeeper

\*/

public void connectZookeeper() throws Exception \{

zk = new ZooKeeper("spark01:2181", SESSION\_TIMEOUT, new Watcher() \{

public void process(WatchedEvent event) \{

try \{

// 子节点发生变化

if (event.getType() == EventType.NodeChildrenChanged && event.getPath().equals("/" + groupNode)) \{

// thisPath是否是列表中的最小节点

List<String> childrenNodes = zk.getChildren("/" + groupNode, true);

String thisNode = thisPath.substring(("/" + groupNode + "/").length());

// 排序

Collections.sort(childrenNodes);

if (childrenNodes.indexOf(thisNode) == 0) \{

doSomething();

thisPath = zk.create("/" + groupNode + "/" + subNode, null, Ids.OPEN\_ACL\_UNSAFE,

CreateMode.EPHEMERAL\_SEQUENTIAL);

\} catch (Exception e) \{

e.printStackTrace();

\});

// 创建子节点

thisPath = zk.create("/" + groupNode + "/" + subNode, null, Ids.OPEN\_ACL\_UNSAFE,

CreateMode.EPHEMERAL\_SEQUENTIAL);

// wait一小会, 让结果更清晰一些

Thread.sleep(new Random().nextInt(1000));

// 监听子节点的变化

List<String> childrenNodes = zk.getChildren("/" + groupNode, true);

// 列表中只有一个子节点, 那肯定就是thisPath, 说明client获得锁

if (childrenNodes.size() == 1) \{

doSomething();

thisPath = zk.create("/" + groupNode + "/" + subNode, null, Ids.OPEN\_ACL\_UNSAFE,

CreateMode.EPHEMERAL\_SEQUENTIAL);

/\*\*

\* 共享资源的访问逻辑写在这个方法中

\*/

private void doSomething() throws Exception \{

try \{

System.out.println("gain lock: " + thisPath);

Thread.sleep(2000);

// do something

\} finally \{

System.out.println("finished: " + thisPath);

// 将thisPath删除, 监听thisPath的client将获得通知

// 相当于释放锁

zk.delete(this.thisPath, -1);

public static void main(String\[\] args) throws Exception \{

DistributedClientMy dl = new DistributedClientMy();

dl.connectZookeeper();

Thread.sleep(Long.MAX\_VALUE);

[20171212162657386]: https://img-blog.csdn.net/20171212162657386
[Image 1]: 
[20171212162726867]: /images/20220603/99c2c9743e0948bba2bc7a3dfe6e8ba2.png