基于Apache Curator框架的ZooKeeper使用详解

• orSetData() 方法：如果节点存在则 Curator 将会使用给出的数据设置这个节点的值，相当于 setData() 方法
• creatingParentContainersIfNeeded() 方法：如果指定节点的父节点不存在，则 Curator 将会自动级联创建父节点
• guaranteed() 方法：如果服务端可能删除成功，但是 client 没有接收到删除成功的提示，Curator 将会在后台持续尝试删除该节点
• deletingChildrenIfNeeded() 方法：如果待删除节点存在子节点，则 Curator 将会级联删除该节点的子节点

事务管理：

/**
 * 事务管理：碰到异常，事务会回滚
 * @throws Exception
 */
@Test
public void testTransaction() throws Exception {
    //定义几个基本操作
    CuratorOp createOp = client.transactionOp().create()
            .forPath("/curator/one_path","some data".getBytes());
    CuratorOp setDataOp = client.transactionOp().setData()
            .forPath("/curator","other data".getBytes());
    CuratorOp deleteOp = client.transactionOp().delete()
            .forPath("/curator");
    //事务执行结果
    List<CuratorTransactionResult> results = client.transaction()
            .forOperations(createOp,setDataOp,deleteOp);
    //遍历输出结果
    for(CuratorTransactionResult result : results){
        System.out.println("执行结果是：" + result.getForPath() + "--" + result.getType());
    }
}
//因为节点'/curator'存在子节点，所以在删除的时候将会报错，事务回滚

三 监听器

Curator 提供了三种 Watcher(Cache) 来监听结点的变化：

• Path Cache：监视一个路径下 1) 孩子结点的创建、2) 删除，3) 以及结点数据的更新。产生的事件会传递给注册的 PathChildrenCacheListener。
• Node Cache：监视一个结点的创建、更新、删除，并将结点的数据缓存在本地。
• Tree Cache：Path Cache 和 Node Cache 的'合体'，监视路径下的创建、更新、删除事件，并缓存路径下所有孩子结点的数据。

/**
 * 在注册监听器的时候，如果传入此参数，当事件触发时，逻辑由线程池处理
 */
ExecutorService pool = Executors.newFixedThreadPool(2);
/**
 * 监听数据节点的变化情况
 */
final NodeCache nodeCache = new NodeCache(client, "/zk-huey/cnode", false);
nodeCache.start(true);
nodeCache.getListenable().addListener(
        new NodeCacheListener() {
            @Override
            public void nodeChanged() throws Exception {
                System.out.println("Node data is changed, new data: " + new String(nodeCache.getCurrentData().getData()));
            }
        }, pool
);
/**
 * 监听子节点的变化情况
 */
final PathChildrenCache childrenCache = new PathChildrenCache(client, "/zk-huey", true);
childrenCache.start(StartMode.POST_INITIALIZED_EVENT);
childrenCache.getListenable().addListener(
        new PathChildrenCacheListener() {
            @Override
            public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception {
                switch (event.getType()) {
                    case CHILD_ADDED:
                        System.out.println("CHILD_ADDED: " + event.getData().getPath());
                        break;
                    case CHILD_REMOVED:
                        System.out.println("CHILD_REMOVED: " + event.getData().getPath());
                        break;
                    case CHILD_UPDATED:
                        System.out.println("CHILD_UPDATED: " + event.getData().getPath());
                        break;
                    default:
                        break;
                }
            }
        }, pool
);
client.setData().forPath("/zk-huey/cnode", "world".getBytes());
Thread.sleep(10 * 1000);
pool.shutdown();
client.close();

四 分布式锁

分布式编程时，比如最容易碰到的情况就是应用程序在线上多机部署，于是当多个应用同时访问某一资源时，就需要某种机制去协调它们。例如，现在一台应用正在 rebuild 缓存内容，要临时锁住某个区域暂时不让访问；又比如调度程序每次只想一个任务被一台应用执行等等。

下面的程序会启动两个线程 t1 和 t2 去争夺锁，拿到锁的线程会占用 5 秒。运行多次可以观察到，有时是 t1 先拿到锁而 t2 等待，有时又会反过来。Curator 会用我们提供的 lock 路径的结点作为全局锁，这个结点的数据类似这种格式：[_c_64e0811f-9475-44ca-aa36-c1db65ae5350-lock-0000000005]，每次获得锁时会生成这种串，释放锁时清空数据。

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.locks.InterProcessMutex;
import org.apache.curator.retry.RetryNTimes;
import java.util.concurrent.TimeUnit;

