本文共 7273 字，大约阅读时间需要 24 分钟。

zookeeper

一、linux客户端命令操作

1.服务端开启

2.客户端开启

3.查询操作

4.创建节点

5.修改节点

6.删除节点

二、使用Curator客户端

1.坐标依赖

<!--curator,zookeeper客户端操作依赖--><dependency>    <groupId>org.apache.curator</groupId>    <artifactId>curator-framework</artifactId>    <version>4.0.0</version></dependency><dependency>    <groupId>org.apache.curator</groupId>    <artifactId>curator-recipes</artifactId>    <version>4.0.0</version></dependency>

2.具体操作

package com.lzb;import org.apache.curator.RetryPolicy;import org.apache.curator.RetrySleeper;import org.apache.curator.framework.CuratorFramework;import org.apache.curator.framework.CuratorFrameworkFactory;import org.apache.curator.framework.api.BackgroundCallback;import org.apache.curator.framework.api.CuratorEvent;import org.apache.curator.retry.ExponentialBackoffRetry;import org.apache.zookeeper.CreateMode;import org.apache.zookeeper.data.Stat;import org.junit.Test;import java.util.Arrays;import java.util.List;public class CuratorTest {       @Test    public void create() throws Exception {           CuratorFramework client = CuratorFrameworkFactory.newClient("192.168.31.81:2181", 60 * 1000, 15 * 1000, new ExponentialBackoffRetry(3000, 3));        client.start();        //不支持名称空间，似乎版本有问题？//        client = client.usingNamespace("test");        //创建节点        client.create().forPath("/test/p7", "hhhh".getBytes());        //删除节点        client.delete().forPath("/test");        //创建临时节点        client.create().withMode(CreateMode.EPHEMERAL).forPath("/test/p1");        //创建多级节点        client.create().creatingParentsIfNeeded().forPath("/test2");        client.close();    }    @Test    public void get() throws Exception {           CuratorFramework client = CuratorFrameworkFactory.newClient("192.168.31.81:2181", 60 * 1000, 15 * 1000, new ExponentialBackoffRetry(3000, 3));        client.start();        //查询数据        System.out.println(new String(client.getData().forPath("/test/p2")));        //查询子节点        List<String> list = client.getChildren().forPath("/");        System.out.println(list.toString());        //查询状态信息        Stat stat = new Stat();        client.getData().storingStatIn(stat).forPath("/test/p2");        System.out.println(stat);        client.close();    }    @Test    public void set() throws Exception {           CuratorFramework client = CuratorFrameworkFactory.newClient("192.168.31.81:2181", 60 * 1000, 15 * 1000, new ExponentialBackoffRetry(3000, 3));        client.start();        //修改数据        client.setData().forPath("/test/p1" ,"set".getBytes());        System.out.println(new String(client.getData().forPath("/test/p1")));        //根据版本修改数据        Stat stat = new Stat();        client.getData().storingStatIn(stat).forPath("/test/p2");        client.setData().withVersion(stat.getVersion()).forPath("/test/p2","版本修改".getBytes());        System.out.println(new String(client.getData().forPath("/test/p2")));        client.close();    }    @Test    public void delete() throws Exception {           CuratorFramework client = CuratorFrameworkFactory.newClient("192.168.31.81:2181", 60 * 1000, 15 * 1000, new ExponentialBackoffRetry(3000, 3));        client.start();        //删除节点        client.delete().forPath("/test/p7");        //删除带有字节点的节点        client.delete().deletingChildrenIfNeeded().forPath("/test/p6");        //保证删除节点以及子节点        client.delete().guaranteed().deletingChildrenIfNeeded().forPath("/test/p6");        //删除后的回调函数        client.delete().guaranteed().deletingChildrenIfNeeded().inBackground(new BackgroundCallback() {               @Override            public void processResult(CuratorFramework client, CuratorEvent curatorEvent) throws Exception {                   System.out.println(curatorEvent);            }        }).forPath("/test/p5");        client.close();    }}

这里名称空间不能设置，有人看到了能回答下吗？

3.节点监听机制

package com.lzb;import org.apache.curator.framework.CuratorFramework;import org.apache.curator.framework.CuratorFrameworkFactory;import org.apache.curator.framework.api.BackgroundCallback;import org.apache.curator.framework.api.CuratorEvent;import org.apache.curator.framework.recipes.cache.*;import org.apache.curator.retry.ExponentialBackoffRetry;import org.apache.zookeeper.CreateMode;import org.apache.zookeeper.data.Stat;import org.junit.Test;import java.util.List;public class CuratorWatcherTest {       @Test//单节点监听    public void nodeCache() throws Exception {           CuratorFramework client = CuratorFrameworkFactory.newClient("192.168.31.81:2181", 60 * 1000, 15 * 1000, new ExponentialBackoffRetry(3000, 3));        client.start();        NodeCache nodeCache = new NodeCache(client, "/test/p1");        nodeCache.getListenable().addListener(new NodeCacheListener() {               @Override            public void nodeChanged() throws Exception {                   System.out.println("/test/p1变化了");                ChildData currentData = nodeCache.getCurrentData();                System.out.println(currentData);                System.out.println(new String(currentData.getData()));            }        });        nodeCache.start(true);        while (true) {           }    }    @Test//子节点监听    public void pathChildCache() throws Exception {           CuratorFramework client = CuratorFrameworkFactory.newClient("192.168.31.81:2181", 60 * 1000, 15 * 1000, new ExponentialBackoffRetry(3000, 3));        client.start();        PathChildrenCache cache = new PathChildrenCache(client, "/test", true);        cache.getListenable().addListener(new PathChildrenCacheListener() {               @Override            public void childEvent(CuratorFramework framework, PathChildrenCacheEvent event) throws Exception {                   System.out.println("/test/变化了");                PathChildrenCacheEvent.Type type = event.getType();                if (type.equals(PathChildrenCacheEvent.Type.CHILD_UPDATED)) {                       System.out.println(new String(event.getData().getData()));                }            }        });        cache.start();        while (true) {           }    }    @Test//节点树监听    public void treeCache() throws Exception {           CuratorFramework client = CuratorFrameworkFactory.newClient("192.168.31.81:2181", 60 * 1000, 15 * 1000, new ExponentialBackoffRetry(3000, 3));        client.start();        TreeCache cache = new TreeCache(client,"/test");        cache.getListenable().addListener(new TreeCacheListener() {               @Override            public void childEvent(CuratorFramework curatorFramework, TreeCacheEvent event) throws Exception {                   System.out.println("节点树变化了");                System.out.println(event);            }        });        cache.start();        while (true) {                   }    }}

4.分布式锁

获取锁原理：所有消费者在同一个节点lock下创建自己的临时顺序节点,再判断自己的锁节点值是否最小,最小则获取到锁。否则监听任意一个更小的锁节点,当其删除时再判断自己的锁节点值,重复此动作。临时是为了能够及时释放锁,顺序是为了为了监听判断。

InterProcessMutex mutex= new InterProcessMutex(client,"/lock");mutex.acquire(3,TimeUnit.SECONDS);mutex.release();

5.集群搭建

集群搭建：在data目录下创建一个myid文件，写入对应编号集群搭建：在zoo.cfg文件内声明集群，以下三个范例集群搭建：server.1=192.168.31.81:2881:3881  集群搭建：server.2=192.168.31.82:2881:3881集群搭建：server.3=192.168.31.83:2881:3881集群角色leader、follower、observerleader：事务操作，同步状态follower：查询操作，事务转发，选举投票observer：查询操作，事务转发，不参与投票

总结

zookeeper是分布式应用程序的协调服务，它具有：①配置中心组件，用来分布公有的配置信息；②分布式锁，用来实现不同机器之间使用同一把锁； ③集群管理，作为注册中心来使用

转载地址：http://agwr.baihongyu.com/