Zookeeper 技术详细介绍

Zookeeper 技术详解

目录

• 1. Zookeeper 简介
  • 1.1 什么是 Zookeeper
  • 1.2 核心特性
  • 1.3 架构组件
  • 1.4 节点类型
• 2. 使用场景
  • 2.1 配置管理
  • 2.2 服务注册与发现
  • 2.3 分布式锁
  • 2.4 集群管理
• 3. 核心流程
  • 3.1 ZAB协议（Zookeeper Atomic Broadcast）
  • 3.2 会话管理
  • 3.3 数据同步
• 4. 重难点分析
  • 4.1 脑裂问题（Split Brain）
  • 4.2 数据一致性保证
  • 4.3 性能优化
  • 4.4 故障处理
• 5. 高频面试点
  • 5.1 基础概念类
  • 5.2 技术实现类
  • 5.3 性能优化类
  • 5.4 故障处理类
  • 5.5 应用场景类
  • 5.6 源码分析类
• 6. Zookeeper 部署与运维
  • 6.1 环境准备
  • 6.2 单机部署
  • 6.3 集群部署
  • 6.4 Docker 部署
  • 6.5 部署问题检查
  • 6.6 监控和运维

1. Zookeeper 简介

1.1 什么是 Zookeeper

Apache Zookeeper 是一个开源的分布式协调服务，为分布式应用提供一致性服务。它主要用于解决分布式系统中的数据一致性问题，提供配置维护、域名服务、分布式同步、组服务等功能。

1.2 核心特性

1.2.1 一致性保证

• 顺序一致性：客户端更新将按发送顺序执行
• 原子性：更新要么成功要么失败，不会出现部分成功
• 单一系统映像：无论连接到哪个服务器，客户端看到的是相同的服务视图
• 可靠性：一旦更新被应用，它将从那时起持续存在，直到客户端覆盖更新
• 实时性：在特定时间范围内，客户端看到的系统状态是最新的

1.2.2 数据模型

Zookeeper 的数据模型是一个类似文件系统的树形结构：

/
├── app1
│   ├── config
│   │   ├── database
│   │   └── cache
│   └── locks
├── app2
│   ├── services
│   └── coordination
└── global
    ├── leader
    └── workers

1.3 架构组件

Client

Zookeeper Cluster

Leader

Follower 1

Follower 2

Observer

ZAB协议

只读操作

数据一致性

提高读性能

1.4 节点类型

1.4.1 持久节点（Persistent）

• 创建后一直存在，直到被显式删除
• 适用于配置信息存储

1.4.2 临时节点（Ephemeral）

• 客户端会话结束时自动删除
• 适用于服务注册与发现

1.4.3 顺序节点（Sequential）

• 节点名后自动添加递增序号
• 适用于分布式锁、队列等场景

1.4.4 临时顺序节点（Ephemeral Sequential）

• 结合临时节点和顺序节点的特性
• 最常用的节点类型

2. 使用场景

2.1 配置管理

2.1.1 场景描述

在分布式系统中，多个服务需要共享配置信息，如数据库连接字符串、缓存配置等。

2.1.2 实现方案

1// 配置管理示例
2public class ConfigManager {
3    private ZooKeeper zk;
4    private String configPath = "/app/config";
5  
6    public void init() throws Exception {
7        zk = new ZooKeeper("localhost:2181", 3000, new Watcher() {
8            public void process(WatchedEvent event) {
9                if (event.getType() == Event.EventType.NodeDataChanged) {
10                    // 配置变更，重新加载配置
11                    loadConfig();
12                }
13            }
14        });
15    }
16  
17    public void loadConfig() {
18        try {
19            byte[] data = zk.getData(configPath, true, null);
20            String config = new String(data);
21            // 更新应用配置
22            updateApplicationConfig(config);
23        } catch (Exception e) {
24            e.printStackTrace();
25        }
26    }
27  
28    public void updateConfig(String newConfig) throws Exception {
29        zk.setData(configPath, newConfig.getBytes(), -1);
30    }
31}
32

2.1.3 配置管理流程图

应用程序 Zookeeper 配置中心 1. 监听配置节点 2. 更新配置数据 3. 触发配置变更事件 4. 获取最新配置 5. 重新加载配置 应用程序 Zookeeper 配置中心

