Java客户端

maven依赖

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

应用示例

@Slf4j
public class ConfigCenter {
    private final static  String CONNECT_STR="192.168.109.200:2181";
    private final static Integer  SESSION_TIMEOUT=30*1000;
    private static ZooKeeper zooKeeper=null;
    private static CountDownLatch countDownLatch=new CountDownLatch(1);
    public static void main(String[] args) throws IOException, InterruptedException, KeeperException {
        zooKeeper=new ZooKeeper(CONNECT_STR, SESSION_TIMEOUT, new Watcher() {
            @Override
            public void process(WatchedEvent event) {
                    if (event.getType()== Event.EventType.None
                            && event.getState() == Event.KeeperState.SyncConnected){
                        log.info("连接已建立");
                        countDownLatch.countDown();
                    }
            }
        });
        countDownLatch.await();
        MyConfig myConfig = new MyConfig();
        myConfig.setKey("anykey");
        myConfig.setName("anyName");
        ObjectMapper objectMapper=new ObjectMapper();
        byte[] bytes = objectMapper.writeValueAsBytes(myConfig);
        String s = zooKeeper.create("/myconfig", bytes, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        Watcher watcher = new Watcher() {
            @SneakyThrows
            @Override
            public void process(WatchedEvent event) {
                if (event.getType()== Event.EventType.NodeDataChanged
                        && event.getPath()!=null && event.getPath().equals("/myconfig")){
                    log.info(" PATH:{}  发生了数据变化" ,event.getPath());
                    byte[] data = zooKeeper.getData("/myconfig", this, null);
                    MyConfig newConfig = objectMapper.readValue(new String(data), MyConfig.class);
                    log.info("数据发生变化: {}",newConfig);
                }
            }
        };
        byte[] data = zooKeeper.getData("/myconfig", watcher, null);
        MyConfig originalMyConfig = objectMapper.readValue(new String(data), MyConfig.class);
        log.info("原始数据: {}", originalMyConfig);
        TimeUnit.SECONDS.sleep(Integer.MAX_VALUE);
    }
}

zookeeper api

ZooKeeper(String connectString, int sessionTimeout, Watcher watcher)
ZooKeeper(String connectString, int sessionTimeout, Watcher watcher, ZKClientConfig)
ZooKeeper(String connectString, int sessionTimeout, Watcher watcher, boolean canBeReadOnly, HostProvider)
ZooKeeper(String connectString, int sessionTimeout, Watcher watcher, boolean canBeReadOnly, HostProvider, ZKClientConfig)
ZooKeeper(String connectString, int sessionTimeout, Watcher watcher, boolean canBeReadOnly)
ZooKeeper(String connectString, int sessionTimeout, Watcher watcher, boolean canBeReadOnly, ZKClientConfig)
ZooKeeper(String connectString, int sessionTimeout, Watcher watcher, long, byte[])
ZooKeeper(String connectString, int sessionTimeout, Watcher watcher, long, byte[], boolean, HostProvider)
ZooKeeper(String connectString, int sessionTimeout, Watcher watcher, long, byte[], boolean, HostProvider, ZKClientConfig)
ZooKeeper(String connectString, int  sessionTimeout, Watcher watcher, long, byte[], boolean)

参数解释

参数名称	含义
connectString	ZooKeeper服务器列表，由英文逗号分开的host:port字符串组成，每一个都代表一台ZooKeeper机器，如，host1:port1,host2:port2,host3:port3。另外，也可以在connectString中设置客户端连接上ZooKeeper 后的根目录，方法是在host:port字符串之后添加上这个根目录，例如,host1:port1,host2:port2,host3:port3/zk-base,这样就指定了该客户端连接上ZooKeeper服务器之后,所有对ZooKeeper 的操作，都会基于这个根目录。例如，客户端对/sub-node 的操作，最终创建 /zk-node/sub-node, 这个目录也叫Chroot，即客户端隔离命名空间。
sessionTimeout	会话的超时时间，单位毫秒。在ZooKeeper中有会话的概念，在一个会话周期内，ZooKeeper客户端和服务器之间会通过心跳检测机制来维持会话的有效性，一旦在sessionTimeout时间内没有进行有效的心跳检测，会话就会失效。
watcher	Watcher事件通知处理器。可以为null
canBeReadOnly	boolean，用于标识当前会话是否支持“read-only(只读)”模式。默认情况下，在ZooKeeper集群中，一个机器如果和集群中过半及以上机器失去了网络连接，那么这个机器将不再处理客户端请求（包括读写请求)。但是在某些使用场景下，当ZooKeeper服务器发生此类故障的时候，我们还是希望ZooKeeper服务器能够提供读服务（当然写服务肯定无法提供）—— 这就是 ZooKeeper的“read-only”模式。
sessionId和 ses sionPasswd	分别代表会话ID和会话秘钥。这两个参数能够唯一确定一个会话，同时客户端使用这两个参数可以实现客户端会话复用，从而达到恢复会话的效果。具体使用方法是，第一次连接上ZooKeeper服务器时，通过调用ZooKeeper对象实例的以下两个接口，即可获得当前会话的ID和秘钥: long getSessionId(); byte[]getSessionPasswd( ); 荻取到这两个参数值之后，就可以在下次创建ZooKeeper对象实例的时候传入构造方法了

