ZooKeeper 是 Apache 的分布式协调服务，为分布式应用提供一致性数据存储、配置管理、命名服务、分布式锁和集群管理。
1. 核心特性
| 特性 | 说明 |
|---|---|
| 顺序一致性 | 客户端的更新按发送顺序执行 |
| 原子性 | 更新操作要么全部成功，要么全部失败 |
| 单一视图 | 所有客户端看到相同的数据视图 |
| 可靠性 | 写入一旦完成，数据将持续有效 |
| 实时性 | 在一定时间内，客户端能读到最新数据 |
2. ZAB 协议 (ZooKeeper Atomic Broadcast)
ZAB 是 ZooKeeper 的一致性协议，类似 Raft 但专为协调服务设计。
2.1 协议阶段
阶段 1: 崩溃恢复（Leader 选举）
- 选举拥有最新数据的节点为 Leader
- Leader 同步数据给 Follower
阶段 2: 消息广播（正常运行）
- 所有写请求转发给 Leader
- Leader 生成 ZXID，广播给 Follower
- 过半 Follower ACK 后提交
2.2 ZXID 结构
ZXID: 64 位
高 32 位: epoch（Leader 轮次）
低 32 位: 事务计数（单调递增）
3. 数据模型
ZooKeeper 提供类文件系统的树形命名空间：
/
├── zookeeper # 内置节点
│ └── config
├── services # 服务注册
│ ├── user-service
│ │ ├── node-001 [10.0.1.1:8080]
│ │ └── node-002 [10.0.1.2:8080]
│ └── order-service
├── config # 配置管理
│ └── database
│ └── url [jdbc:mysql://...]
└── locks # 分布式锁
└── order-123
节点类型：
| 类型 | 说明 |
|---|---|
| Persistent | 持久节点，客户端断开后仍存在 |
| Ephemeral | 临时节点，会话结束自动删除（用于服务注册） |
| Sequential | 顺序节点，自动附加递增序号 |
4. Watcher 机制
// 一次性 Watch
zk.getData("/config/database", new Watcher() {
@Override
public void process(WatchedEvent event) {
if (event.getType() == EventType.NodeDataChanged) {
// 重新读取并重新注册 Watch
byte[] data = zk.getData("/config/database", this, null);
System.out.println("配置更新: " + new String(data));
}
}
}, null);
// Curator 框架的持久 Watch
curator.watchers().add().withMode(AddWatchMode.PERSISTENT)
.usingWatcher((CuratorWatcher) event -> {
System.out.println("事件: " + event);
}).forPath("/services");
注意：Watch 是一次性的，触发后需重新注册。
5. 分布式锁
5.1 排他锁
// Curator 互斥锁
InterProcessMutex lock = new InterProcessMutex(curator, "/locks/resource-1");
try {
if (lock.acquire(10, TimeUnit.SECONDS)) {
// 执行业务逻辑
}
} finally {
lock.release();
}
5.2 共享锁（读写锁）
InterProcessReadWriteLock rwlock = new InterProcessReadWriteLock(
curator, "/locks/rw-resource");
// 读锁
InterProcessMutex readLock = rwlock.readLock();
readLock.acquire();
// 写锁
InterProcessMutex writeLock = rwlock.writeLock();
writeLock.acquire();
5.3 羊群效应优化
原始方案：所有客户端 Watch 父节点，导致大量通知。
优化方案：每个客户端只 Watch 前一个顺序节点。
public class ImprovedDistributedLock {
public void lock() throws Exception {
// 1. 创建临时顺序节点
String node = curator.create()
.withMode(CreateMode.EPHEMERAL_SEQUENTIAL)
.forPath("/locks/mylock/lock-");
// 2. 获取所有子节点
List<String> nodes = curator.getChildren().forPath("/locks/mylock");
Collections.sort(nodes);
// 3. 如果是第一个，获取锁
if (node.endsWith(nodes.get(0))) {
return;
}
// 4. 找到前一个节点，只 Watch 它
int index = nodes.indexOf(node.substring(node.lastIndexOf('/') + 1));
String prevNode = nodes.get(index - 1);
CountDownLatch latch = new CountDownLatch(1);
curator.watchers().add()
.usingWatcher((org.apache.zookeeper.Watcher) event -> {
if (event.getType() == Event.EventType.NodeDeleted) {
latch.countDown();
}
}).forPath("/locks/mylock/" + prevNode);
// 5. 等待前一个节点删除
latch.await();
}
}
6. Leader 选举
// Curator Leader 选举
LeaderSelector selector = new LeaderSelector(curator, "/election/leader",
new LeaderSelectorListenerAdapter() {
@Override
public void takeLeadership(CuratorFramework client) throws Exception {
System.out.println("成为 Leader！");
// 保持领导地位（此方法不返回则保持领导权）
Thread.sleep(Long.MAX_VALUE);
}
});
selector.autoRequeue();
selector.start();
7. 配置管理
// 配置中心
public class ZkConfigCenter {
private static final String CONFIG_PATH = "/config/app";
private final Map<String, String> configCache = new ConcurrentHashMap<>();
public void init() throws Exception {
// 加载配置
List<String> keys = curator.getChildren().forPath(CONFIG_PATH);
for (String key : keys) {
byte[] data = curator.getData().forPath(CONFIG_PATH + "/" + key);
configCache.put(key, new String(data));
}
// 监听变更
curator.watchers().add().withMode(AddWatchMode.PERSISTENT_RECURSIVE)
.usingWatcher(event -> reloadConfig())
.forPath(CONFIG_PATH);
}
}
8. 与 etcd 对比
| 特性 | ZooKeeper | etcd |
|---|---|---|
| 协议 | ZAB | Raft |
| API | 专用客户端 | gRPC + HTTP |
| Watch | 一次性 | 长期流式 |
| 事务 | 多版本并发 | 简单事务 |
| K8s | 旧版用 | 原生集成 |
| 性能 | 万级 TPS | 万级 TPS |
| 生态 | Hadoop/HBase/Kafka | K8s/CoreDNS/TiKV |
总结
ZooKeeper 虽然年代久远，但在 Hadoop 生态中不可替代。新项目如需分布式协调，更推荐 etcd（K8s 原生、API 更现代）。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。