Zookeeper 概述与使用指南

什么是Zookeeper？

Apache Zookeeper是一个开源的分布式协调服务，它为分布式应用提供一致性服务，包括配置维护、域名服务、分布式同步和组服务等。Zookeeper最初是Hadoop的一个子项目，但现在已成为许多分布式系统的核心组件。

Zookeeper的主要作用

1. 分布式协调：在分布式系统中协调各个节点的状态和行为
2. 配置管理：集中管理分布式系统的配置信息
3. 命名服务：提供类似DNS的服务，将名称映射到资源
4. 分布式锁：实现分布式环境下的互斥访问
5. 集群管理：监控集群节点状态，实现主节点选举
6. 队列管理：实现简单的分布式队列

使用Zookeeper的框架

1. Hadoop：用于NameNode的高可用性
2. Kafka：用于broker的元数据管理和控制器选举
3. HBase：用于RegionServer的协调和主节点选举
4. Dubbo：用于服务注册与发现
5. Solr Cloud：用于集群管理和配置存储

典型使用场景

1. 配置管理

// 创建Zookeeper客户端
ZooKeeper zk = new ZooKeeper("localhost:2181", 3000, null);

// 存储配置
zk.create("/config/app1/setting1", "value1".getBytes(), 
          ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);

// 获取配置
byte[] data = zk.getData("/config/app1/setting1", false, null);
System.out.println("配置值: " + new String(data));

// 监听配置变化
zk.getData("/config/app1/setting1", event -> {
    if (event.getType() == Event.EventType.NodeDataChanged) {
        System.out.println("配置已更新");
    }
}, null);

2. 分布式锁

public class DistributedLock {
    private final ZooKeeper zk;
    private final String lockPath;
    private String currentLockPath;
    
    public DistributedLock(ZooKeeper zk, String lockPath) {
        this.zk = zk;
        this.lockPath = lockPath;
    }
    
    public void lock() throws Exception {
        // 创建临时顺序节点
        currentLockPath = zk.create(lockPath + "/lock-", 
                                  new byte[0],
                                  ZooDefs.Ids.OPEN_ACL_UNSAFE,
                                  CreateMode.EPHEMERAL_SEQUENTIAL);
        
        // 获取所有子节点并排序
        List<String> children = zk.getChildren(lockPath, false);
        Collections.sort(children);
        
        // 检查当前节点是否是最小的
        String smallest = children.get(0);
        if (!currentLockPath.endsWith(smallest)) {
            // 如果不是最小的，等待前一个节点释放
            String previous = lockPath + "/" + children.get(Collections.binarySearch(children, 
                currentLockPath.substring(currentLockPath.lastIndexOf('/') + 1)) - 1);
            
            final CountDownLatch latch = new CountDownLatch(1);
            Stat stat = zk.exists(previous, event -> {
                if (event.getType() == Event.EventType.NodeDeleted) {
                    latch.countDown();
                }
            });
            
            if (stat != null) {
                latch.await();
            }
        }
    }
    
    public void unlock() throws Exception {
        zk.delete(currentLockPath, -1);
    }
}

3. 服务注册与发现

// 服务注册
public void registerService(String serviceName, String serviceAddress) throws Exception {
    String servicePath = "/services/" + serviceName;
    if (zk.exists(servicePath, false) == null) {
        zk.create(servicePath, null, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
    }
    
    String instancePath = servicePath + "/instance-";
    zk.create(instancePath, serviceAddress.getBytes(), 
             ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
}

// 服务发现
public List<String> discoverServices(String serviceName) throws Exception {
    String servicePath = "/services/" + serviceName;
    List<String> instances = zk.getChildren(servicePath, false);
    List<String> addresses = new ArrayList<>();
    
    for (String instance : instances) {
        byte[] data = zk.getData(servicePath + "/" + instance, false, null);
        addresses.add(new String(data));
    }
    
    return addresses;
}

Zookeeper的缺陷

1. 性能瓶颈：写入性能随节点数量增加而下降
2. 脑裂问题：在网络分区情况下可能出现多个主节点
3. 配置复杂：需要合理设置超时和重试参数
4. 存储限制：不适合存储大量数据，设计用于存储元数据
5. 客户端复杂性：需要处理连接丢失和会话过期等问题

与其他协调服务的比较

特性	Zookeeper	etcd	Consul
一致性算法	ZAB协议	Raft协议	Raft协议
接口协议	自定义二进制协议	HTTP/JSON	HTTP/JSON
服务发现	需要额外开发	内置	内置
健康检查	有限支持	有限支持	全面支持
KV存储	支持	支持	支持
多数据中心	需要额外配置	有限支持	原生支持
监控	有限	有限	全面

实际案例：Kafka使用Zookeeper

Kafka使用Zookeeper进行以下操作：

1. Broker注册：每个broker启动时在Zookeeper中注册自己
2. Topic配置：存储topic的分区信息和配置
3. 消费者偏移量：老版本Kafka将消费者偏移量存储在Zookeeper中
4. 控制器选举：选举集群控制器来管理分区leader和副本

// Kafka使用Zookeeper的示例代码（简化版）
public class KafkaZookeeperExample {
    public static void main(String[] args) throws Exception {
        ZooKeeper zk = new ZooKeeper("localhost:2181", 3000, null);
        
        // 获取所有broker
        List<String> brokers = zk.getChildren("/brokers/ids", false);
        System.out.println("活跃的Kafka brokers: " + brokers);
        
        // 获取所有topic
        List<String> topics = zk.getChildren("/brokers/topics", false);
        System.out.println("存在的topics: " + topics);
        
        // 监听controller变化
        zk.getData("/controller", event -> {
            if (event.getType() == Event.EventType.NodeDataChanged) {
                System.out.println("Kafka控制器已变更");
            }
        }, null);
    }
}

最佳实践

1. 合理设置会话超时：通常设置在2-20秒之间
2. 使用连接池：避免频繁创建和关闭连接
3. 处理连接丢失：实现重试逻辑和故障转移
4. 避免大节点：单个节点数据不宜过大（建议<1MB）
5. 合理使用watch：避免过度使用watch导致性能问题

Zookeeper在分布式系统中扮演着重要角色，虽然它有一些局限性，但在需要强一致性的场景中仍然是许多系统的首选协调服务。