Java连接Zookeeper实现Zookeeper教程

在Java项目中，可以使用zookeeper来实现分布式锁、服务注册与发现等功能，本文将详细介绍Java如何连接zookeeper并实现相关功能。

1. Zookeeper简介

Zookeeper是用来实现分布式应用程序协调的开源软件，它是Google的Chubby的开源实现。Zookeeper的设计目标是将那些复杂且容易出错的分布式应用程序中的协调操作封装起来，提供简单而健壮的接口给开发人员使用。Zookeeper可以提供诸如统一命名服务、配置管理、分布式同步、组服务等功能。更多关于Zookeeper的信息可以查看官网。

2. Java连接Zookeeper

2.1 Maven添加依赖

Zookeeper的Java客户端可以通过Maven来进行依赖管理，在pom.xml中添加以下代码依赖：

<dependency>
    <groupId>org.apache.zookeeper</groupId>
    <artifactId>zookeeper</artifactId>
    <version>3.4.10</version>
</dependency>

2.2 创建Zookeeper实例

使用Java连接Zookeeper需要先创建Zookeeper实例，以下是创建Zookeeper实例示例：

public class ZookeeperClient {

    private static Watcher watcher = new Watcher() {
        @Override
        public void process(WatchedEvent event) {

        }
    };

    private static ZooKeeper zk;

    public static void main(String[] args) throws IOException, InterruptedException {
        String connectString = "localhost:2181";
        int sessionTimeout = 5000;
        zk = new ZooKeeper(connectString, sessionTimeout, watcher);
        System.out.println("Zookeeper session established.");
    }
}

其中，connectString是Zookeeper服务器的地址和端口，sessionTimeout是Zookeeper的会话超时时间。

2.3 建立Zookeeper会话

建立Zookeeper会话后，要想在这个会话中操作Zookeeper，需要先等到连接成功后才能进行相关操作。以下是建立Zookeeper会话示例：

public class ZookeeperClient {

    private static Watcher watcher = new Watcher() {
        @Override
        public void process(WatchedEvent event) {
            System.out.println("ZookeeperClient received event: " + event.getType());
        }
    };

    private static ZooKeeper zk;

    public static void main(String[] args) throws IOException, InterruptedException {
        String connectString = "localhost:2181";
        int sessionTimeout = 5000;
        zk = new ZooKeeper(connectString, sessionTimeout, watcher);
        System.out.println("Zookeeper session established.");
        //等待连接成功
        while (zk.getState() != ZooKeeper.States.CONNECTED) {
            Thread.sleep(1000);
        }
        System.out.println("Zookeeper session established and connected.");
    }
}

其中，Watcher是Zookeeper提供的用来接收各种事件的接口，zk.getState() != ZooKeeper.States.CONNECTED语句用来判断是否连接成功。

2.4 创建Znode节点

在Zookeeper中，节点是事先不需要创建，只要在使用的时候调用创建节点的方法就可以了。以下是创建Znode节点的示例：

public class ZookeeperClient {

    private static Watcher watcher = new Watcher() {
        @Override
        public void process(WatchedEvent event) {
            System.out.println("ZookeeperClient received event: " + event.getType());
        }
    };

    private static ZooKeeper zk;

    public static void main(String[] args) throws IOException, InterruptedException, KeeperException {
        String connectString = "localhost:2181";
        int sessionTimeout = 5000;
        zk = new ZooKeeper(connectString, sessionTimeout, watcher);
        System.out.println("Zookeeper session established.");
        //等待连接成功
        while (zk.getState() != ZooKeeper.States.CONNECTED) {
            Thread.sleep(1000);
        }
        System.out.println("Zookeeper session established and connected.");
        //创建节点
        String path = "/test";
        byte[] data = "znode_data".getBytes();
        zk.create(path, data, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        System.out.println("Znode created: " + path);
    }
}

在Zookeeper中，每个节点的路径和数据是用byte数组来表示的，CreateMode.PERSISTENT表示创建的节点是永久的。

2.5 获取Znode节点内容

在Zookeeper中，通过节点的路径来获取节点的数据。以下是获取Znode节点内容的示例：

public class ZookeeperClient {

    private static Watcher watcher = new Watcher() {
        @Override
        public void process(WatchedEvent event) {
            System.out.println("ZookeeperClient received event: " + event.getType());
        }
    };

    private static ZooKeeper zk;

    public static void main(String[] args) throws IOException, InterruptedException, KeeperException {
        String connectString = "localhost:2181";
        int sessionTimeout = 5000;
        zk = new ZooKeeper(connectString, sessionTimeout, watcher);
        System.out.println("Zookeeper session established.");
        //等待连接成功
        while (zk.getState() != ZooKeeper.States.CONNECTED) {
            Thread.sleep(1000);
        }
        System.out.println("Zookeeper session established and connected.");
        //获取节点内容
        String path = "/test";
        byte[] data = zk.getData(path, watcher, null);
        System.out.println("Znode data: " + new String(data));
    }
}

其中，watcher用来监听节点的变化。

3. 示例说明

3.1 分布式锁示例

分布式锁的实现原理是使用zookeeper的临时节点。当客户端需要获取锁的时候，就在zookeeper中创建一个临时节点，只有创建成功的客户端才能访问共享资源。以下是分布式锁的示例：

public class DistributedLock {

