Zookeeper之Zkclic+Zkcient+Curator基本使用

冷不防 2022-05-25 06:36 170阅读 0赞

===============================================zookeeper下载和搭建==================================================  
安装和配置：  
官网：http://zookeeper.apache.org/  
找到一个stable版本的下载地址：  
wget http://mirrors.hust.edu.cn/apache/zookeeper/stable/zookeeper-3.4.10.tar.gz  
  
解压：tar -zxvf zookeeper-3.4.10.tar.gz  
  
在同一个机器中配置三个实例（Ensemble）：  
\# cd zookeeper-3.4.10/conf/  
  
配置第一个：  
\# cp zoo\_sample.cfg zk1.cfg  
\# vi  zk1.cfg  
配置内容：  
clientPort=2181  
dataDir=/var/lib/zookeeper/1  
server.1=localhost:2888:3888  
server.2=localhost:2889:3889  
server.3=localhost:2890:3890  
  
各个参数的意义：  
  
    tickTime：心跳检测的时间间隔（毫秒），缺省：2000  
    clientPort：其他应用（比如solr）访问ZooKeeper的端口，缺省：2181  
    initLimit：初次同步的阶段（followers连接到leader的阶段），允许的时长（tick数量），缺省：10  
    syncLimit：允许followers同步到ZooKeeper的时长（tick数量），缺省：5  
    dataDir：数据（比如所管理的配置文件）的存放路径，初始时应该为空  
    server.X：X是ensemble中一个服务器的id，后面指定该server的hostname、  
    第一个端口号用于ZooKeeper之间的通信、  
    第二个端口用于和其他应用之间的通信  
  
  
配置第二个：  
\# cp zk1.cfg zk2.cfg  
修改：dataDir=/var/lib/zookeeper/2  
修改：clientPort=2182  
  
配置第三个：  
\# cp zk1.cfg zk3.cfg  
修改：dataDir=/var/lib/zookeeper/3  
修改：clientPort=2183  
  
创建相应的dataDir目录和myid文件：  
\# cd /var/lib  
\# mkdir zookeeper  
\# cd zookeeper  
\# mkdir 1  
\# mkdir 2  
\# mkdir 3  
\# echo "1" > 1/myid  
\# echo "2" > 2/myid  
\# echo "3" > 3/myid  
  
启动ZooKeeper：  
\# cd /usr/local/zookeeper-3.4.10  
\# bin/zkServer.sh start conf/zk1.cfg  
\# bin/zkServer.sh start conf/zk2.cfg  
\# bin/zkServer.sh start conf/zk3.cfg  
  
停止ZooKeeper：  
\# cd /usr/local/zookeeper-3.4.10  
\# bin/zkServer.sh stop conf/zk1.cfg  
\# bin/zkServer.sh stop conf/zk2.cfg  
\# bin/zkServer.sh stop conf/zk3.cfg  
  
重启ZooKeeper：  
\# cd /usr/local/zookeeper-3.4.10  
\# bin/zkServer.sh restart conf/zk1.cfg  
\# bin/zkServer.sh restart conf/zk2.cfg  
\# bin/zkServer.sh restart conf/zk3.cfg  
  
启动结果：  
JMX enabled by default  
Using config: bin/zkServer.sh conf/zk1.cfg  
Starting zookeeper ... STARTED  
  
JMX enabled by default  
Using config: bin/zkServer.sh conf/zk2.cfg  
Starting zookeeper ... STARTED  
  
JMX enabled by default  
Using config: bin/zkServer.sh conf/zk3.cfg  
Starting zookeeper ... STARTED  
  
查看状态：  
用jps命令查看进程：  
\# jps  
24617 QuorumPeerMain （这个就是zookeeper进程）  
  
bin/zkServer.sh status zk1.cfg  
JMX enabled by default  
Using config: bin/zkServer.sh conf/zk1.cfg  
Error contacting service. It is probably not running.  
  
说明有错误，查看日志文件：  
\# cd /usr/local/zookeeper-3.4.10  
\# less zookeeper.out  
java.net.BindException: 地址已在使用  
杀掉当前进程：  
\# kill -9 24617  
  