/**
 * Curator framework's distributed lock test.
 */
public class CuratorDistrLockTest {
    /** Zookeeper info */
    private static final String ZK_ADDRESS = "192.168.1.100:2181";
    private static final String ZK_LOCK_PATH = "/zktest";

    public static void main(String[] args) throws InterruptedException {
        // 1.Connect to zk
        CuratorFramework client = CuratorFrameworkFactory.newClient(
                ZK_ADDRESS, new RetryNTimes(10, 5000)
        );
        client.start();
        System.out.println("zk client start successfully!");
        Thread t1 = new Thread(() -> {
            doWithLock(client);
        }, "t1");
        Thread t2 = new Thread(() -> {
            doWithLock(client);
        }, "t2");
        t1.start();
        t2.start();
    }

    private static void doWithLock(CuratorFramework client) {
        InterProcessMutex lock = new InterProcessMutex(client, ZK_LOCK_PATH);
        try {
            if (lock.acquire(10 * 1000, TimeUnit.SECONDS)) {
                System.out.println(Thread.currentThread().getName() + " hold lock");
                Thread.sleep(5000L);
                System.out.println(Thread.currentThread().getName() + " release lock");
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            try {
                lock.release();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }
}

五 Leader 选举

当集群里的某个服务 down 机时，我们可能要从 slave 结点里选出一个作为新的 master，这时就需要一套能在分布式环境中自动协调的 Leader 选举方法。Curator 提供了 LeaderSelector 监听器实现 Leader 选举功能。同一时刻，只有一个 Listener 会进入 takeLeadership() 方法，说明它是当前的 Leader。注意：当 Listener 从 takeLeadership() 退出时就说明它放弃了'Leader 身份'，这时 Curator 会利用 Zookeeper 再从剩余的 Listener 中选出一个新的 Leader。autoRequeue() 方法使放弃 Leadership 的 Listener 有机会重新获得 Leadership，如果不设置的话放弃了的 Listener 是不会再变成 Leader 的。

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.leader.LeaderSelector;
import org.apache.curator.framework.recipes.leader.LeaderSelectorListener;
import org.apache.curator.framework.state.ConnectionState;
import org.apache.curator.retry.RetryNTimes;
import org.apache.curator.utils.EnsurePath;

/**
 * Curator framework's leader election test.
 * Output:
 * LeaderSelector-2 take leadership!
 * LeaderSelector-2 relinquish leadership!
 * LeaderSelector-1 take leadership!
 * LeaderSelector-1 relinquish leadership!
 * LeaderSelector-0 take leadership!
 * LeaderSelector-0 relinquish leadership!
 * ...
 */
public class CuratorLeaderTest {
    /** Zookeeper info */
    private static final String ZK_ADDRESS = "192.168.1.100:2181";
    private static final String ZK_PATH = "/zktest";

    public static void main(String[] args) throws InterruptedException {
        LeaderSelectorListener listener = new LeaderSelectorListener() {
            @Override
            public void takeLeadership(CuratorFramework client) throws Exception {
                System.out.println(Thread.currentThread().getName() + " take leadership!");
                // takeLeadership() method should only return when leadership is being relinquished.
                Thread.sleep(5000L);
                System.out.println(Thread.currentThread().getName() + " relinquish leadership!");
            }

            @Override
            public void stateChanged(CuratorFramework client, ConnectionState state) {
            }
        };
        new Thread(() -> {
            registerListener(listener);
        }).start();
        new Thread(() -> {
            registerListener(listener);
        }).start();
        new Thread(() -> {
            registerListener(listener);
        }).start();
        Thread.sleep(Integer.MAX_VALUE);
    }

    private static void registerListener(LeaderSelectorListener listener) {
        // 1.Connect to zk
        CuratorFramework client = CuratorFrameworkFactory.newClient(
                ZK_ADDRESS, new RetryNTimes(10, 5000)
        );
        client.start();
        // 2.Ensure path
        try {
            new EnsurePath(ZK_PATH).ensure(client.getZookeeperClient());
        } catch (Exception e) {
            e.printStackTrace();
        }
        // 3.Register listener
        LeaderSelector selector = new LeaderSelector(client, ZK_PATH, listener);
        selector.autoRequeue();
        selector.start();
    }
}