ZooKeeper分布式锁、配置管理、服务发现在Java开发中的应用

ZooKeeper提供了多种功能，包括分布式锁、配置管理、服务发现、领导选举等。

下面是一些常见的ZooKeeper功能及其在Java中的应用示例代码。

分布式锁

import org.apache.zookeeper.*;
import java.io.IOException;
import java.util.concurrent.CountDownLatch;public class DistributedLock implements Watcher {private static final String ZOOKEEPER_ADDRESS = "localhost:2181";private static final int SESSION_TIMEOUT = 5000;private static final String LOCK_PATH = "/distributed-lock";private ZooKeeper zooKeeper;private String currentLockPath;private CountDownLatch lockSignal;public DistributedLock() throws IOException, InterruptedException, KeeperException {// 创建ZooKeeper对象，建立与ZooKeeper服务器的连接zooKeeper = new ZooKeeper(ZOOKEEPER_ADDRESS, SESSION_TIMEOUT, this);lockSignal = new CountDownLatch(1);// 确保锁的根节点存在ensurePathExists(LOCK_PATH);}public void lock() throws KeeperException, InterruptedException {// 创建临时顺序节点作为锁节点String lockNodePath = zooKeeper.create(LOCK_PATH + "/lock-", new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);while (true) {// 获取锁节点下的所有子节点List<String> children = zooKeeper.getChildren(LOCK_PATH, false);Collections.sort(children);// 获取当前锁节点在所有子节点中的位置int index = children.indexOf(lockNodePath.substring(LOCK_PATH.length() + 1));if (index == 0) {// 如果当前锁节点是第一个节点，则获取到了锁this.currentLockPath = lockNodePath;return;} else {// 如果当前锁节点不是第一个节点，则监听前一个节点的删除事件，然后等待String previousLockPath = LOCK_PATH + "/" + children.get(index - 1);zooKeeper.exists(previousLockPath, true);lockSignal.await();}}}public void unlock() throws KeeperException, InterruptedException {// 删除当前锁节点zooKeeper.delete(currentLockPath, -1);currentLockPath = null;}private void ensurePathExists(String path) throws KeeperException, InterruptedException {// 确保路径存在，如果不存在则创建持久节点if (zooKeeper.exists(path, false) == null) {zooKeeper.create(path, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);}}@Overridepublic void process(WatchedEvent watchedEvent) {if (watchedEvent.getType() == Event.EventType.NodeDeleted && watchedEvent.getPath().equals(currentLockPath)) {// 当前锁节点被删除时，唤醒等待线程lockSignal.countDown();}}
}

配置管理

import org.apache.zookeeper.*;
import java.io.IOException;
import java.util.concurrent.CountDownLatch;public class ConfigManager implements Watcher {private static final String ZOOKEEPER_ADDRESS = "localhost:2181";private static final int SESSION_TIMEOUT = 5000;private static final String CONFIG_PATH = "/config";private ZooKeeper zooKeeper;private CountDownLatch configSignal;private String currentConfig;public ConfigManager() throws IOException, InterruptedException, KeeperException {// 创建ZooKeeper对象，建立与ZooKeeper服务器的连接zooKeeper = new ZooKeeper(ZOOKEEPER_ADDRESS, SESSION_TIMEOUT, this);configSignal = new CountDownLatch(1);// 确保配置节点存在ensurePathExists(CONFIG_PATH);}public String getConfig() throws KeeperException, InterruptedException {// 获取配置节点的数据，并等待配置更新byte[] data = zooKeeper.getData(CONFIG_PATH, true, null);configSignal.await();return new String(data);}private void ensurePathExists(String path) throws KeeperException, InterruptedException {// 确保路径存在，如果不存在则创建持久节点if (zooKeeper.exists(path, false) == null) {zooKeeper.create(path, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);}}@Overridepublic void process(WatchedEvent watchedEvent) {if (watchedEvent.getType() == Event.EventType.NodeDataChanged && watchedEvent.getPath().equals(CONFIG_PATH)) {try {// 当配置节点数据发生变化时，获取最新的配置数据，并唤醒等待线程byte[] data = zooKeeper.getData(CONFIG_PATH, true, null);currentConfig = new String(data);configSignal.countDown();} catch (KeeperException | InterruptedException e) {e.printStackTrace();}}}
}

服务发现

import org.apache.zookeeper.*;
import java.io.IOException;
import java.util.List;
import java.util.concurrent.CountDownLatch;public class ServiceDiscovery implements Watcher {private static final String ZOOKEEPER_ADDRESS = "localhost:2181";private static final int SESSION_TIMEOUT = 5000;private static final String SERVICE_PATH = "/services";private ZooKeeper zooKeeper;private CountDownLatch serviceSignal;private List<String> currentServices;public ServiceDiscovery() throws IOException, InterruptedException, KeeperException {// 创建ZooKeeper对象，建立与ZooKeeper服务器的连接zooKeeper = new ZooKeeper(ZOOKEEPER_ADDRESS, SESSION_TIMEOUT, this);serviceSignal = new CountDownLatch(1);// 确保服务节点存在ensurePathExists(SERVICE_PATH);}public List<String> getServices() throws KeeperException, InterruptedException {// 获取服务节点的子节点列表，并等待服务更新List<String> children = zooKeeper.getChildren(SERVICE_PATH, true);serviceSignal.await();return currentServices;}private void ensurePathExists(String path) throws KeeperException, InterruptedException {// 确保路径存在，如果不存在则创建持久节点if (zooKeeper.exists(path, false) == null) {zooKeeper.create(path, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);}}@Overridepublic void process(WatchedEvent watchedEvent) {if (watchedEvent.getType() == Event.EventType.NodeChildrenChanged && watchedEvent.getPath().equals(SERVICE_PATH)) {try {// 当服务节点的子节点发生变化时，获取最新的服务列表，并唤醒等待线程List<String> children = zooKeeper.getChildren(SERVICE_PATH, true);currentServices = children;serviceSignal.countDown();} catch (KeeperException | InterruptedException e) {e.printStackTrace();}}}
}

以上是对示例代码的详细注释，希望能够帮助您理解代码的功能和使用方式。