ZooKeeper分布式协调实战指南

ZooKeeper 分布式协调实战指南

分布式系统需要"共识"：谁是主节点？哪些节点活着？配置怎么统一？ZooKeeper 就是解决这些协调问题的基石组件（Kafka、Dubbo、HBase 都依赖它），也是 CSDN 中间件板块的经典热门话题。本文从协调问题讲起，覆盖 ZK 数据模型、ZAB 协议、Watch 机制、集群部署、Java 实战（Curator）、典型应用（注册中心/分布式锁/配置中心）、常见坑，全程带可运行命令与代码。

一、分布式协调解决什么问题

1.1 分布式系统的三大问题

1. 选主：集群里谁是 Master？（Kafka 谁当 Leader）
2. 成员管理：哪些节点活着？（动态上下线）
3. 配置共享：多台机器统一配置？（改一处全部生效）

1.2 ZooKeeper 是什么

ZooKeeper = 分布式协调服务
– 提供"分布式文件系统"式数据存储（ZNode 树）
– 提供"分布式锁/选主/配置管理"原语
– 核心：分布式一致性（ZAB 协议）

类似"分布式操作系统的注册表"

用 ZK 的经典组件：

Kafka：Broker 注册、Controller 选主
Dubbo：服务注册与发现
HBase：HMaster 选主、Region 分配
分布式锁：Redis 之外的另一选择

二、数据模型（ZNode）

2.1 树形结构

ZooKeeper 数据 = 一棵树：
/zookeeper
├── /apps
│ ├── /order-service （服务节点）
│ │ └── /10.0.0.1:8080
│ └── /config
│ ├── /db-url （配置节点）
│ └── /thread-pool

ZNode（节点）：

• 每个节点存数据（最大 1MB，存小配置）；
• 有路径（类似文件系统）；
• 有版本号（CAS 更新）。

2.2 节点类型（必考）

# 创建节点（命令行）
zkCli.sh
# 持久节点
create /apps/order-service "10.0.0.1:8080"
# 临时节点（-e）
create -e /apps/order-service/ephemeral "10.0.0.1:8080"
# 顺序节点（-s）
create -s /locks/lock "owner"
# 结果：/locks/lock0000000001

# 查看
ls /apps
get /apps/order-service

# 删除
delete /apps/order-service

2.3 会话（Session）

客户端连接 ZK 建立会话：
– 会话超时（sessionTimeout，默认 10-30 秒）
– 心跳维持（ping）
– 临时节点随会话消失

三、ZAB 协议（一致性核心）

3.1 为什么需要 ZAB

多台 ZK 组成集群，客户端读写要一致：
– 写请求到哪台？怎么同步？
– Leader 挂了怎么办？
ZAB（原子广播协议）解决

3.2 ZAB 两阶段

1. 选主（Leader Election）：
– 集群启动/Leader 挂 → 重新选主
– 多数派投票（2n+1 台，挂 n 台仍可用）
– 最新数据（事务 ID 最大）者优先

2. 原子广播（写请求）：
Leader 接收写 → 广播给 Follower
→ 超过半数确认 → 提交 → 通知全部
→ 保证所有节点数据一致

3.3 集群可用性

集群节点数必须是奇数（3、5、7）：
3 台：最多挂 1 台（2 台可用）
5 台：最多挂 2 台
原因：过半机制需要"多数"

脑裂问题：网络分区 → 只有多数派能提供服务

四、Watch 机制（事件通知）

4.1 什么是 Watch

客户端监听节点变化 → 节点数据/子节点/状态变化 → 通知客户端

一次性：触发一次后失效，需要重新注册

监听 /apps/order-service/10.0.0.1:8080
→ 节点消失（服务下线）→ 客户端收到事件 → 更新本地列表

4.2 Watch 类型

exists：节点是否存在（创建/删除触发）
getData：节点数据变化
getChildren：子节点变化

事件类型：
NodeCreated / NodeDeleted / NodeDataChanged / NodeChildrenChanged