再重启那三个实例后，查看状态：  
\# bin/zkServer.sh status zk1.cfg  
JMX enabled by default  
Using config: /bin/../conf/zk1.cfg  
Mode: follower  
\# bin/zkServer.sh status zk2.cfg  
JMX enabled by default  
Using config: /bin/../conf/zk2.cfg  
Mode: leader  
\# bin/zkServer.sh status zk3.cfg  
JMX enabled by default  
Using config: /bin/../conf/zk3.cfg  
Mode: follower  
  
Java连接注意细节:  
    KeeperErrorCode = ConnectionLoss for \*    //关闭防火墙  
    KeeperErrorCode = Unimplemented for \*    //curator-framework版本过高  
  
=========================================zkCli基于控制台的使用=============================================  
启动zkCli      
\# ./bin/zkCli.sh  
  
创建节点(临时/临时有序)  
\# create /lock "hello zookeeper"    持久节点  
\# create -s /lock "hello zookeeper"    持久有序  
\# create -e /lock "hello zookeeper"    临时节点  
  
获取节点数据  
\# get /lock  
  
设置节点监听  
\# get /lock 1  
  
设置节点数据  
\# set /lock "lock"  
  
列出某一节点下的子节点信息  
\# ls /lock  
  
列出当前节点的子节点，同时列出节点状态  
\# ls2 /lock  
  
检查状态  
\# stat /lock  
        在zookeeper中，每一次对节点的写操作都认为是一次事务，每一个事务，系统都会分配一个唯一的事务ID。  
    　　czxid：节点被创建的事务ID  
    　　ctime: 创建时间  
    　　mzxid: 最后一次被更新的事务ID  
    　　mtime: 修改时间  
    　　pzxid：子节点列表最后一次被更新的事务ID  
    　　cversion：子节点的版本号  
    　　dataversion：数据版本号  
    　　aclversion：权限版本号  
    　　ephemeralOwner：用于临时节点，代表临时节点的事务ID，如果为持久节点则为0  
    　　dataLength：节点存储的数据的长度  
    　　numChildren：当前节点的子节点个数  
  
移除节点  
\# rmr /lock  
  
移除空的节点  
\# delete /lock  
  
=========================================================ZkClient===========================================================  
1:创建会话  
    public ZkClient(String serverstring)  
    public ZkClient(String zkServers, int connectionTimeout)  
    public ZkClient(String zkServers, int sessionTimeout, int connectionTimeout)  
    public ZkClient(String zkServers, int sessionTimeout, int connectionTimeout, ZkSerializer zkSerializer)  
    public ZkClient(final String zkServers, final int sessionTimeout, final int connectionTimeout, final ZkSerializer zkSerializer, final long operationRetryTimeout)  
    public ZkClient(IZkConnection connection)  
    public ZkClient(IZkConnection connection, int connectionTimeout)  
    public ZkClient(IZkConnection zkConnection, int connectionTimeout, ZkSerializer zkSerializer)  
    public ZkClient(final IZkConnection zkConnection, final int connectionTimeout, final ZkSerializer zkSerializer, final long operationRetryTimeout)  
2:创建节点  
    public void createPersistent(String path)  
    public void createPersistent(String path, boolean createParents)  
    public void createPersistent(String path, boolean createParents, List<ACL> acl)  
    public void createPersistent(String path, Object data)  
    public void createPersistent(String path, Object data, List<ACL> acl)  
    public String createPersistentSequential(String path, Object data)  
    public String createPersistentSequential(String path, Object data, List<ACL> acl)  
    public void createEphemeral(final String path)  
    public void createEphemeral(final String path, final List<ACL> acl)  
    public String create(final String path, Object data, final CreateMode mode)  
    public String create(final String path, Object data, final List<ACL> acl, final CreateMode mode)  
    public void createEphemeral(final String path, final Object data)  
    public void createEphemeral(final String path, final Object data, final List<ACL> acl)  
    public String createEphemeralSequential(final String path, final Object data)  
    public String createEphemeralSequential(final String path, final Object data, final List<ACL> acl)  