    private String lockPath; // Zookeeper锁节点的路径
    private String lockName; // 锁节点的名称
    private ZooKeeper zk; // Zookeeper客户端
    private Watcher watcher = new Watcher() {
        @Override
        public void process(WatchedEvent event) {
            if (event.getType() == Event.EventType.NodeDeleted) {
                synchronized (this) {
                    this.notifyAll();
                }
            }
        }
    };

    public DistributedLock(String connectString, String lockPath, String lockName, int sessionTimeout) throws IOException {
        this.lockPath = lockPath;
        this.lockName = lockName;
        this.zk = new ZooKeeper(connectString, sessionTimeout, watcher);
    }

    // 获取锁
    public void lock() throws KeeperException, InterruptedException {
        // 创建临时节点
        String path = zk.create(lockPath + "/" + lockName, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
        // 获取所有子节点
        List<String> children = zk.getChildren(lockPath, false);
        // 排序
        Collections.sort(children);
        // 判断当前临时节点是否为最小的子节点
        if (path.equals(lockPath + "/" + children.get(0))) {
            return;
        }
        // 不是最小的子节点就监听它之前的子节点
        String prevNodeName = children.get(Collections.binarySearch(children, lockName.substring(6)) - 1);
        String prevPath = lockPath + "/" + prevNodeName;
        Stat stat = zk.exists(prevPath, watcher);
        if (stat != null) {
            synchronized (watcher) {
                watcher.wait();
            }
        } else {
            lock();
        }
    }

    // 释放锁
    public void unlock() throws KeeperException, InterruptedException {
        zk.delete(lockPath + "/" + lockName, -1);
        zk.close();
    }
}

可以使用以下方式进行测试：

public class DistributedLockTest {

    public static void main(String[] args) {
        try {
            DistributedLock lock1 = new DistributedLock("localhost:2181", "/locks", "testLock", 2000);
            lock1.lock();
            System.out.println("Client1 has obtained the lock.");
            DistributedLock lock2 = new DistributedLock("localhost:2181", "/locks", "testLock", 2000);
            lock2.lock();
            System.out.println("Client2 has obtained the lock.");
            lock1.unlock();
            System.out.println("Client1 has released the lock.");
            lock2.unlock();
            System.out.println("Client2 has released the lock.");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

3.2 服务注册与发现示例

服务注册与发现的实现原理是使用zookeeper的节点创建和删除。服务提供者在zookeeper中创建一个永久节点，并将自己的IP地址和端口号加入到节点数据中；服务消费者在zookeeper中监听这个节点，当服务提供者的节点被删除或者更新时，服务消费者可以获取到相应的服务地址和端口号。

以下是服务注册与发现的示例：

public class ServiceRegistry {

    private String registryPath;
    private ZooKeeper zk;

    public ServiceRegistry(String connectString, String registryPath, int sessionTimeout) throws IOException {
        this.registryPath = registryPath;
        this.zk = new ZooKeeper(connectString, sessionTimeout, null);
    }

    // 注册服务
    public void registerService(String serviceName, String serviceAddr) throws KeeperException, InterruptedException {
        String nodePath = registryPath + "/" + serviceName;
        Stat stat = zk.exists(nodePath, false);
        if (stat == null) {
            zk.create(nodePath, null, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        }
        String serviceNodePath = nodePath + "/" + UUID.randomUUID();
        zk.create(serviceNodePath, serviceAddr.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
        System.out.println("Service " + serviceName + " registered on " + serviceAddr);
    }

    // 发现服务
    public List<String> discoverService(String serviceName) throws KeeperException, InterruptedException {
        String nodePath = registryPath + "/" + serviceName;
        List<String> serviceNodes = zk.getChildren(nodePath, false);
        List<String> serviceAddrs = new ArrayList<>();
        for (String serviceNode : serviceNodes) {
            byte[] serviceAddrBytes = zk.getData(nodePath + "/" + serviceNode, false, null);
            serviceAddrs.add(new String(serviceAddrBytes));
        }
        System.out.println("Service " + serviceName + " discovered: " + serviceAddrs);
        return serviceAddrs;
    }
}

可以使用以下方式进行测试：

public class ServiceRegistryTest {

    public static void main(String[] args) {
        try {
            ServiceRegistry registry = new ServiceRegistry("localhost:2181", "/services", 2000);
            registry.registerService("testService", "127.0.0.1:8080");
            registry.discoverService("testService");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

4. 总结

Zookeeper是一款非常好用的分布式协同服务框架，使用Java连接Zookeeper可以为分布式应用程序提供诸如统一命名服务、配置管理、分布式同步、组服务等功能。本文详细介绍了Java连接Zookeeper的相关知识，并且给出了两个示例说明。