【面试】Zookeeper - IT评测·应用市场-qidao123.com

package com.example.test.other.zk;
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;
public class ZookepeerLock1 {
private final String connectionString = "127.0.0.1:2181,127.0.0.1:2182,127.0.0.1:2183";
private final int sessionTimeout = 2000;
private final ZooKeeper zk;
private CountDownLatch countDownLatch = new CountDownLatch(1);
private CountDownLatch waitLatch = new CountDownLatch(1);
private String waitPath;
private String currentMode;
public ZookepeerLock1() throws IOException, InterruptedException, KeeperException {
// 获取连接
zk = new ZooKeeper(connectionString, sessionTimeout, new Watcher() {
@Override
public void process(WatchedEvent watchedEvent) {
// connectLatch 如果连接上zk 可以释放
if (watchedEvent.getState() == Event.KeeperState.SyncConnected) {
countDownLatch.countDown();
}
// waitLatch 需要释放
if (watchedEvent.getType() == Event.EventType.NodeDeleted && watchedEvent.getPath().equals(waitPath)) {
waitLatch.countDown();
}
}
});
// 等待zk正常连接后，往下走程序
countDownLatch.await();
// 判断根节点/locks是否存在
Stat stat = zk.exists("/lockZookeeper", false);
if (stat == null) {
// 创建根节点，这是⼀个完全开放的ACL，持久节点
zk.create("/lockZookeeper", "lockZookeeper".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
}
}
// 对zk加锁
public void zkLock() {
try {
// 创建对应的临时顺序节点
currentMode =
zk.create("/lockZookeeper/" + "seq-", null, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
// 判断创建的节点是否是序号最小的节点，如果是获取到锁，如果不是，监听他序号前一个节点
List<String> children = zk.getChildren("/lockZookeeper", false);
if (children.size() == 1) {
return;
} else {
//[seq-0000000016, seq-0000000017]
Collections.sort(children);
// 获取节点名称 /locks/seq-0000000017 -> seq-0000000017
String thisNode = currentMode.substring("/lockZookeeper/".length());
// 通过seq-0000000017获取该节点在children集合的位置
int index = children.indexOf(thisNode);
// 判断
if (index == -1) {
System.out.println("数据异常");
} else if (index == 0) {
// 就一个节点，可以获取锁了
return;
} else {
// 需要监听他前一个结点的变化
waitPath = "/lockZookeeper/" + children.get(index - 1);
zk.getData(waitPath, true, null);
// 等待监听
waitLatch.await();
}
}
} catch (KeeperException e) {
e.printStackTrace();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
// 解锁
public void unZkLock() {
// 删除节点
try {
zk.delete(currentMode, -1);
} catch (InterruptedException e) {
e.printStackTrace();
} catch (KeeperException e) {
e.printStackTrace();
}
}
}

复制代码

package com.example.test.other.zk;
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.framework.recipes.locks.InterProcessSemaphoreMutex;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.KeeperException;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import java.io.IOException;
/**
* Description: 1、@Bean单独注解方法时，每次调用方法都是执行方法内的逻辑并返回新创建的对象bean，而且SpringIOC并没有该bean的存在。
* 2、@Bean + @Configuration ，在调用@Bean注解的方法时返回的实例bean是从IOC容器获取的，已经注入的，且是单例的，而不是新创建的。
* 3、@Bean + @Component，虽然@Bean注解的方法返回的实例已经注入到SpringIOC容器中，但是每次调用@Bean注解的方法时，都会创建新的对象实例bean返回，并不会从IOC容器中获取。
* Author: yangjj_tc
* Date: 2023/5/18 13:25
*/
@Configuration
public class ZookeperLock1Test {
private final String ZOOKEEPER_ADDRESS = "127.0.0.1:2181,127.0.0.1:2182,127.0.0.1:2183";
/**
* Description: Author:
* yangjj_tc
* 1、会话连接是异步的，需要自己去处理。比如使用CountDownLatch
* 2、Watch需要重复注册，不然就不能生效
* 3、开发的复杂性还是比较高的
* 4、不支持多节点删除和创建。需要自己去递归
* Date: 2023/5/18 12:48
*/
@Bean
public CuratorFramework getCuratorFramework(){
// 重试策略，重试间隔时间为1秒，重试次数为3次。curator管理了zookeeper的连接，在操作zookeeper的过程中出现连接问题会自动重试。
RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
// 初始化客户端，通过工厂创建连接
// zk地址会话超时时间，默认60秒连接超时时间，默认15秒重试策略
CuratorFramework zkClient = CuratorFrameworkFactory.newClient(ZOOKEEPER_ADDRESS, 5000, 15000, retryPolicy);
// 开始连接
zkClient.start();
return zkClient;
}
public static void main(String[] args) throws InterruptedException, IOException, KeeperException {
// 获得两个客户端
CuratorFramework client1 = new ZookeperLock1Test().getCuratorFramework();
CuratorFramework client2 = new ZookeperLock1Test().getCuratorFramework();
// 可重入锁，意味着同一个客户端在拥有锁的同时，可以多次获取，不会被阻塞。如想重入，则需要使用同一个InterProcessMutex对象。
final InterProcessLock lock1 = new InterProcessMutex(client1, "/lockCurator");
final InterProcessLock lock2 = new InterProcessMutex(client2, "/lockCurator");
// 不可重入锁，区别在于该锁是不可重入的，在同一个线程中不可重入
final InterProcessSemaphoreMutex lock3 = new InterProcessSemaphoreMutex(client1, "/lockCurator");
final InterProcessSemaphoreMutex lock4 = new InterProcessSemaphoreMutex(client2, "/lockCurator");
// 模拟两个线程
new Thread(() -> {
try {
// 线程加锁
lock3.acquire();
lock3.acquire();
System.out.println("线程1获取锁");
// 线程沉睡
Thread.sleep(5 * 1000);
// 线程解锁
lock1.release();
System.out.println("线程1释放了锁");
} catch (Exception e) {
e.printStackTrace();
}
}).start();
// 线程2
new Thread(() -> {
// 线程加锁
try {
lock2.acquire();
System.out.println("线程2获取到锁");
// 线程沉睡
Thread.sleep(5 * 1000);
lock2.release();
System.out.println("线程2释放锁");
} catch (Exception e) {
e.printStackTrace();
}
}).start();
}
}

复制代码