/data/host1
, /data/host2
, 这个路径也可以理解为是 Znode 的 Name/data/host1
, 其值是一个字符串 "192.168.0.1"
Zookeeper集群是一个基于主从框架的高可用集群
每个服务器承担如下三种角色中的一种
Leader一个Zookeeper集群同一时间只会有一个实际工作的Leader,它会发起并维护与各Follwer及Observer间的心跳。所有的写操作必须要通过Leader完成再有Leader将写操作广播给其他服务器。
Follower一个Zookeeper集群可能同时存在多个Follower,它会响应Leader的心跳。
follower可直接处理并返回客户端的读请求,同时会将写请求转发给Leader处理,并且负责在Leader处理写请求时对请求进行投票。
Observer角色与Follower类似,但是无投票权。
数据发布/订阅系统,需要发布者将数据发布到Zookeeper的节点上,供订阅者进行数据订阅,进而达到动态获取数据的目的,实现配置信息的集中式管理和数据的动态更新。
发布/订阅一般有两种设计模式:推模式和拉模式,服务端主动将数据更新发送给所有订阅的客户端称为推模式;客户端主动请求获取最新数据称为拉模式.
Zookeeper采用了推拉相结合的模式,客户端向服务端注册自己需要关注的节点,一旦该节点数据发生变更,那么服务端就会向相应的客户端推送Watcher事件通知,客户端接收到此通知后,主动到服务端获取最新的数据。
命名服务是分步实现系统中较为常见的一类场景,分布式系统中,被命名的实体通常可以是集群中的机器、提供的服务地址或远程对象等,通过命名服务,客户端可以根据指定名字来获取资源的实体,在分布式环境中,上层应用仅仅需要一个全局唯一的名字。Zookeeper可以实现一套分布式全局唯一ID的分配机制。
通过调用Zookeeper节点创建的API接口就可以创建一个顺序节点,并且在API返回值中会返回这个节点的完整名字,利用此特性,可以生成全局ID,其步骤如下
1. 客户端根据任务类型,在指定类型的任务下通过调用接口创建一个顺序节点,如"job-"。
2. 创建完成后,会返回一个完整的节点名,如"job-00000001"。
3. 客户端拼接type类型和返回值后,就可以作为全局唯一ID了,如"type2-job-00000001"。
Zookeeper中特有的Watcher注册于异步通知机制,能够很好地实现分布式环境下不同机器,甚至不同系统之间的协调与通知,从而实现对数据变更的实时处理。通常的做法是不同的客户端都对Zookeeper上的同一个数据节点进行Watcher注册,监听数据节点的变化(包括节点本身和子节点),若数据节点发生变化,那么所有订阅的客户端都能够接收到相应的Watcher通知,并作出相应处理。
在绝大多数分布式系统中,系统机器间的通信无外乎心跳检测、工作进度汇报和系统调度。
① 心跳检测,不同机器间需要检测到彼此是否在正常运行,可以使用Zookeeper实现机器间的心跳检测,基于其临时节点特性(临时节点的生存周期是客户端会话,客户端若当即后,其临时节点自然不再存在),可以让不同机器都在Zookeeper的一个指定节点下创建临时子节点,不同的机器之间可以根据这个临时子节点来判断对应的客户端机器是否存活。通过Zookeeper可以大大减少系统耦合。
② 工作进度汇报,通常任务被分发到不同机器后,需要实时地将自己的任务执行进度汇报给分发系统,可以在Zookeeper上选择一个节点,每个任务客户端都在这个节点下面创建临时子节点,这样不仅可以判断机器是否存活,同时各个机器可以将自己的任务执行进度写到该临时节点中去,以便中心系统能够实时获取任务的执行进度。
③ 系统调度,Zookeeper能够实现如下系统调度模式:分布式系统由控制台和一些客户端系统两部分构成,控制台的职责就是需要将一些指令信息发送给所有的客户端,以控制他们进行相应的业务逻辑,后台管理人员在控制台上做一些操作,实际上就是修改Zookeeper上某些节点的数据,Zookeeper可以把数据变更以时间通知的形式发送给订阅客户端。
分布式锁用于控制分布式系统之间同步访问共享资源的一种方式,可以保证不同系统访问一个或一组资源时的一致性,主要分为排它锁和共享锁。
排它锁又称为写锁或独占锁,若事务T1对数据对象O1加上了排它锁,那么在整个加锁期间,只允许事务T1对O1进行读取和更新操作,其他任何事务都不能再对这个数据对象进行任何类型的操作,直到T1释放了排它锁。
① 获取锁,在需要获取排它锁时,所有客户端通过调用接口,在/exclusive_lock节点下创建临时子节点/exclusive_lock/lock。Zookeeper可以保证只有一个客户端能够创建成功,没有成功的客户端需要注册/exclusive_lock节点监听。
② 释放锁,当获取锁的客户端宕机或者正常完成业务逻辑都会导致临时节点的删除,此时,所有在/exclusive_lock节点上注册监听的客户端都会收到通知,可以重新发起分布式锁获取。
共享锁又称为读锁,若事务T1对数据对象O1加上共享锁,那么当前事务只能对O1进行读取操作,其他事务也只能对这个数据对象加共享锁,直到该数据对象上的所有共享锁都被释放。在需要获取共享锁时,所有客户端都会到/shared_lock下面创建一个临时顺序节点
有一些时候,多个团队需要共同完成一个任务,比如,A团队将Hadoop集群计算的结果交给B团队继续计算,B完成了自己任务再交给C团队继续做。这就有点像业务系统的工作流一样,一环一环地传下 去.
分布式环境下,我们同样需要一个类似单进程队列的组件,用来实现跨进程、跨主机、跨网络的数据共享和数据传递,这就是我们的分布式队列。
Leader选举是保证分布式数据一致性的关键所在。当Zookeeper集群中的一台服务器出现以下两种情况之一时,需要进入Leader选举。
若进行Leader选举,则至少需要两台机器,这里选取3台机器组成的服务器集群为例。在集群初始化阶段,当有一台服务器Server1启动时,其单独无法进行和完成Leader选举,当第二台服务器Server2启动时,此时两台机器可以相互通信,每台机器都试图找到Leader,于是进入Leader选举过程。选举过程如下
(1) 每个Server发出一个投票。由于是初始情况,Server1和Server2都会将自己作为Leader服务器来进行投票,每次投票会包含所推举的服务器的myid和ZXID,使用(myid, ZXID)来表示,此时Server1的投票为(1, 0),Server2的投票为(2, 0),然后各自将这个投票发给集群中其他机器。
(2) 接受来自各个服务器的投票。集群的每个服务器收到投票后,首先判断该投票的有效性,如检查是否是本轮投票、是否来自LOOKING状态的服务器。
(3) 处理投票。针对每一个投票,服务器都需要将别人的投票和自己的投票进行PK,PK规则如下
· 优先检查ZXID。ZXID比较大的服务器优先作为Leader。
· 如果ZXID相同,那么就比较myid。myid较大的服务器作为Leader服务器。
对于Server1而言,它的投票是(1, 0),接收Server2的投票为(2, 0),首先会比较两者的ZXID,均为0,再比较myid,此时Server2的myid最大,于是更新自己的投票为(2, 0),然后重新投票,对于Server2而言,其无须更新自己的投票,只是再次向集群中所有机器发出上一次投票信息即可。
(4) 统计投票。每次投票后,服务器都会统计投票信息,判断是否已经有过半机器接受到相同的投票信息,对于Server1、Server2而言,都统计出集群中已经有两台机器接受了(2, 0)的投票信息,此时便认为已经选出了Leader。
(5) 改变服务器状态。一旦确定了Leader,每个服务器就会更新自己的状态,如果是Follower,那么就变更为FOLLOWING,如果是Leader,就变更为LEADING。
在Zookeeper运行期间,Leader与非Leader服务器各司其职,即便当有非Leader服务器宕机或新加入,此时也不会影响Leader,但是一旦Leader服务器挂了,那么整个集群将暂停对外服务,进入新一轮Leader选举,其过程和启动时期的Leader选举过程基本一致过程相同。
集群规划
服务器IP | 主机名 | myid的值 |
---|---|---|
192.168.174.10 | hadoop01 | 1 |
192.168.174.11 | hadoop02 | 2 |
192.168.174.12 | hadoop03 | 3 |
第一步:下载zookeeeper的压缩包,下载网址如下
我们在这个网址下载我们使用的zk版本为3.4.9
下载完成之后,上传到我们的linux的/root路径下准备进行安装
第二步:解压
解压zookeeper的压缩包到/myapp/Zookeeper路径下去,然后准备进行安装
tar -zxf /root/zookeeper-3.4.9.tar.gz -C /myapp/Zookeeper/
第三步:修改配置文件
第一台机器修改配置文件
cd /myapp/Zookeeper/zookeeper-3.4.9/conf/
cp zoo_sample.cfg zoo.cfg
mkdir -p /myapp/Zookeeper/zookeeper-3.4.9/zkdatas/
vim zoo.cfg
dataDir=/myapp/Zookeeper/zookeeper-3.4.9/zkdatas/
# 保留多少个快照
autopurge.snapRetainCount=3
# 日志多少小时清理一次
autopurge.purgeInterval=1
# 集群中服务器地址
server.1=hadoop01:2888:3888
server.2=hadoop02:2888:3888
server.3=hadoop03:2888:3888
第四步:添加myid配置
在第一台机器的
/export/servers/zookeeper-3.4.9/zkdatas /这个路径下创建一个文件,文件名为myid ,文件内容为1
echo 1 > /myapp/Zookeeper/zookeeper-3.4.9/zkdatas/myid
第五步:安装包分发并修改myid的值
安装包分发到其他机器的前提是:其他主机也要有相应的目录。
第一台机器上面执行以下两个命令
scp -r /myapp/Zookeeper/zookeeper-3.4.9/ node02:/myapp/Zookeeper/
scp -r /myapp/Zookeeper/zookeeper-3.4.9/ node03:/myapp/Zookeeper/
第二台机器上修改myid的值为2
echo 2 > /myapp/Zookeeper/zookeeper-3.4.9/zkdatas/myid
第三台机器上修改myid的值为3
echo 3 > /myapp/Zookeeper/zookeeper-3.4.9/zkdatas/myid
第六步:三台机器启动zookeeper服务
三台机器启动zookeeper服务
这个命令三台机器都要执行
/myapp/Zookeeper/zookeeper-3.4.9/bin/zkServer.sh start
查看启动状态
/myapp/Zookeeper//zookeeper-3.4.9/bin/zkServer.sh status
#!/bin/bash
if [ -d /myapp/Zookeeper/zookeeper-3.4.9 ]; then
echo "Zookeeper已经安装"
else
echo "请输入一共多少主机"
read rootsum
echo "请分别输入主机名"
for ((i=1; i<=${rootsum}; i++)) {
echo "请输入第${i}台主机名:"
read rootname[i]
ssh ${rootname[i]} "mkdir -p /myapp/Zookeeper"
}
echo "请输入Zookeeper压缩包路径"
read newfile
echo "正在为你安装Zookeeper..........."
tar -zxf $newfile -C /myapp/Zookeeper/
cp /myapp/Zookeeper/zookeeper-3.4.9/conf/zoo_sample.cfg /myapp/Zookeeper/zookeeper-3.4.9/conf/zoo.cfg
sed -i -e '12d' /myapp/Zookeeper/zookeeper-3.4.9/conf/zoo.cfg
mkdir -p /myapp/Zookeeper/zookeeper-3.4.9/zkdatas/
arr1=("dataDir=/myapp/Zookeeper/zookeeper-3.4.9/zkdatas" "# 保留多少个快照" "autopurge.snapRetainCount=3" "# 日志多少小时清理一次" "autopurge.purgeInterval=1" "# 集群中服务器地址" "server.1=hadoop01:2888:3888" "server.2=hadoop02:2888:3888" "server.3=hadoop03:2888:3888")
#echo ${#arr1[*]}
for (( i=0; i<=${#arr1[*]}; i++)) {
echo ${arr1[$i]} >> /myapp/Zookeeper/zookeeper-3.4.9/conf/zoo.cfg
}
echo 1 >> /myapp/Zookeeper/zookeeper-3.4.9/zkdatas/myid
echo "==============================================================================="
echo "==============================================================================="
echo "==============================================================================="
for ((e=1; e<=${rootsum}; e++)){
if test $[ e + 1 ] -gt $rootsum; then
ssh ${rootname[e]} sed -i -e s/1/${e}/ /myapp/Zookeeper/zookeeper-3.4.9/zkdatas/myid
else
scp -r /myapp/Zookeeper/zookeeper-3.4.9 ${rootname[e + 1]}:/myapp/Zookeeper
ssh ${rootname[e]} sed -i -e s/1/${e}/ /myapp/Zookeeper/zookeeper-3.4.9/zkdatas/myid
fi
}
nl /myapp/Zookeeper/zookeeper-3.4.9/conf/zoo.cfg
for i in ${rootname[*]};
do
ssh $i "echo '#Zookeeper' >> /etc/profile"
ssh $i "echo 'export ZOOKEEPER_HOME=/myapp/Zookeeper/zookeeper-3.4.9' >> /etc/profile"
ssh $i "echo 'export PATH=\$PATH:\$ZOOKEEPER_HOME/bin' >> /etc/profile"
done
echo "==================================================================================="
echo "Zookeeper 安装已完成"
fi
但也有不同之处:
9.1、
Zonde有两种,分别为临时节点和永久节点。节点的类型在创建时即被确定并且不能改变。
9.2、
Zonde还有一个序列化的特性,如果创建的时候指定的话,该Znode的名字后面会自动追加一个不断增加的序列号。序列号对于此节点的父节点来说是唯一的,这样便会记录每个子节点创建的先后顺序。它的格式为"%10d"(10位数字,没有数值的数位用0补充,例如"0000000001")
9.3、
这样便会存在四种类型的Zonde节点,分别对应:
bin/zkCli.sh -server hadoop01:2181
命令 | 说明 | 参数 |
---|---|---|
create [-s] [-e] path data acl |
创建Znode | -s 指定是顺序节点 -e 指定是临时节点 |
ls path [watch] |
列出Path下所有子Znode | |
get path [watch] |
获取Path对应的Znode的数据和属性 | |
ls2 path [watch] |
查看Path下所有子Znode以及子Znode的属性 | |
set path data [version] |
更新节点 | version 数据版本 |
delete path [version] |
删除节点, 如果要删除的节点有子Znode则无法删除 | version 数据版本 |
rmr path |
删除节点, 如果有子Znode则递归删除 | |
setquota -n|-b val path |
修改Znode配额 | -n 设置子节点最大个数 -b 设置节点数据最大长度 |
history |
列出历史记录 |
create /app1 hello
create -s /app3 world
create -s /app3 world
create -s -e /tempnode2 aaa
get /app1
set /app1 xxx
delete /app1 删除的节点不能有子节点
rmr /app1 递归删除
Znode
Znode
, 需要使用路径的形式, 例如 /test1/test11
Znode
中数据是有大小限制的, 最大只能为1M
Znode
是由三个部分构成
stat
: 状态, Znode的权限信息, 版本等data
: 数据, 每个Znode都是可以携带数据的, 无论是否有子节点children
: 子节点列表Znode
有两大特性, 可以构成四种不同类型的Znode
持久
客户端断开时, 不会删除持有的Znode临时
客户端断开时, 删除所有持有的Znode, 临时Znode不允许有子Znode有序
创建的Znode有先后顺序, 顺序就是在后面追加一个序列号, 序列号是由父节点管理的自增无序
创建的Znode没有先后顺序每个Znode都包含一系列的属性,通过命令get,可以获得节点的属性。
dataVersion
:数据版本, 每次当Znode
中的数据发生变化的时候, dataVersion
都会增加1(即使设置的是相同的数据),可有效避免了数据更新时出现的先后顺序问题。cversion
: 节点版本, 每次当Znode
的节点发生变化的时候, cversion
都会增加1。aclVersion
:ACL(Access Control List)
的版本号, 当Znode
的权限信息发生变化的时候aclVersion会自增cZxid
:zonde创建的事务IDmZxid
:Zonde被修改的事务id,即每次对znode的修改都会更新mZxid。
ctime
:节点创建的时间戳。mtime
:节点最新 一次更新发生的时间戳。ephemeralOwner
:如果Znode
为临时节点, ephemeralOwner
表示与该节点关联的SessionId
,如果不是临时节点ephemeralOwner值为0。Session
中的, 客户端和服务器创建一个连接的时候同时也会创建一个Session
Session
会在不同的状态之间进行切换: CONNECTING
, CONNECTED
, RECONNECTING
, RECONNECTED
, CLOSED
Watcher
, 当Znode发生变化的时候, WatchManager
会调用对应的Watcher
Watcher
会得到通知Watcher
的特点
Watcher
只会被触发一次, 如果需要继续监听, 则需要再次添加 Watcher
Watcher
得到的事件是被封装过的, 包括三个内容 keeperState, eventType, path
KeeperState | EventType | 触发条件 | 说明 |
---|---|---|---|
None | 连接成功 | ||
SyncConnected | NodeCreated | Znode被创建 | 此时处于连接状态 |
SyncConnected | NodeDeleted | Znode被删除 | 此时处于连接状态 |
SyncConnected | NodeDataChanged | Znode数据被改变 | 此时处于连接状态 |
SyncConnected | NodeChildChanged | Znode的子Znode数据被改变 | 此时处于连接状态 |
Disconnected | None | 客户端和服务端断开连接 | 此时客户端和服务器处于断开连接状态 |
Expired | None | 会话超时 | 会收到一个SessionExpiredException |
AuthFailed | None | 权限验证失败 | 会收到一个AuthFailedException |
这里操作Zookeeper的JavaAPI使用的是一套zookeeper客户端框架Curator,解决了很多Zookeeper客户端非常底层的细节开发工作。
Curator包含了几个包:
Maven依赖(使用curator的版本:2.12.0,对应Zookeeper的版本为:3.4.x,如果跨版本会有兼容性问题,很有可能导致节点操作失败);
创建maven java工程,导入jar包
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>org.example</groupId>
<artifactId>Zookeeper_JavaAPI</artifactId>
<version>1.0-SNAPSHOT</version>
<dependencies>
<dependency>
<groupId>org.apache.curator</groupId>
<artifactId>curator-framework</artifactId>
<version>2.12.0</version>
</dependency>
<dependency>
<groupId>org.apache.curator</groupId>
<artifactId>curator-recipes</artifactId>
<version>2.12.0</version>
</dependency>
<dependency>
<groupId>com.google.collect</groupId>
<artifactId>google-collections</artifactId>
<version>1.0</version>
</dependency>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<version>RELEASE</version>
</dependency>
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-simple</artifactId>
<version>1.7.25</version>
</dependency>
<dependency>
<groupId>org.apache.zookeeper</groupId>
<artifactId>zookeeper</artifactId>
<!-- same version with the zookeeper-->
<version>3.4.9</version>
</dependency>
</dependencies>
</project>
package cn.itcast.zookeeper_api;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.junit.Test;
public class ZookeeperAPITest {
@Test
public void createZnode() throws Exception {
//1. 定制一个重试策略
/*
param1:重试的间隔时间
param2:重试的最大次数
*/
RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000,1);
//2. 获取一个客户端对象
/*
param1:要连接的Zookeeper服务器列表
param2:会话的超时时间
param3:链接超时时间
param4:重试策略
*/
String connectionStr="192.168.1.10:2181,192.168.1.11:2181,192.168.1.12:2181";
CuratorFramework client = CuratorFrameworkFactory.newClient(connectionStr,8000,8000,retryPolicy);
//3。开启客户端
client.start();
//4. 创建永久节点
/*
PERSISTENT:永久节点
PERSISTENT_SEQUENTIAL:永久序列化节点
EPHEMERAL:临时节点
EPHEMERAL_SEQUENTIAL:临时序列化节点
*/
client.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT).forPath("/world01","world".getBytes());
//5. 关闭客户端
client.close();
}
}
package cn.itcast.zookeeper_api;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.junit.Test;
public class ZookeeperAPITest {
@Test
public void createZnode() throws Exception {
//1. 定制一个重试策略
/*
param1:重试的间隔时间
param2:重试的最大次数
*/
RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000,1);
//2. 获取一个客户端对象
/*
param1:要连接的Zookeeper服务器列表
param2:会话的超时时间
param3:链接超时时间
param4:重试策略
*/
String connectionStr="192.168.1.10:2181,192.168.1.11:2181,192.168.1.12:2181";
CuratorFramework client = CuratorFrameworkFactory.newClient(connectionStr,8000,8000,retryPolicy);
//3。开启客户端
client.start();
//4. 创建临时节点
/*
PERSISTENT:永久节点
PERSISTENT_SEQUENTIAL:永久序列化节点
EPHEMERAL:临时节点
EPHEMERAL_SEQUENTIAL:临时序列化节点
*/
client.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL).forPath("/world01","world".getBytes());
Thread.sleep(5000);
//5. 关闭客户端
client.close();
}
}
package cn.itcast.zookeeper_api;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.junit.Test;
public class ZookeeperAPITest {
@Test
public void SetZnodeData() throws Exception {
//1. 定制一个重试策略
/*
param1:重试的间隔时间
param2:重试的最大次数
*/
RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000,1);
//2. 获取一个客户端对象
/*
param1:要连接的Zookeeper服务器列表
param2:会话的超时时间
param3:链接超时时间
param4:重试策略
*/
String connectionStr="192.168.1.10:2181,192.168.1.11:2181,192.168.1.12:2181";
CuratorFramework client = CuratorFrameworkFactory.newClient(connectionStr,8000,8000,retryPolicy);
//3。开启客户端
client.start();
//4. 修改节点数据
client.setData().forPath("/hell04","hello7".getBytes());
//5. 关闭客户端
client.close();
}
}
package cn.itcast.zookeeper_api;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.junit.Test;
public class ZookeeperAPITest {
@Test
public void getZnodeData() throws Exception {
//1. 定制一个重试策略
RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000,1);
//2. 获取客户端
String conectionStr="192.168.1.10:2181,192.168.1.11:2181,192.168.1.12:2181,";
CuratorFramework client = CuratorFrameworkFactory.newClient(conectionStr,8000,8000,retryPolicy);
//3. 启动客户端
client.start();
//4. 获取节点数据
byte[] bytes = client.getData().forPath("/app1");
System.out.println(new String(bytes));
//5. 关闭客户端
client.close();
}
}
package cn.itcast.zookeeper_api;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.junit.Test;
public class ZookeeperAPITest {
@Test
public void WatchZnode() throws Exception {
//1. 定制一个重试策略
RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000,1);
//2. 获取客户端
String conectionStr="192.168.1.10:2181,192.168.1.11:2181,192.168.1.12:2181,";
CuratorFramework client = CuratorFrameworkFactory.newClient(conectionStr,8000,8000,retryPolicy);
//3. 启动客户端
client.start();
//4. 创建一个TreeCache对象,指定要监控的节点路径
final TreeCache treeCache = new TreeCache(client, "/hell04");
//5. 自定义一个监听器
treeCache.getListenable().addListener(new TreeCacheListener() {
public void childEvent(CuratorFramework curatorFramework, TreeCacheEvent treeCacheEvent) throws Exception {
ChildData data = treeCacheEvent.getData();
if (data != null) {
switch (treeCacheEvent.getType()){
case NODE_ADDED:
System.out.println("监控到有新增节点!");
break;
case NODE_REMOVED:
System.out.println("监控到有节点被移除!");
break;
case NODE_UPDATED:
System.out.println("监控到节点被更新!");
break;
default:
break;
}
}
}
});
//6. 开始监听
treeCache.start();
//7. 设置休眠时间
Thread.sleep(100000000);
//8. 关闭客户端
client.close();
}
}
联系客服