4.3 为什么重要

Watch 让客户端"感知变化"而不是"轮询查询"：
服务发现：新服务上线自动知道（不用每 5 秒查一次）
配置中心：配置改了自动知道（不用重启）

五、集群部署

5.1 部署 3 节点

# 三台机器（或一台三个端口）
# 配置 zoo.cfg（每台）
tickTime=2000
initLimit=10
syncLimit=5
dataDir=/data/zookeeper
clientPort=2181

# 集群节点（server.1/2/3 与 myid 对应）
server.1=192.168.1.11:2888:3888
server.2=192.168.1.12:2888:3888
server.3=192.168.1.13:2888:3888

# 每台写 myid（1、2、3）
echo "1" > /data/zookeeper/myid

# 启动
zkServer.sh start

# 查看状态（哪个是 Leader）
zkServer.sh status
# Mode: leader / follower

# 客户端连接
zkCli.sh -server 192.168.1.11:2181,192.168.1.12:2181

5.2 Docker 部署

# docker-compose.yml（3 节点）
version: '3'
services:
zk1:
image: zookeeper:3.8
ports: ["2181:2181"]
environment:
ZOO_MY_ID: 1
ZOO_SERVERS: server.1=zk1:2888:3888;server.2=zk2:2888:3888;server.3=zk3:2888:3888
zk2:
image: zookeeper:3.8
ports: ["2182:2181"]
environment:
ZOO_MY_ID: 2
ZOO_SERVERS: server.1=zk1:2888:3888;server.2=zk2:2888:3888;server.3=zk3:2888:3888
zk3:
image: zookeeper:3.8
ports: ["2183:2181"]
environment:
ZOO_MY_ID: 3
ZOO_SERVERS: server.1=zk1:2888:3888;server.2=zk2:2888:3888;server.3=zk3:2888:3888

六、Java 实战（Curator）

6.1 依赖

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

6.2 客户端初始化

@Configuration
public class ZkConfig {

@Bean(destroyMethod = "close")
public CuratorFramework curatorFramework() {
CuratorFramework client = CuratorFrameworkFactory
.builder()
.connectString("192.168.1.11:2181,192.168.1.12:2181,192.168.1.13:2181")
.sessionTimeoutMs(30000)
.connectionTimeoutMs(10000)
// 重试策略：最多 3 次，间隔 1 秒
.retryPolicy(new ExponentialBackoffRetry(1000, 3))
.build();
client.start();
return client;
}
}

6.3 基础操作

@Service
public class ConfigService {

@Autowired
private CuratorFramework client;

// 写配置（持久节点）
public void setConfig(String key, String value) throws Exception {
String path = "/config/" + key;
// 不存在则创建（parents 自动建父节点）
if (client.checkExists().forPath(path) == null) {
client.create().creatingParentsIfNeeded()
.forPath(path, value.getBytes());
} else {
client.setData().forPath(path, value.getBytes());
}
}

// 读配置
public String getConfig(String key) throws Exception {
String path = "/config/" + key;
byte[] data = client.getData().forPath(path);
return data == null ? null : new String(data);
}

// 监听配置变化（Watch）
public void watchConfig(String key) throws Exception {
String path = "/config/" + key;
client.getData().usingWatcher(new Watcher() {
@Override
public void process(WatchedEvent event) {
System.out.println("配置变化: " + event.getType());
// 重新读取 + 重新注册监听（一次性）
try {
byte[] data = client.getData()
.usingWatcher(this) // 重新监听
.forPath(path);
refreshLocalCache(new String(data));
} catch (Exception e) {
e.printStackTrace();
}
}
}).forPath(path);
}
}

七、典型应用一：服务注册与发现