同步创建节点

@Test
public void createTest() throws KeeperException, InterruptedException {
    String path = zooKeeper.create(ZK_NODE, "data".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
    log.info("created path: {}",path);
}

异步创建节点

@Test
public void createAsycTest() throws InterruptedException {
     zooKeeper.create(ZK_NODE, "data".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE,
             CreateMode.PERSISTENT,
             (rc, path, ctx, name) -> log.info("rc  {},path {},ctx {},name {}",rc,path,ctx,name),"context");
    TimeUnit.SECONDS.sleep(Integer.MAX_VALUE);
}

修改节点

@Test
public void setTest() throws KeeperException, InterruptedException {
    Stat stat = new Stat();
    byte[] data = zooKeeper.getData(ZK_NODE, false, stat);
    log.info("修改前: {}",new String(data));
    zooKeeper.setData(ZK_NODE, "changed!".getBytes(), stat.getVersion());
     byte[] dataAfter = zooKeeper.getData(ZK_NODE, false, stat);
    log.info("修改后: {}",new String(dataAfter));
}

Apache curator 客户端

推荐使用
封装了如下功能

Leader选举
分布式计数器

分布式锁

引入mean

<dependency>
 <groupId>org.apache.curator</groupId>
 <artifactId>curator-recipes</artifactId>
 <version>5.0.0</version>
 <exclusions>
     <exclusion>
         <groupId>org.apache.zookeeper</groupId>
         <artifactId>zookeeper</artifactId>
     </exclusion>
 </exclusions>
</dependency>

核心类

CuratorFramework
一个zk客户端对象

创建客户端代码

@Before
public void init() {
    RetryPolicy retryPolicy = new ExponentialBackoffRetry(5000, 30);
    curatorFramework = CuratorFrameworkFactory.builder().connectString("192.168.109.200:2181")
        //重试策略 
        .retryPolicy(retryPolicy)
        // session超时时间，作用在服务端
        .sessionTimeoutMs(sessionTimeoutMs)
        // 链接时间，作用在客户端
        .connectionTimeoutMs(connectionTimeoutMs)
        .canBeReadOnly(true)
        .build();
    curatorFramework.getConnectionStateListenable().addListener((client, newState) -> {
        if (newState == ConnectionState.CONNECTED) {
            log.info("连接成功！");
        }
    });
    log.info("连接中......");
    curatorFramework.start();
}

RetryPolicy : 重试策略

策略名称	描述
ExponentialBackoffRetry	重试一组次数，重试之间的睡眠时间增加
RetryNTimes	重试最大次数
RetryOneTime	只重试一次
RetryUntilElapsed	在给定的时间结束之前重试

创建节点

// forPath对应路径，信息 
String path = curatorFramework.create().forPath("/curator-node");
//  withMode :指定节点类型持久还是临时等，
curatorFramework.create().withMode(CreateMode.PERSISTENT).forPath("/curator-node","some-data".getBytes());
// 创建带层级的节点
String path = curatorFramework.create().creatingParentsIfNeeded().forPath(pathWithParent);
    log.info("curator create node :{}  successfully.",path);

获取数据

//获取的是字节数组
byte[] bytes = curatorFramework.getData().forPath("/curator-node");
log.info("get data from  node :{}  successfully.",new String(bytes));
//inBackground 异步获取
//executorService 指定异步的线程池，可以不执行
ExecutorService executorService = Executors.newSingleThreadExecutor();
String ZK_NODE="/zk-node";
byte[] bytes = curatorFramework.getData().inBackground((client, event) -> {
    log.info(" background: {}", event);
}, executorService).forPath(ZK_NODE);

更新节点

 curatorFramework.setData().forPath("/curator-node","changed!".getBytes());

删除节点

String pathWithParent="/node-parent";
//guaranteed : 只要改客户端回话有效，就会在后台持续发请求，确保删除
//deletingChildrenIfNeeded : 递归删除子节点
curatorFramework.delete().guaranteed().deletingChildrenIfNeeded().forPath(pathWithParent);

异步接口

BackgroundCallback

public interface BackgroundCallback
{
    /**
     * Called when the async background operation completes
     *
     * @param client the client
     * @param event operation result details
     * @throws Exception errors
     */
    public void processResult(CuratorFramework client, CuratorEvent event) throws Exception;
}

Curator 监听器：

public interface CuratorListener{
    /**
     * Called when a background task has completed or a watch has triggered
     *
     * @param client client
     * @param event the event
     * @throws Exception any errors
     */
    public void eventReceived(CuratorFramework client, CuratorEvent event) throws Exception;
}

inBackground 方法会异步获得监听

Curator Caches:

服务端事件监听，本地缓存视图与服务端视图的对比过程
可以反复注册

**nodeCache 节点监听

NodeCache nodeCache = new NodeCache(curatorFramework, NODE_CACHE);
nodeCache.getListenable().addListener(new NodeCacheListener() {
    @Override
    public void nodeChanged() throws Exception {
        log.info("{} path nodeChanged: ",NODE_CACHE);
        CuratorFramework curatorFramework = getCuratorFramework();
        byte[] bytes = curatorFramework.getData().forPath(NODE_CACHE);
        log.info("data: {}",new String(bytes));
    }
});
nodeCache.start();

**path cache子节点监听

PathChildrenCache pathChildrenCache = new PathChildrenCache(curatorFramework, PATH, true);
pathChildrenCache.getListenable().addListener(new PathChildrenCacheListener() {
    @Override
    public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception {
        log.info("event:  {}",event);
    }
});
// 如果设置为true则在首次启动时就会缓存节点内容到Cache中
pathChildrenCache.start(true);

3. tree cache:
treeNode维护树结构，可以递归子节点数据编号

@Test
    public void testTreeCache() throws Exception {
        createIfNeed(TREE_CACHE);
        TreeCache treeCache = new TreeCache(curatorFramework, TREE_CACHE);
        treeCache.getListenable().addListener(new TreeCacheListener() {
            @Override
            public void childEvent(CuratorFramework client, TreeCacheEvent event) throws Exception {
                log.info(" tree cache: {}",event);
            }
        });
        treeCache.start();
    }

Watcher 监听

private static  Watcher watcher = new Watcher() {
    @Override
    public void process(WatchedEvent event) {
        if (event.getType() == Event.EventType.None
            && event.getState() == Event.KeeperState.SyncConnected){
            countDownLatch.countDown();
            log.info(" 连接建立");
            // start to watch config
            try {
                log.info(" 开始监听：{}",ZooDefs.CONFIG_NODE);
                zookeeper.getConfig(true,null);
            } catch (KeeperException e) {
                e.printStackTrace();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }else if( event.getPath()!=null  &&  event.getPath().equals(ZooDefs.CONFIG_NODE)){
            try {
                byte[] config = zookeeper.getConfig(this, null);
                String clientConfigStr = ConfigUtils.getClientConfigStr(new String(config));
                log.info(" 配置发生变更: {}",clientConfigStr);
                zookeeper.updateServerList(clientConfigStr.split(" ")[1]);
            } catch (KeeperException e) {
                e.printStackTrace();
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
   };

zookeeper集群

三种类型的角色

Leader 处理写城区，可以处理读请求只能有一个
Follower 只能处理读请求，可以参与节点选举
Observer：只能处理读请求，不可以参与选举，不参与过半写成功

1：修改配置文件

vim conf/zoo1.cfg dataDir=/usr/local/data/zookeeper-1 clientPort=2181 //客户端链接的端口 server.1=127.0.0.1:2001:3001:participant// participant 可以不用写，默认就是participant server.2=127.0.0.1:2002:3002:participant //前面一个是通讯端口，后面一个是选举端口 server.3=127.0.0.1:2003:3003:participant server.4=127.0.0.1:2004:3004:observer

2：标识ServerId

在 dataDir=/usr/local/data/zookeeper-1 路径下
创建myid文件，写入配置的 server.1 中的1

3：依次创建其他文件

以及zoo2.cfg zoo3.cfg zoo4.cfg

4：启动

bin/zkServer.sh start conf/zoo1.cfg bin/zkServer.sh start conf/zoo2.cfg bin/zkServer.sh start conf/zoo3.cfg

5：检测是否启动

zkServer.sh status conf/zoo1.cfg

链接客户端 bin/zkCli.sh -server ip1:port1,ip2:port2,ip3:port3

Zookeeper 3.5.0 新特性：集群动态配置

Java 进阶

Zookeeper客户端与集群

Java客户端

maven依赖

应用示例

zookeeper api

Apache curator 客户端

引入mean

核心类

创建客户端代码

创建节点

更新节点

删除节点

异步接口

Curator 监听器：

Curator Caches:

Watcher 监听

zookeeper集群

1：修改配置文件

2：标识ServerId

3：依次创建其他文件

4：启动

5：检测是否启动

Zookeeper 3.5.0 新特性：集群动态配置

Zookeeper客户端与集群

Java客户端

maven依赖

应用示例

zookeeper api

Apache curator 客户端

引入mean

核心类

创建客户端代码

创建节点

更新节点

删除节点

异步接口

Curator 监听器：

Curator Caches:

Watcher 监听

zookeeper集群

1：修改配置文件

2：标识ServerId

3：依次创建其他文件

4：启动

5：检测是否启动

Zookeeper 3.5.0 新特性： 集群动态配置

Zookeeper 3.5.0 新特性：集群动态配置