2.2 服务注册与发现

2.2.1 场景描述

在微服务架构中，服务需要注册自己的地址信息，其他服务通过服务发现机制找到目标服务。

2.2.2 服务注册实现

1// 服务注册实现
2public class ServiceRegistry {
3    private ZooKeeper zk;
4    private String servicePath = "/services";
5  
6    public void registerService(String serviceName, String serviceAddress) {
7        try {
8            // 创建服务节点
9            String serviceNode = servicePath + "/" + serviceName;
10            if (zk.exists(serviceNode, false) == null) {
11                zk.create(serviceNode, null, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
12            }
13      
14            // 创建服务实例节点（临时节点）
15            String instanceNode = serviceNode + "/instance-";
16            String instancePath = zk.create(instanceNode, serviceAddress.getBytes(), 
17                ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
18      
19            System.out.println("服务注册成功: " + instancePath);
20        } catch (Exception e) {
21            e.printStackTrace();
22        }
23    }
24  
25    public List<String> discoverServices(String serviceName) {
26        List<String> services = new ArrayList<>();
27        try {
28            String serviceNode = servicePath + "/" + serviceName;
29            List<String> instances = zk.getChildren(serviceNode, false);
30      
31            for (String instance : instances) {
32                String instancePath = serviceNode + "/" + instance;
33                byte[] data = zk.getData(instancePath, false, null);
34                services.add(new String(data));
35            }
36        } catch (Exception e) {
37            e.printStackTrace();
38        }
39        return services;
40    }
41}
42

2.2.3 服务发现流程图

服务启动

注册到Zookeeper

创建临时节点

其他服务发现

获取服务列表

负载均衡选择

调用目标服务

服务下线

临时节点自动删除

其他服务感知到变化

更新服务列表

2.3 分布式锁

2.3.1 场景描述

在分布式系统中，多个进程需要互斥访问共享资源，需要实现分布式锁。

2.3.2 分布式锁实现

1// 分布式锁实现
2public class DistributedLock {
3    private ZooKeeper zk;
4    private String lockPath = "/locks";
5    private String lockName;
6    private String lockNode;
7    private CountDownLatch lockAcquired = new CountDownLatch(1);
8  
9    public DistributedLock(ZooKeeper zk, String lockName) {
10        this.zk = zk;
11        this.lockName = lockName;
12    }
13  
14    public boolean tryLock() {
15        try {
16            // 创建临时顺序节点
17            lockNode = zk.create(lockPath + "/" + lockName + "-", 
18                Thread.currentThread().getName().getBytes(),
19                ZooDefs.Ids.OPEN_ACL_UNSAFE, 
20                CreateMode.EPHEMERAL_SEQUENTIAL);
21      
22            // 获取所有锁节点
23            List<String> lockNodes = zk.getChildren(lockPath, false);
24            Collections.sort(lockNodes);
25      
26            // 检查是否获得锁
27            String lockNodeName = lockNode.substring(lockNode.lastIndexOf("/") + 1);
28            int lockIndex = lockNodes.indexOf(lockNodeName);
29      
30            if (lockIndex == 0) {
31                // 获得锁
32                return true;
33            } else {
34                // 监听前一个节点
35                String prevNode = lockPath + "/" + lockNodes.get(lockIndex - 1);
36                zk.exists(prevNode, new LockWatcher());
37                lockAcquired.await();
38                return true;
39            }
40        } catch (Exception e) {
41            e.printStackTrace();
42            return false;
43        }
44    }
45  
46    public void unlock() {
47        try {
48            zk.delete(lockNode, -1);
49        } catch (Exception e) {
50            e.printStackTrace();
51        }
52    }
53  
54    private class LockWatcher implements Watcher {
55        @Override
56        public void process(WatchedEvent event) {
57            if (event.getType() == Event.EventType.NodeDeleted) {
58                lockAcquired.countDown();
59            }
60        }
61    }
62}
63

2.3.3 分布式锁流程图

客户端1 客户端2 Zookeeper 1. 创建临时顺序节点 2. 创建临时顺序节点 3. 返回节点: lock-0000000001 4. 返回节点: lock-0000000002 5. 获取所有锁节点 6. 获取所有锁节点 节点序号最小，获得锁 监听前一个节点 7. 执行业务逻辑 8. 删除节点，释放锁 9. 通知前一个节点已删除 10. 获得锁，执行业务逻辑 客户端1 客户端2 Zookeeper

2.4 集群管理

2.4.1 场景描述

在分布式系统中，需要管理集群中的节点，如选举主节点、监控节点状态等。

2.4.2 主节点选举实现

1// 主节点选举实现
2public class LeaderElection {
3    private ZooKeeper zk;
4    private String electionPath = "/election";
5    private String nodePath;
6    private boolean isLeader = false;
7  
8    public void participateInElection() {
9        try {
10            // 创建临时顺序节点
11            nodePath = zk.create(electionPath + "/node-", 
12                InetAddress.getLocalHost().getHostAddress().getBytes(),
13                ZooDefs.Ids.OPEN_ACL_UNSAFE, 
14                CreateMode.EPHEMERAL_SEQUENTIAL);
15      
16            // 检查是否成为主节点
17            checkForLeadership();
18      
19            // 监听节点变化
20            zk.getChildren(electionPath, new ElectionWatcher());
21      
22        } catch (Exception e) {
23            e.printStackTrace();
24        }
25    }
26  
27    private void checkForLeadership() {
28        try {
29            List<String> children = zk.getChildren(electionPath, false);
30            Collections.sort(children);
31      
32            String nodeName = nodePath.substring(nodePath.lastIndexOf("/") + 1);
33            int nodeIndex = children.indexOf(nodeName);
34      
35            if (nodeIndex == 0) {
36                // 成为主节点
37                isLeader = true;
38                System.out.println("成为主节点: " + nodeName);
39                onBecomeLeader();
40            } else {
41                // 不是主节点
42                isLeader = false;
43                System.out.println("成为从节点: " + nodeName);
44                onBecomeFollower();
45            }
46        } catch (Exception e) {
47            e.printStackTrace();
48        }
49    }
50  
51    private void onBecomeLeader() {
52        // 主节点逻辑
53        System.out.println("开始执行主节点任务...");
54    }
55  
56    private void onBecomeFollower() {
57        // 从节点逻辑
58        System.out.println("开始执行从节点任务...");
59    }
60  
61    private class ElectionWatcher implements Watcher {
62        @Override
63        public void process(WatchedEvent event) {
64            if (event.getType() == Event.EventType.NodeChildrenChanged) {
65                checkForLeadership();
66                try {
67                    zk.getChildren(electionPath, this);
68                } catch (Exception e) {
69                    e.printStackTrace();
70                }
71            }
72        }
73    }
74}
75

3. 核心流程

3.1 ZAB协议（Zookeeper Atomic Broadcast）

3.1.1 协议概述

ZAB协议是Zookeeper的核心协议，用于保证分布式数据的一致性。

3.1.2 协议流程

是

否

客户端请求

Leader接收

生成事务ID

广播事务提案

Follower接收提案

Follower响应ACK

收到多数ACK?

提交事务

等待更多ACK

通知Follower提交

返回客户端响应

3.1.3 协议实现

1// ZAB协议简化实现
2public class ZABProtocol {
3    private List<Server> followers;
4    private Server leader;
5    private long lastZxid = 0;
6  
7    public void processRequest(Transaction transaction) {
8        // 1. 生成事务ID
9        transaction.setZxid(++lastZxid);
10  
11        // 2. 广播事务提案
12        Proposal proposal = new Proposal(transaction);
13        broadcastProposal(proposal);
14  
15        // 3. 等待多数派确认
16        waitForMajorityAck(proposal);
17  
18        // 4. 提交事务
19        commitTransaction(transaction);
20    }
21  
22    private void broadcastProposal(Proposal proposal) {
23        for (Server follower : followers) {
24            follower.sendProposal(proposal);
25        }
26    }
27  
28    private void waitForMajorityAck(Proposal proposal) {
29        int ackCount = 0;
30        int majority = followers.size() / 2 + 1;
31  
32        while (ackCount < majority) {
33            // 等待ACK响应
34            try {
35                Thread.sleep(10);
36            } catch (InterruptedException e) {
37                e.printStackTrace();
38            }
39            ackCount = proposal.getAckCount();
40        }
41    }
42  
43    private void commitTransaction(Transaction transaction) {
44        // 提交到本地存储
45        leader.commit(transaction);
46  
47        // 通知Follower提交
48        for (Server follower : followers) {
49            follower.sendCommit(transaction);
50        }
51    }
52}
53

3.2 会话管理

3.2.1 会话生命周期

客户端连接

连接成功

心跳正常

会话超时

主动关闭

清理资源

清理资源

CONNECTING

CONNECTED

EXPIRED

CLOSED

3.2.2 会话超时处理

1// 会话管理实现
2public class SessionManager {
3    private Map<Long, Session> sessions = new ConcurrentHashMap<>();
4    private long sessionTimeout = 30000; // 30秒
5  
6    public void createSession(long sessionId, String clientAddress) {
7        Session session = new Session(sessionId, clientAddress);
8        sessions.put(sessionId, session);
9  
10        // 启动会话超时检查
11        scheduleSessionCheck(session);
12    }
13  
14    public void updateSession(long sessionId) {
15        Session session = sessions.get(sessionId);
16        if (session != null) {
17            session.updateLastAccessTime();
18        }
19    }
20  
21    public void closeSession(long sessionId) {
22        Session session = sessions.remove(sessionId);
23        if (session != null) {
24            // 清理会话相关资源
25            cleanupSession(session);
26        }
27    }
28  
29    private void scheduleSessionCheck(Session session) {
30        Timer timer = new Timer();
31        timer.schedule(new TimerTask() {
32            @Override
33            public void run() {
34                if (session.isExpired(sessionTimeout)) {
35                    // 会话超时，清理资源
36                    closeSession(session.getSessionId());
37                    timer.cancel();
38                } else {
39                    // 继续检查
40                    scheduleSessionCheck(session);
41                }
42            }
43        }, sessionTimeout);
44    }
45  
46    private void cleanupSession(Session session) {
47        // 删除临时节点
48        deleteEphemeralNodes(session.getSessionId());
49  
50        // 通知相关客户端
51        notifySessionExpired(session.getSessionId());
52    }
53}
54

3.3 数据同步

3.3.1 数据同步流程

客户端 Leader Follower1 Follower2 1. 写请求 2. 生成事务日志 3. 发送事务提案 4. 发送事务提案 5. 响应ACK 6. 响应ACK 7. 收到多数ACK 8. 返回成功响应 9. 发送提交命令 10. 发送提交命令 客户端 Leader Follower1 Follower2

3.3.2 数据同步实现

1// 数据同步实现
2public class DataSync {
3    private ZooKeeper zk;
4    private String dataPath = "/data";
5  
6    public void syncData() {
7        try {
8            // 1. 获取当前数据版本
9            Stat stat = new Stat();
10            byte[] data = zk.getData(dataPath, false, stat);
11            long currentVersion = stat.getVersion();
12      
13            // 2. 检查是否需要同步
14            if (needsSync(currentVersion)) {
15                // 3. 执行数据同步
16                performSync(currentVersion);
17            }
18        } catch (Exception e) {
19            e.printStackTrace();
20        }
21    }
22  
23    private boolean needsSync(long currentVersion) {
24        // 检查本地数据版本是否最新
25        return getLocalVersion() < currentVersion;
26    }
27  
28    private void performSync(long targetVersion) {
29        try {
30            // 获取目标版本的数据
31            Stat stat = new Stat();
32            byte[] data = zk.getData(dataPath, false, stat);
33      
34            // 更新本地数据
35            updateLocalData(data);
36      
37            // 更新本地版本
38            updateLocalVersion(targetVersion);
39      
40        } catch (Exception e) {
41            e.printStackTrace();
42        }
43    }
44}
45

4. 重难点分析

4.1 脑裂问题（Split Brain）

4.1.1 问题描述

脑裂是指分布式系统中，由于网络分区导致集群被分割成多个独立的部分，每个部分都认为自己是主节点。

4.1.2 问题分析

原始集群

网络分区

分区1: Leader A

分区2: Leader B

服务客户端1

服务客户端2

数据不一致

系统故障

4.1.3 解决方案

1. 多数派原则

1// 多数派选举实现
2public class MajorityElection {
3    private int totalNodes;
4    private int requiredMajority;
5  
6    public MajorityElection(int totalNodes) {
7        this.totalNodes = totalNodes;
8        this.requiredMajority = totalNodes / 2 + 1;
9    }
10  
11    public boolean canBecomeLeader(int availableNodes) {
12        return availableNodes >= requiredMajority;
13    }
14  
15    public boolean isQuorumAvailable(int availableNodes) {
16        return availableNodes >= requiredMajority;
17    }
18}
19

2. 租约机制

1// 租约机制实现
2public class LeaseMechanism {
3    private long leaseTimeout = 5000; // 5秒租约
4    private long lastHeartbeat = 0;
5  
6    public boolean isLeaseValid() {
7        return System.currentTimeMillis() - lastHeartbeat < leaseTimeout;
8    }
9  
10    public void renewLease() {
11        lastHeartbeat = System.currentTimeMillis();
12    }
13  
14    public void checkLease() {
15        if (!isLeaseValid()) {
16            // 租约过期，放弃领导权
17            relinquishLeadership();
18        }
19    }
20}
21

4.2 数据一致性保证

4.2.1 一致性模型

强一致性

顺序一致性

因果一致性

最终一致性

弱一致性

4.2.2 一致性实现

1. 两阶段提交（2PC）

1// 2PC实现
2public class TwoPhaseCommit {
3    private List<Participant> participants;
4  
5    public boolean executeTransaction(Transaction transaction) {
6        // 阶段1：准备阶段
7        if (preparePhase(transaction)) {
8            // 阶段2：提交阶段
9            return commitPhase(transaction);
10        } else {
11            // 回滚
12            rollbackPhase(transaction);
13            return false;
14        }
15    }
16  
17    private boolean preparePhase(Transaction transaction) {
18        for (Participant participant : participants) {
19            if (!participant.prepare(transaction)) {
20                return false;
21            }
22        }
23        return true;
24    }
25  
26    private boolean commitPhase(Transaction transaction) {
27        for (Participant participant : participants) {
28            if (!participant.commit(transaction)) {
29                return false;
30            }
31        }
32        return true;
33    }
34  
35    private void rollbackPhase(Transaction transaction) {
36        for (Participant participant : participants) {
37            participant.rollback(transaction);
38        }
39    }
40}
41

2. 三阶段提交（3PC）

1// 3PC实现
2public class ThreePhaseCommit {
3    private List<Participant> participants;
4  
5    public boolean executeTransaction(Transaction transaction) {
6        // 阶段1：CanCommit
7        if (canCommitPhase(transaction)) {
8            // 阶段2：PreCommit
9            if (preCommitPhase(transaction)) {
10                // 阶段3：DoCommit
11                return doCommitPhase(transaction);
12            } else {
13                abortPhase(transaction);
14                return false;
15            }
16        } else {
17            abortPhase(transaction);
18            return false;
19        }
20    }
21  
22    private boolean canCommitPhase(Transaction transaction) {
23        for (Participant participant : participants) {
24            if (!participant.canCommit(transaction)) {
25                return false;
26            }
27        }
28        return true;
29    }
30  
31    private boolean preCommitPhase(Transaction transaction) {
32        for (Participant participant : participants) {
33            if (!participant.preCommit(transaction)) {
34                return false;
35            }
36        }
37        return true;
38    }
39  
40    private boolean doCommitPhase(Transaction transaction) {
41        for (Participant participant : participants) {
42            if (!participant.doCommit(transaction)) {
43                return false;
44            }
45        }
46        return true;
47    }
48}
49

4.3 性能优化

4.3.1 读写性能优化

1. 批量操作

1// 批量操作实现
2public class BatchOperations {
3    private ZooKeeper zk;
4  
5    public void batchCreate(List<String> paths, List<byte[]> data) {
6        List<Op> ops = new ArrayList<>();
7  
8        for (int i = 0; i < paths.size(); i++) {
9            ops.add(Op.create(paths.get(i), data.get(i), 
10                ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT));
11        }
12  
13        try {
14            zk.multi(ops);
15        } catch (Exception e) {
16            e.printStackTrace();
17        }
18    }
19  
20    public void batchDelete(List<String> paths) {
21        List<Op> ops = new ArrayList<>();
22  
23        for (String path : paths) {
24            ops.add(Op.delete(path, -1));
25        }
26  
27        try {
28            zk.multi(ops);
29        } catch (Exception e) {
30            e.printStackTrace();
31        }
32    }
33}
34