3:删除节点  
    public boolean delete(final String path)  
    public boolean delete(final String path, final int version)  
    public boolean deleteRecursive(String path)  
4:获取节点列表  
    public List<String> getChildren(String path)  
    可以注册的Listener为，接口IZkChildListener下面的方法来实现  
        zkClient.subscribeChildChanges(path, new IZkChildListener() \{  
            @Override  
            public void handleChildChange(String parentPath, List<String> currentChilds) throws Exception \{  
            System.out.println("路径" + parentPath +"下面的子节点变更。子节点为：" + currentChilds );  
            \}  
        \});  
5:获取节点内容  
    public <T extends Object> T readData(String path)  
    public <T extends Object> T readData(String path, boolean returnNullIfPathNotExists)  
    public <T extends Object> T readData(String path, Stat stat)  
    对本接口实现监听的接口为IZkDataListener，分别提供了处理数据变化和删除操作的监听：  
        zkClient.subscribeDataChanges(path, new IZkDataListener() \{  
            @Override  
            public void handleDataChange(String path, Object data) throws Exception \{  
            System.out.println("路径" + path +"数据变更为：" + data );  
            \}  
  
            @Override  
            public void handleDataDeleted(String path) throws Exception \{  
            System.out.println("路径" + path +"下面的数据被删除了");  
            \}  
        \});  
6:更新数据  
    public void writeData(String path, Object object)  
    public void writeData(final String path, Object datat, final int expectedVersion)  
    public Stat writeDataReturnStat(final String path, Object datat, final int expectedVersion)  
7:监测节点是否存在  
    protected boolean exists(final String path, final boolean watch);  
8:注册监听  
    IZkChildListener     ZkClient的subscribeChildChanges方法     ZkClient的unsubscribeChildChanges方法  
    IZkDataListener     ZkClient的subscribeDataChanges方法     ZkClient的subscribeChildChanges方法  
    IZkStateListener     ZkClient的subscribeStateChanges方法     ZkClient的unsubscribeStateChanges方法  
  
  
  
=========================================================Curator===========================================================  
1:创建会话  
    public static CuratorFramework newClient(String connectString, RetryPolicy retryPolicy)  
    public static CuratorFramework newClient(String connectString, int sessionTimeoutMs, int connectionTimeoutMs, RetryPolicy retryPolicy)  
  
    RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000,3);  
        CuratorFramework client = CuratorFrameworkFactory.newClient("127.0.0.1:2181",retryPolicy);  
        client.start();  
  
    RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000,3);  
    CuratorFramework client = CuratorFrameworkFactory.newClient("127.0.0.1:2181", 5000,1000,retryPolicy);  
        client.start();  
  
    备注:RetryPolicy实现接口-ExponentialBackoffRetry、BoundedExponentialBackoffRetry、RetryForever、RetryNTimes、RetryOneTime、RetryUntilElapsed  
2:创建节点  
    1>创建一个初始内容为空的节点  
        client.create().forPath(path);  
    2>创建一个包含内容的节点  
        client.create().forPath(path,"我是内容".getBytes());  
    3>创建临时节点，并递归创建父节点  
        client.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL).forPath(path);  
3:删除节点  
    1>删除一个子节点  
        client.delete().forPath(path);  
    2>删除节点并递归删除其子节点  
        client.delete().deletingChildrenIfNeeded().forPath(path);  
    3>指定版本进行删除  
        client.delete().withVersion(1).forPath(path);  
    4>强制保证删除一个节点  
        client.delete().guaranteed().forPath(path);  
4:读取数据  
    1>普通查询  
        client.getData().forPath(path);  
    2>包含状态查询  
        Stat stat = new Stat();  
        client.getData().storingStatIn(stat()).forPath(path);  
5:更新数据  
    1>普通更新  
        client.setData().forPath(path,"新内容".getBytes());  
    2>指定版本更新  
        client.setData().withVersion(1).forPath(path);  
6:异步接口  
    public T inBackground(BackgroundCallback callback, Executor executor);