文章标签： #zookeeper #分布式 #协调服务 #zab #一致性协议 #分布式锁 #面试
- 引言：Zookeeper的本质
- 理论基础：为什么需要分布式协调
- 演进史：从Chubby到Zookeeper
- 核心原理深度解析
- 实战案例：工业级应用
- 对比分析：Zookeeper vs etcd vs Consul
- 性能分析：吞吐与延迟优化
- 常见陷阱与最佳实践
- 面试题与参考答案
Zookeeper不是简单的"分布式配置中心"或"注册中心"，而是一个分布式协调内核（Distributed Coordination Kernel）。
核心认知：
单机系统的协调：通过共享内存、锁、信号量等OS原语
分布式系统的协调：通过网络通信达成一致性，面临：
- 网络分区（Network Partition）
- 节点故障（Node Failure）
- 消息延迟/丢失（Message Delay/Loss）
- 时钟不同步（Clock Skew）
Zookeeper的本质：
在不可靠的分布式环境中，提供可靠的协调原语：
- 统一命名（Naming）
- 配置管理（Configuration Management）
- 组成员管理（Group Membership）
- 分布式锁（Distributed Locks）
- 领导者选举（Leader Election）
关键洞察：Zookeeper的设计哲学是在CAP中优先保证CP（一致性和分区容错性），通过牺牲部分可用性（写入时Leader选举期间不可用）来换取强一致性保证。
CAP定理（Brewer, 2000）：
在分布式系统中，Consistency（一致性）、Availability（可用性）、Partition Tolerance（分区容错性）三者不可兼得。
Zookeeper的选择：CP（Consistency + Partition Tolerance）
原因分析：
- 协调服务的数据一致性是核心诉求
- 短暂的不可用（Leader选举期间）是可接受的
- 数据不一致（如脑裂）是不可接受的
Zookeeper的可用性保证：
- 读操作：任何节点都可处理，高可用
- 写操作：Leader处理，Leader故障时短暂不可用
- 集群半数以上存活即可对外服务（读）
Zookeeper提供顺序一致性（Sequential Consistency），而非线性一致性（Linearizability）：
顺序一致性的含义：
- 所有客户端看到的操作顺序与全局顺序一致
- 来自同一客户端的操作按发送顺序执行
- 不同客户端的操作顺序可能交错，但全局一致
示例：
客户端A：set /x 1 → set /x 2
客户端B：get /x
保证：
- 客户端A的两次set按顺序执行
- 客户端B可能读到1或2，但不会读到不一致的状态
问题1：两将军问题（Two Generals Problem）
- 通信不可靠时，无法达成确定性共识
- Zookeeper的解决：基于多数派的投票机制
问题2：拜占庭将军问题（Byzantine Generals）
- 节点可能发送错误信息（恶意或故障）
- Zookeeper的解决：假设故障停止（Fail-Stop）模型，非拜占庭故障
问题3：脑裂（Split-Brain）
- 网络分区导致多个节点认为自己是Leader
- Zookeeper的解决：Quorum机制，只有获得半数以上投票的节点才能成为Leader
Paxos（Lamport, 1989）：
- 角色：Proposer、Acceptor、Learner
- 阶段：Prepare -> Promise -> Accept -> Accepted
- 优点：理论严谨，安全性保证强
- 缺点：复杂难懂，工程实现困难，活锁问题
ZAB（Zookeeper Atomic Broadcast）：
- 角色：Leader、Follower、Observer
- 阶段：崩溃恢复 -> 消息广播
- 优点：专为Zookeeper设计，工程实现简单，性能更好
- 缺点：相比Paxos，理论证明较少
关键区别：
- Paxos是通用共识协议
- ZAB是专为主备复制设计的原子广播协议
- ZAB保证消息的全局有序（FIFO）
背景：
- Google内部需要分布式锁服务
- 基于Paxos算法实现
- 提供粗粒度锁（Coarse-Grained Locking）
设计特点：
- 基于文件系统的接口
- 提供建议性锁（Advisory Locks）
- 事件通知机制
局限性：
- 未开源
- 性能受限于Paxos的复杂度
- 仅适用于Google内部生态
背景：
- Yahoo!为Hadoop生态开发
- 受Chubby启发，但使用更简单的ZAB协议
- 2008年成为Apache顶级项目
设计目标：
- 简单易用：类似文件系统的API
- 高性能：每秒数万次操作
- 可靠：支持集群部署，自动故障恢复
- 有序：保证操作的全局顺序
Zookeeper 3.3.x（2012）：
- 引入Observer节点
- 提高读性能，不影响写投票
Zookeeper 3.4.x（2013）：
- 引入Netty作为NIO框架
- 支持SSL通信
- 改进Leader选举算法
Zookeeper 3.5.x（2019）：
- 动态重新配置（Dynamic Reconfiguration）
- 无需停机即可增删节点
- 改进的监控和管理接口
Zookeeper 3.6.x（2020）：
- 支持TLS 1.3
- 改进的审计日志
- 新的监控指标
Zookeeper 3.7.x（2021）：
- 支持Java 11
- 改进的Snapshot传输
- 新的Admin命令
Zookeeper 3.8.x（2022）：
- 支持Java 17
- 性能优化：更快的Leader选举
- 改进的内存管理
当前状态（2026）：
- 仍然是Hadoop/Kafka/HBase等生态的核心依赖
- 面临etcd在Kubernetes领域的竞争
- 在云原生场景中被逐步替代
- 但在传统大数据领域仍不可替代
Zookeeper命名空间：
/
├── /zookeeper # 内置节点，存储配额信息
│ └── /zookeeper/quota
├── /app1 # 应用1的命名空间
│ ├── /app1/config # 配置节点
│ ├── /app1/leader # 领导者节点
│ └── /app1/workers # 工作者节点
│ ├── /app1/workers/worker-0000000001
│ └── /app1/workers/worker-0000000002
├── /app2
│ └── /app2/locks # 分布式锁
│ ├── /app2/locks/lock-0000000001
│ └── /app2/locks/lock-0000000002
└── /services # 服务注册中心
├── /services/user-service
│ ├── /services/user-service/192.168.1.1:8080
│ └── /services/user-service/192.168.1.2:8080
└── /services/order-service
└── /services/order-service/192.168.1.3:8080
/**
* ZNode类型枚举
*/
public enum CreateMode {
/**
* 持久节点
* 特点：客户端断开连接后仍然存在
* 用途：配置信息、元数据、持久化数据
*/
PERSISTENT(0, false, false, false, false),
/**
* 持久顺序节点
* 特点：持久 + 自动编号（10位数字）
* 用途：分布式队列、顺序任务调度
*/
PERSISTENT_SEQUENTIAL(2, false, true, false, false),
/**
* 临时节点
* 特点：客户端会话结束自动删除
* 用途：服务注册、心跳检测、分布式锁
*/
EPHEMERAL(1, true, false, false, false),
/**
* 临时顺序节点
* 特点：临时 + 自动编号
* 用途：Master选举、分布式公平锁
*/
EPHEMERAL_SEQUENTIAL(3, true, true, false, false),
/**
* 容器节点（3.5.3+）
* 特点：当最后一个子节点删除时自动删除
* 用途：Leader选举、TTL管理
*/
CONTAINER(4, false, false, false, true),
/**
* 持久TTL节点（3.5.3+）
* 特点：持久但有过期时间
* 用途：带过期时间的配置
*/
PERSISTENT_WITH_TTL(5, false, false, true, false),
/**
* 持久顺序TTL节点（3.5.3+）
* 特点：持久顺序 + TTL
*/
PERSISTENT_SEQUENTIAL_WITH_TTL(6, false, true, true, false);
}/**
* ZNode的元数据结构
*/
public class Stat {
private long czxid; // 创建事务ID（Create ZXID）
private long mzxid; // 最后修改事务ID（Modified ZXID）
private long ctime; // 创建时间（毫秒）
private long mtime; // 最后修改时间（毫秒）
private int version; // 数据版本号（乐观锁）
private int cversion; // 子节点版本号
private int aversion; // ACL版本号
private long ephemeralOwner; // 临时节点所属会话ID
private int dataLength; // 数据长度（字节）
private int numChildren; // 子节点数量
private long pzxid; // 最后修改子节点的事务ID
// 版本号机制实现乐观锁
// setData(path, data, version) 如果version不匹配则更新失败
}Watcher架构：
客户端 Zookeeper Server
| |
|---- 1. getData("/app1", watcher) --->|
| |
|<--- 2. 返回数据，注册Watcher ----|
| |
| | 3. 其他客户端修改 /app1
| |
|<--- 4. 发送Watcher事件通知 ----|
| |
|---- 5. 重新注册Watcher（一次性）--->|
/**
* Watcher事件类型
*/
public interface Watcher {
void process(WatchedEvent event);
}
/**
* 事件类型（EventType）
*/
public enum EventType {
None(-1), // 连接状态变化
NodeCreated(1), // 节点创建
NodeDeleted(2), // 节点删除
NodeDataChanged(3), // 节点数据变化
NodeChildrenChanged(4), // 子节点列表变化
DataWatchRemoved(5), // 数据Watcher被移除（3.5+）
ChildWatchRemoved(6), // 子节点Watcher被移除（3.5+）
PersistentWatchRemoved(7); // 持久Watcher被移除（3.5+）
}
/**
* 连接状态（KeeperState）
*/
public enum KeeperState {
Unknown(-1),
Disconnected(0), // 连接断开
NoSyncConnected(1),
SyncConnected(3), // 连接成功
AuthFailed(4), // 认证失败
ConnectedReadOnly(5), // 只读连接（Observer）
SaslAuthenticated(6), // SASL认证成功
Expired(-112); // 会话过期
}/**
* Watcher使用示例
*/
public class WatcherExample {
private ZooKeeper zk;
public void init() throws IOException {
// 创建连接，设置默认Watcher（连接状态变化）
zk = new ZooKeeper("localhost:2181", 3000, event -> {
if (event.getState() == Watcher.Event.KeeperState.SyncConnected) {
System.out.println("连接成功");
} else if (event.getState() == Watcher.Event.KeeperState.Disconnected) {
System.out.println("连接断开");
} else if (event.getState() == Watcher.Event.KeeperState.Expired) {
System.out.println("会话过期，需要重新创建连接");
}
});
}
/**
* 监听节点数据变化（一次性）
*/
public void watchData(String path) throws KeeperException, InterruptedException {
zk.getData(path, event -> {
if (event.getType() == Event.EventType.NodeDataChanged) {
System.out.println("节点数据变化: " + event.getPath());
try {
// 重新获取数据并注册Watcher
byte[] data = zk.getData(path, this::watchData, null);
System.out.println("新数据: " + new String(data));
} catch (Exception e) {
e.printStackTrace();
}
}
}, null);
}
/**
* 监听子节点变化（一次性）
*/
public void watchChildren(String path) throws KeeperException, InterruptedException {
zk.getChildren(path, event -> {
if (event.getType() == Event.EventType.NodeChildrenChanged) {
System.out.println("子节点列表变化: " + event.getPath());
try {
// 重新获取子节点列表并注册Watcher
List<String> children = zk.getChildren(path, this::watchChildren);
System.out.println("当前子节点: " + children);
} catch (Exception e) {
e.printStackTrace();
}
}
});
}
}Watcher的三大特点：
1. 一次性触发（One-time Trigger）：
- Watcher触发后即被移除
- 需要业务代码中重新注册
- 设计原因：避免Watcher累积导致内存泄漏
2. 轻量级通知：
- 只通知事件类型，不传输具体数据
- 客户端需要主动查询最新数据
- 设计原因：减少网络流量
3. 异步发送：
- Watcher事件异步发送到客户端
- 不保证与操作的原子性
- 可能丢失事件（触发了但客户端还没重新注册）
Watcher的限制：
- 不能监听所有子节点的数据变化（只能监听子节点列表变化）
- 不能监听孙节点变化
- 会话过期后所有Watcher失效
- 网络分区期间可能丢失事件
ZAB协议架构：
┌─────────────────────────────────────────┐
│ ZAB Protocol │
├─────────────────────────────────────────┤
│ Phase 1: Discovery（发现） │
│ - 节点交换信息，确定epoch │
├─────────────────────────────────────────┤
│ Phase 2: Synchronization（同步） │
│ - Leader同步数据给Follower │
├─────────────────────────────────────────┤
│ Phase 3: Broadcast（广播） │
│ - 正常处理客户端请求 │
└─────────────────────────────────────────┘
两种模式：
1. 崩溃恢复（Crash Recovery）：Leader宕机或集群启动时
2. 消息广播（Message Broadcast）：正常工作模式
客户端写请求处理流程：
Client Leader Follower 1 Follower 2
| | | |
|-- 1. 写请求 ->| | |
| | | |
| |-- 2. Proposal -->| |
| |-- 2. Proposal ------->| |
| | | |
| |<- 3. ACK ------| |
| |<- 3. ACK ------------| |
| | | |
| |-- 4. Commit --->| |
| |-- 4. Commit ------->| |
| | | |
|<- 5. 响应 ----| | |
关键约束：
- Leader必须收到半数以上ACK才发送Commit
- Follower收到Commit后才真正应用事务
- 保证消息的全局有序和原子广播
ZXID是64位整数，结构如下：
┌──────────────────────┬──────────────────────┐
│ 高32位：epoch │ 低32位：计数器 │
│ (Leader周期编号) │ (事务序号) │
└──────────────────────┴──────────────────────┘
示例：0x0000000100000002
- epoch = 1（第1个Leader周期）
- 计数器 = 2（该周期的第2个事务）
ZXID的作用：
1. 标识事务的全局顺序
2. Leader选举时比较数据新旧
3. 崩溃恢复时确定同步点
/**
* ZAB协议核心流程伪代码
*/
public class ZABProtocol {
/**
* Leader处理写请求
*/
public void handleWriteRequest(Request request) {
// 1. 生成zxid
long zxid = generateZxid();
request.setZxid(zxid);
// 2. 写入本地日志（WAL）
log.append(request);
// 3. 发送Proposal给所有Follower
Proposal proposal = new Proposal(zxid, request);
for (Follower f : followers) {
f.send(proposal);
}
// 4. 等待半数以上ACK
int ackCount = 1; // 自己算一票
for (Follower f : followers) {
if (f.waitForAck(zxid, timeout)) {
ackCount++;
}
if (ackCount > followers.size() / 2) {
break;
}
}
// 5. 发送Commit
if (ackCount > followers.size() / 2) {
commit(zxid);
for (Follower f : followers) {
f.sendCommit(zxid);
}
}
}
/**
* Follower处理Proposal
*/
public void handleProposal(Proposal proposal) {
// 1. 写入本地日志
log.append(proposal);
// 2. 发送ACK
sendAck(proposal.getZxid());
}
/**
* Follower处理Commit
*/
public void handleCommit(long zxid) {
// 应用事务到内存数据树
applyTransaction(zxid);
}
}Leader选举触发条件：
1. 集群启动时
2. Leader宕机时
3. 网络分区恢复时
4. 动态重新配置时
选举状态：
- LOOKING：寻找Leader
- FOLLOWING：跟随者
- LEADING：领导者
- OBSERVING：观察者
/**
* Leader选举投票比较逻辑
*/
public class Vote {
private long sid; // 服务器ID（myid）
private long zxid; // 最新事务ID
private long epoch; // 逻辑时钟（选举轮次）
/**
* 比较两个投票，返回更优的
*
* 优先级：
* 1. epoch大的优先（更新选举轮次）
* 2. zxid大的优先（数据更新）
* 3. sid大的优先（服务器ID大）
*/
public static Vote compare(Vote v1, Vote v2) {
if (v1.epoch > v2.epoch) return v1;
if (v1.epoch < v2.epoch) return v2;
if (v1.zxid > v2.zxid) return v1;
if (v1.zxid < v2.zxid) return v2;
if (v1.sid > v2.sid) return v1;
return v2;
}
}集群启动时Leader选举流程（5节点为例）：
Server 1 (myid=1): Server 2 (myid=2): Server 3 (myid=3):
初始投票：(1, 0, 1) 初始投票：(2, 0, 1) 初始投票：(3, 0, 1)
| | |
|--- 广播投票 ----------->| |
|<--- 广播投票 -----------| |
| |--- 广播投票 ----------->|
|<--- 广播投票 -----------------------------------|
| |<--- 广播投票 ----------|
| | |
收到投票： 收到投票： 收到投票：
(2,0,1), (3,0,1) (1,0,1), (3,0,1) (1,0,1), (2,0,1)
| | |
更新投票： 更新投票： 更新投票：
(3,0,1) 最优 (3,0,1) 最优 (3,0,1) 最优
| | |
广播新投票 广播新投票 广播新投票
| | |
最终：所有节点投票给Server 3，Server 3获得3票（>5/2=2.5），成为Leader
会话生命周期：
┌─────────┐ connect ┌─────────┐ 心跳/操作 ┌─────────┐
│ 未连接 │ -------------> │ 已连接 │ -------------> │ 已连接 │
└─────────┘ └─────────┘ └─────────┘
| |
| 心跳超时/网络断开 |
v |
┌─────────┐ |
│ 已断开 │ |
└─────────┘ |
| |
| 超时未重连 |
v |
┌─────────┐ |
│ 已过期 │<--------------|
└─────────┘ 重连成功（在超时窗口内）
/**
* 会话配置
*/
public class SessionConfig {
// tickTime: 基本时间单位（毫秒），默认3000ms
private int tickTime = 3000;
// 会话超时时间范围
// minSessionTimeout = tickTime * 2
// maxSessionTimeout = tickTime * 20
// 客户端设置的超时时间必须在此范围内
// 心跳间隔：tickTime / 3 = 1000ms（默认）
// 即每1秒发送一次心跳
}/**
* ACL权限控制
*/
public class ACLExample {
/**
* 内置权限方案
*/
public void builtInSchemes() {
// world: 任何人
ACL worldAcl = new ACL(ZooDefs.Perms.ALL,
new Id("world", "anyone"));
// auth: 已认证用户
ACL authAcl = new ACL(ZooDefs.Perms.ALL,
new Id("auth", ""));
// digest: 用户名密码（SHA1加密）
ACL digestAcl = new ACL(ZooDefs.Perms.ALL,
new Id("digest", "user:password_digest"));
// ip: IP地址限制
ACL ipAcl = new ACL(ZooDefs.Perms.READ,
new Id("ip", "192.168.1.0/24"));
// super: 超级用户（配置文件中指定）
}
/**
* 权限位（Permission Bits）
*/
public void permissionBits() {
int CREATE = 1; // 创建子节点
int READ = 2; // 读取节点数据/子节点列表
int WRITE = 4; // 修改节点数据
int DELETE = 8; // 删除子节点
int ADMIN = 16; // 设置权限
int ALL = 31; // 所有权限
}
}/**
* 基于Zookeeper的分布式配置中心
*/
public class ZookeeperConfigCenter {
private ZooKeeper zk;
private ConcurrentHashMap<String, String> configCache = new ConcurrentHashMap<>();
private List<ConfigListener> listeners = new CopyOnWriteArrayList<>();
public ZookeeperConfigCenter(String connectString) throws IOException {
zk = new ZooKeeper(connectString, 5000, event -> {
if (event.getState() == Watcher.Event.KeeperState.Expired) {
// 会话过期，重新初始化
reconnect();
}
});
}
/**
* 获取配置（带本地缓存）
*/
public String getConfig(String key) {
// 先查本地缓存
String value = configCache.get(key);
if (value != null) {
return value;
}
// 从Zookeeper获取
try {
String path = "/config/" + key;
byte[] data = zk.getData(path, event -> {
if (event.getType() == Event.EventType.NodeDataChanged) {
// 配置变化，更新缓存
refreshConfig(key);
}
}, null);
value = new String(data, StandardCharsets.UTF_8);
configCache.put(key, value);
return value;
} catch (Exception e) {
throw new RuntimeException("获取配置失败: " + key, e);
}
}
/**
* 更新配置
*/
public void updateConfig(String key, String value) {
try {
String path = "/config/" + key;
Stat stat = zk.exists(path, false);
if (stat == null) {
zk.create(path, value.getBytes(),
ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
} else {
zk.setData(path, value.getBytes(), stat.getVersion());
}
configCache.put(key, value);
} catch (Exception e) {
throw new RuntimeException("更新配置失败: " + key, e);
}
}
/**
* 刷新配置
*/
private void refreshConfig(String key) {
try {
String path = "/config/" + key;
byte[] data = zk.getData(path, event -> refreshConfig(key), null);
String newValue = new String(data, StandardCharsets.UTF_8);
String oldValue = configCache.put(key, newValue);
if (!newValue.equals(oldValue)) {
// 通知监听器
for (ConfigListener listener : listeners) {
listener.onConfigChange(key, oldValue, newValue);
}
}
} catch (Exception e) {
e.printStackTrace();
}
}
public interface ConfigListener {
void onConfigChange(String key, String oldValue, String newValue);
}
}/**
* 基于临时顺序节点的Master选举
*/
public class MasterElection {
private ZooKeeper zk;
private String electionPath = "/election";
private String myNodePath;
private volatile boolean isMaster = false;
private MasterListener listener;
public MasterElection(ZooKeeper zk, String electionPath, MasterListener listener) {
this.zk = zk;
this.electionPath = electionPath;
this.listener = listener;
}
/**
* 参与选举
*/
public void participate() throws KeeperException, InterruptedException {
// 创建临时顺序节点
myNodePath = zk.create(
electionPath + "/node-",
InetAddress.getLocalHost().getHostAddress().getBytes(),
ZooDefs.Ids.OPEN_ACL_UNSAFE,
CreateMode.EPHEMERAL_SEQUENTIAL
);
System.out.println("创建节点: " + myNodePath);
// 检查是否成为Master
checkMaster();
}
/**
* 检查是否成为Master
*/
private void checkMaster() throws KeeperException, InterruptedException {
// 获取所有子节点
List<String> children = zk.getChildren(electionPath, false);
Collections.sort(children);
// 获取自己的节点名
String myNodeName = myNodePath.substring(myNodePath.lastIndexOf("/") + 1);
// 如果自己的节点是第一个，成为Master
if (myNodeName.equals(children.get(0))) {
becomeMaster();
} else {
// 找到自己的位置
int myIndex = children.indexOf(myNodeName);
// 监听前一个节点
String prevNode = children.get(myIndex - 1);
zk.exists(electionPath + "/" + prevNode, event -> {
if (event.getType() == Event.EventType.NodeDeleted) {
try {
checkMaster();
} catch (Exception e) {
e.printStackTrace();
}
}
});
System.out.println("等待前一个节点删除: " + prevNode);
}
}
private void becomeMaster() {
isMaster = true;
System.out.println("成为Master！");
if (listener != null) {
listener.onMaster();
}
}
public boolean isMaster() {
return isMaster;
}
public interface MasterListener {
void onMaster();
}
}/**
* 基于Zookeeper的分布式可重入锁
*/
public class ZookeeperDistributedLock {
private ZooKeeper zk;
private String lockPath;
private String myNodePath;
private ThreadLocal<String> lockedNode = new ThreadLocal<>();
public ZookeeperDistributedLock(ZooKeeper zk, String lockPath) {
this.zk = zk;
this.lockPath = lockPath;
}
/**
* 获取锁（阻塞）
*/
public void lock() throws InterruptedException {
if (lockedNode.get() != null) {
// 可重入
return;
}
try {
// 创建临时顺序节点
myNodePath = zk.create(
lockPath + "/lock-",
new byte[0],
ZooDefs.Ids.OPEN_ACL_UNSAFE,
CreateMode.EPHEMERAL_SEQUENTIAL
);
// 尝试获取锁
while (true) {
List<String> children = zk.getChildren(lockPath, false);
Collections.sort(children);
String myNodeName = myNodePath.substring(myNodePath.lastIndexOf("/") + 1);
int myIndex = children.indexOf(myNodeName);
if (myIndex == 0) {
// 获取锁成功
lockedNode.set(myNodePath);
return;
}
// 监听前一个节点
String prevNode = children.get(myIndex - 1);
CountDownLatch latch = new CountDownLatch(1);
Stat stat = zk.exists(lockPath + "/" + prevNode, event -> {
if (event.getType() == Event.EventType.NodeDeleted) {
latch.countDown();
}
});
if (stat == null) {
// 前一个节点已删除，重试
continue;
}
// 等待前一个节点删除
latch.await();
}
} catch (KeeperException e) {
throw new RuntimeException("获取锁失败", e);
}
}
/**
* 释放锁
*/
public void unlock() {
String node = lockedNode.get();
if (node != null) {
try {
zk.delete(node, -1);
lockedNode.remove();
} catch (Exception e) {
e.printStackTrace();
}
}
}
/**
* 尝试获取锁（非阻塞）
*/
public boolean tryLock(long timeout, TimeUnit unit) {
// 实现略，基于lock()改造
return false;
}
}/**
* 基于Zookeeper的服务注册与发现
*/
public class ServiceRegistry {
private ZooKeeper zk;
private String servicesPath = "/services";
public ServiceRegistry(ZooKeeper zk) {
this.zk = zk;
}
/**
* 注册服务实例
*/
public void register(String serviceName, String host, int port) {
try {
String servicePath = servicesPath + "/" + serviceName;
// 创建服务持久节点
if (zk.exists(servicePath, false) == null) {
zk.create(servicePath, new byte[0],
ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
}
// 创建临时节点作为服务实例
String instanceData = host + ":" + port;
String instancePath = zk.create(
servicePath + "/instance-",
instanceData.getBytes(),
ZooDefs.Ids.OPEN_ACL_UNSAFE,
CreateMode.EPHEMERAL_SEQUENTIAL
);
System.out.println("注册服务: " + serviceName + " -> " + instanceData);
} catch (Exception e) {
throw new RuntimeException("服务注册失败", e);
}
}
/**
* 发现服务实例
*/
public List<ServiceInstance> discover(String serviceName) {
try {
String servicePath = servicesPath + "/" + serviceName;
List<String> children = zk.getChildren(servicePath, false);
List<ServiceInstance> instances = new ArrayList<>();
for (String child : children) {
byte[] data = zk.getData(servicePath + "/" + child, false, null);
String[] parts = new String(data).split(":");
instances.add(new ServiceInstance(parts[0], Integer.parseInt(parts[1])));
}
return instances;
} catch (Exception e) {
throw new RuntimeException("服务发现失败", e);
}
}
/**
* 订阅服务变化
*/
public void subscribe(String serviceName, ServiceListener listener) {
try {
String servicePath = servicesPath + "/" + serviceName;
zk.getChildren(servicePath, event -> {
if (event.getType() == Event.EventType.NodeChildrenChanged) {
List<ServiceInstance> instances = discover(serviceName);
listener.onServiceChange(serviceName, instances);
}
});
} catch (Exception e) {
e.printStackTrace();
}
}
public static class ServiceInstance {
private String host;
private int port;
public ServiceInstance(String host, int port) {
this.host = host;
this.port = port;
}
// getters...
}
public interface ServiceListener {
void onServiceChange(String serviceName, List<ServiceInstance> instances);
}
}/**
* 基于Zookeeper的分布式FIFO队列
*/
public class DistributedQueue {
private ZooKeeper zk;
private String queuePath;
public DistributedQueue(ZooKeeper zk, String queuePath) {
this.zk = zk;
this.queuePath = queuePath;
}
/**
* 入队
*/
public void enqueue(byte[] data) throws KeeperException, InterruptedException {
zk.create(queuePath + "/item-", data,
ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT_SEQUENTIAL);
}
/**
* 出队（阻塞）
*/
public byte[] dequeue() throws KeeperException, InterruptedException {
while (true) {
List<String> children = zk.getChildren(queuePath, false);
if (children.isEmpty()) {
// 队列为空，等待
Thread.sleep(1000);
continue;
}
// 按序号排序
Collections.sort(children);
String firstNode = children.get(0);
try {
byte[] data = zk.getData(queuePath + "/" + firstNode, false, null);
zk.delete(queuePath + "/" + firstNode, -1);
return data;
} catch (KeeperException.NoNodeException e) {
// 节点已被其他消费者删除，重试
}
}
}
}┌─────────────────────────────────────────────────────────────┐
│ 分布式协调服务对比 │
├──────────────┬──────────────┬──────────────┬────────────────┤
│ 特性 │ Zookeeper │ etcd │ Consul │
├──────────────┼──────────────┼──────────────┼────────────────┤
│ 开发方 │ Apache │ CNCF │ HashiCorp │
│ 一致性协议 │ ZAB │ Raft │ Raft │
│ 数据模型 │ 层次化ZNode │ 扁平KV │ 扁平KV │
│ 监听机制 │ Watcher │ Watch │ Watch/Blocking│
│ 多数据中心 │ 不支持 │ 有限支持 │ 原生支持 │
│ 服务发现 │ 基础支持 │ 基础支持 │ 高级支持 │
│ 健康检查 │ 无 │ 无 │ HTTP/TCP/Script│
│ 性能 │ 中等 │ 高 │ 中等 │
│ 客户端库 │ 丰富 │ 丰富 │ 丰富 │
│ 生态 │Hadoop/Kafka │Kubernetes │Spring Cloud │
└──────────────┴──────────────┴──────────────┴────────────────┘
适用场景选择：
Zookeeper：
- 大数据生态（Hadoop, Kafka, HBase）
- 需要强一致性的协调场景
- 已有Zookeeper基础设施
etcd：
- Kubernetes生态
- 需要高性能KV存储
- 云原生应用
Consul：
- 微服务架构
- 需要服务网格（Service Mesh）
- 多数据中心部署
Zookeeper性能基准（典型硬件）：
读操作（Follower）：
- 吞吐量：~50,000 ops/sec
- 延迟：< 1ms
写操作（Leader）：
- 吞吐量：~15,000 ops/sec
- 延迟：2-5ms
Watcher通知：
- 吞吐量：~10,000 events/sec
- 延迟：< 10ms
1. 读优化：
- 使用Observer节点分担读压力
- 客户端缓存（Watcher + 本地缓存）
- 批量读取（getChildren一次性获取）
2. 写优化：
- 批量写入（事务操作multi）
- 减少Watcher数量（使用共享Watcher）
- 合理设置数据大小（< 1MB）
3. 网络优化：
- 使用Netty NIO（3.4+）
- 调整JVM网络缓冲区
- 专用网络（避免与业务流量竞争）
4. 磁盘优化：
- 事务日志和数据快照分离存储
- 使用SSD
- 预分配事务日志文件
# Zookeeper JVM参数
-Xms2g -Xmx2g # 堆内存
-XX:+UseG1GC # G1垃圾回收器
-XX:MaxGCPauseMillis=50 # 最大GC停顿时间
-XX:+UseLargePages # 大页内存
-Xloggc:/var/log/zookeeper/gc.log # GC日志
-XX:+PrintGCDetails # GC详情
-XX:+PrintGCTimeStamps # GC时间戳陷阱： Watcher触发后即被移除，如果业务逻辑中没有重新注册，后续事件将丢失。
最佳实践：
// 在Watcher回调中立即重新注册Watcher
zk.getData(path, event -> {
if (event.getType() == Event.EventType.NodeDataChanged) {
// 处理事件
processDataChange(event.getPath());
// 重新注册
try {
zk.getData(event.getPath(), this, null);
} catch (Exception e) {
e.printStackTrace();
}
}
}, null);
// 或使用Curator框架，它封装了自动重新注册的逻辑
CuratorFramework client = CuratorFrameworkFactory.newClient(
"localhost:2181", new ExponentialBackoffRetry(1000, 3));
client.start();
// NodeCache自动处理Watcher重新注册
NodeCache cache = new NodeCache(client, "/app1/config");
cache.getListenable().addListener(() -> {
byte[] data = cache.getCurrentData().getData();
System.out.println("配置变化: " + new String(data));
});
cache.start();陷阱： Zookeeper集群需要半数以上节点存活，3节点容忍1个故障，4节点也只能容忍1个故障，但4节点写性能更差。
最佳实践：
- 集群节点数应为奇数（3、5、7），quorum效率最高
- 3节点适合大多数场景，5节点适合高可用要求
- 生产环境至少3节点，避免单点故障
节点数与容错能力：
- 1节点：0容错（仅测试环境）
- 3节点：1容错（推荐最小配置）
- 5节点：2容错（生产推荐）
- 7节点：3容错（大规模集群）
陷阱： 会话超时时间太短导致频繁重新连接和临时节点抖动；太长导致故障检测不及时。
最佳实践：
// 会话超时时间一般设置在2-5秒之间
// tickTime的2-20倍
// 客户端配置
ZooKeeper zk = new ZooKeeper("localhost:2181", 5000, watcher);
// 5000ms = 5秒
// 服务端配置（zoo.cfg）
tickTime=2000 # 基本时间单位2秒
initLimit=10 # 初始化连接超时：10 * tickTime = 20秒
syncLimit=5 # 同步超时：5 * tickTime = 10秒
maxClientCnxns=300 # 单IP最大连接数陷阱： 网络分区或长时间GC导致会话过期，临时节点被删除，但客户端仍认为自己持有锁或注册的服务。
最佳实践：
zk = new ZooKeeper(connectString, sessionTimeout, event -> {
if (event.getState() == Watcher.Event.KeeperState.Expired) {
// 会话过期，需要重新初始化所有临时节点
reinitialize();
}
});
private void reinitialize() {
// 1. 关闭旧连接
// 2. 创建新连接
// 3. 重新创建临时节点
// 4. 重新注册Watcher
}陷阱： Zookeeper单个节点默认最大数据量为1MB，超过会导致创建/更新失败。
最佳实践：
- 单个节点数据 < 1MB（默认）
- 大文件存储应使用HDFS/S3等
- Zookeeper只存储元数据和配置
- 如需调整：jute.maxbuffer=4194304（4MB）
答： Zookeeper有四种基本ZNode类型和三种扩展类型：
基本类型：
- 持久节点（Persistent）：客户端断开连接后仍然存在，用于配置信息、元数据
- 临时节点（Ephemeral）：客户端会话结束自动删除，用于服务发现、分布式锁、心跳检测
- 持久顺序节点（Persistent_Sequential）：持久+自动编号（10位数字），用于分布式队列、顺序任务调度
- 临时顺序节点（Ephemeral_Sequential）：临时+自动编号，用于Master选举、分布式公平锁
扩展类型（3.5.3+）： 5. 容器节点（Container）：当最后一个子节点删除时自动删除 6. 持久TTL节点（Persistent_With_TTL）：持久但有过期时间 7. 持久顺序TTL节点（Persistent_Sequential_With_TTL）：持久顺序+TTL
答： Watcher是Zookeeper的事件监听机制，有三个核心特点：
-
一次性触发：Watcher触发后即被移除，需要业务代码中重新注册。设计原因是避免Watcher累积导致内存泄漏。Curator框架提供了自动重新注册的封装（NodeCache、PathChildrenCache等）。
-
轻量级通知：只通知事件类型（NodeCreated/NodeDeleted/NodeDataChanged/NodeChildrenChanged），不传输具体数据。客户端收到通知后需要主动查询最新数据。设计原因是减少网络流量。
-
异步发送：Watcher事件异步发送到客户端，不保证与操作的原子性。在Watcher触发和重新注册之间可能丢失事件。
注意事项：会话过期（Expired）后所有Watcher失效，需要重新初始化。
答： ZAB（Zookeeper Atomic Broadcast）是Zookeeper的原子广播协议，包含两种模式：
1. 崩溃恢复模式（Crash Recovery）：
- 触发条件：Leader宕机或集群启动时
- 过程：Discovery（发现）-> Synchronization（同步）-> Broadcast（广播）
- 选举Leader，同步数据，确保所有Follower与Leader一致
2. 消息广播模式（Message Broadcast）：
- 正常工作模式
- Leader处理所有写请求
消息广播流程：
- 客户端发送写请求到Leader
- Leader生成zxid，写入本地事务日志
- Leader发送Proposal给所有Follower
- Follower写入本地日志，返回ACK
- Leader收到半数以上ACK后，发送Commit
- Follower收到Commit后应用事务
- Leader返回客户端成功
ZXID结构：64位整数，高32位是epoch（Leader周期），低32位是计数器（事务序号）。
答： Leader选举在集群启动或Leader宕机时触发，过程如下：
- 初始状态：所有Server进入LOOKING状态
- 第一轮投票：每个Server投票给自己（包含myid和zxid）
- 交换投票：节点间通过网络交换投票信息
- 投票比较：优先比较zxid（事务ID），zxid大的优先；zxid相同比较myid（服务器ID）
- 更新投票：每轮投票后，更新为更优的投票
- 选举成功：某节点获得半数以上投票成为Leader，其他成为Follower
选举状态：
- LOOKING：寻找Leader
- FOLLOWING：跟随者（已确定Leader）
- LEADING：领导者
- OBSERVING：观察者（不参与投票）
示例：5节点集群，Server 3（myid=3, zxid=100）获得3票（>5/2=2.5），成为Leader。
答： Zookeeper的典型应用场景包括：
-
配置中心：集中管理配置，利用Watcher实时推送变更。配置存储在持久节点，客户端监听数据变化。
-
Master选举：使用临时顺序节点，序号最小的节点成为Master。非Master节点监听前一个节点，当前一个节点删除时重新选举。
-
分布式锁：
- 非公平锁：使用临时节点，创建成功即获得锁
- 公平锁：使用临时顺序节点，序号最小获得锁，其他节点监听前一个节点
-
服务注册与发现：服务启动时创建临时节点（/services/service-name/instance-x），客户端获取子节点列表并监听变化。
-
分布式队列：使用持久顺序节点，序号小的先出队。
-
分布式屏障（Barrier）：所有节点在某个节点下创建子节点，当子节点数达到阈值时，屏障解除。
答： Zookeeper的会话机制包括：
会话状态：
- CONNECTING：连接中
- CONNECTED：已连接
- RECONNECTING：重新连接中
- RECONNECTED：重新连接成功
- CLOSE：连接关闭
- EXPIRED：会话过期
会话超时：
- 由客户端设置，但必须在服务端允许范围内
- minSessionTimeout = 2 * tickTime
- maxSessionTimeout = 20 * tickTime
- 默认tickTime=3000ms，所以会话超时范围是6-60秒
心跳机制：
- 客户端每tickTime/3发送一次心跳（默认1秒）
- 服务端如果在sessionTimeout内未收到心跳，认为会话过期
- 会话过期后，所有临时节点自动删除，所有Watcher失效
故障恢复：
- 网络闪断：客户端自动重连，会话未过期则恢复
- 会话过期：需要重新创建连接，重新创建临时节点，重新注册Watcher
答： Zookeeper通过多种机制保证数据一致性：
-
顺序一致性：所有客户端看到的操作顺序与全局顺序一致。来自同一客户端的操作按发送顺序执行。
-
原子性：每个更新操作要么成功要么失败，不存在部分成功。
-
单一视图：无论连接哪个Server，客户端看到的数据视图一致。
-
可靠性：一旦更新成功，数据将被持久化，直到被覆盖。
-
实时性：保证客户端最终能看到系统的最新状态（在一定时间内）。
具体实现：
- ZAB协议保证消息的全局有序和原子广播
- 写操作必须经过Leader，Leader生成全局唯一的zxid
- Follower必须收到半数以上ACK才Commit
- 读操作可能读到稍旧的数据，但保证顺序一致性
答： Zookeeper和etcd都是分布式协调服务，但有以下区别：
| 特性 | Zookeeper | etcd |
|---|---|---|
| 一致性协议 | ZAB | Raft |
| 数据模型 | 层次化ZNode树 | 扁平KV |
| 监听机制 | Watcher（一次性） | Watch（可持久） |
| 性能 | 中等 | 高 |
| 多数据中心 | 不支持 | 有限支持 |
| 生态 | Hadoop/Kafka/HBase | Kubernetes |
| 实现语言 | Java | Go |
| API风格 | 类似文件系统 | HTTP/gRPC |
选择建议：
- 大数据生态（Hadoop, Kafka）：Zookeeper
- Kubernetes/云原生：etcd
- 需要服务发现+健康检查：Consul
此文原创，转载请注明出处。