// 服务启动时注册（临时节点：挂了自动消失）
@Component
public class ServiceRegistry {

@Autowired
private CuratorFramework client;

@Value("${server.port}")
private int port;

@PostConstruct
public void register() throws Exception {
String appName = "order-service";
String instance = InetAddress.getLocalHost()
.getHostAddress() + ":" + port;
String path = "/services/" + appName + "/" + instance;

// 临时节点（客户端断开自动删除）
client.create().creatingParentsIfNeeded()
.withMode(CreateMode.EPHEMERAL)
.forPath(path, "http://" + instance.getBytes());
System.out.println("注册服务: " + path);
}
}

// 服务发现：拿到所有实例
@Component
public class ServiceDiscovery {

@Autowired
private CuratorFramework client;

public List<String> getInstances(String appName) throws Exception {
String path = "/services/" + appName;
return client.getChildren().forPath(path);
}

// 监听服务变化（新服务上线自动知道）
public void watchInstances(String appName, Runnable onChanged) {
String path = "/services/" + appName;
try {
client.getChildren().usingWatcher(new Watcher() {
@Override
public void process(WatchedEvent event) {
onChanged.run();
}
}).forPath(path);
} catch (Exception e) {
e.printStackTrace();
}
}
}

八、典型应用二：分布式锁

8.1 原理（临时顺序节点）

1. 在 /locks 下创建临时顺序节点 lock_0001
2. 判断自己是不是最小的节点：
是 → 拿到锁
否 → 监听前一个节点，等待它删除
3. 释放：删除自己的节点（或会话断开自动删）

特点：公平锁（先来先得）、天然防死锁（临时节点）。

8.2 Curator 实现（生产直接用）

@Service
public class ZkLockService {

@Autowired
private CuratorFramework client;

public void executeWithLock(String lockKey, Runnable task) {
// 可重入互斥锁
InterProcessMutex lock =
new InterProcessMutex(client, "/locks/" + lockKey);

try {
// 3 秒内获取，拿不到抛超时
if (lock.acquire(3, TimeUnit.SECONDS)) {
try {
task.run();
} finally {
lock.release();
}
} else {
throw new BusinessException("获取锁超时");
}
} catch (Exception e) {
throw new RuntimeException(e);
}
}
}

// 使用：扣库存
public void deductStock(Long productId) {
lockService.executeWithLock("stock:" + productId, () -> {
int stock = productMapper.selectStock(productId);
if (stock <= 0) throw new BusinessException("库存不足");
productMapper.decrement(productId);
});
}

8.3 ZK 锁 vs Redis 锁（回顾）

九、典型应用三：配置中心

多台服务读同一份配置 → 放 ZK 持久节点
改配置 → 通知所有监听的服务 → 动态生效（不重启）

// 配置监听（变化自动刷新本地缓存）
@Component
public class ConfigListener implements ApplicationRunner {

@Autowired
private CuratorFramework client;

@Override
public void run(ApplicationArguments args) throws Exception {
String path = "/config/db-url";
// 初次加载 + 监听
client.getData().usingWatcher(new Watcher() {
@Override
public void process(WatchedEvent event) {
if (event.getType() == Watcher.Event.EventType.NodeDataChanged) {
try {
byte[] data = client.getData()
.usingWatcher(this)
.forPath(path);
// 刷新数据源配置（热更新）
dataSourceConfig.refresh(new String(data));
} catch (Exception e) { }
}
}
}).forPath(path);
}
}

十、常见坑速查

本章小结

ZooKeeper 的实战主线：数据模型（ZNode 树）→ ZAB 协议（选主+原子广播）→ Watch（变化感知）→ 三大应用（注册中心/分布式锁/配置中心）。核心心法：临时节点管"生命周期"（服务上下线）、顺序节点管"排队公平"（分布式锁）、Watch 管"变化感知"（配置/发现）——这三个机制就是 ZK 的全部精髓。Kafka、Dubbo 都在 ZK 之上构建，理解了 ZK，这些组件的"注册、选主、协调"就都通了。

三篇 CSDN 热门主题文档全部完成：XXL-Job、SkyWalking、ZooKeeper——覆盖任务调度、链路追踪、分布式协调三大方向。

（全文完）