package com.zookeeper.watcher;

import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicInteger; import org.apache.log4j.Logger;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.EventType;
import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.ACL;
import org.apache.zookeeper.data.Stat; public class ZKClient implements Watcher{
private static final Logger logger = Logger.getLogger(ZKClient.class); //定义session失效时间
private static final int SESSION_TIMEOUT = 10000;
//zookeeper服务器地址
private static final String ZOOKEEPER_ADDRESS = "192.168.56.1:2181"; //ZooKeeper变量
private ZooKeeper zk = null;
//定义原子变量
AtomicInteger seq = new AtomicInteger();
//信号量设置,用于等待zookeeper连接建立之后,通知阻塞程序继续向下执行
private CountDownLatch connectedSemaphore = new CountDownLatch(1); public static void main(String[] args) throws InterruptedException {
String parentPath= "/test"; //父节点
String childrenPath = "/test/children"; //子节点 ZKClient test = new ZKClient();
//创建链接
test.createConnection(); boolean isSuccess = test.createNode(parentPath, "abc", Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
if (isSuccess) {
//读取数据
String result = test.getNodeData(parentPath, true);
logger.info("更新前数据:" + result); //更新数据
isSuccess = test.updateNode(parentPath, String.valueOf(System.currentTimeMillis()));
if(isSuccess){
logger.info("更新后数据:" + test.getNodeData(parentPath, true));
} // 创建子节点
isSuccess = test.createNode(childrenPath, String.valueOf(System.currentTimeMillis()), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
if(isSuccess){
test.updateNode(childrenPath, String.valueOf(System.currentTimeMillis()));
} //读取子节点
List<String> childrenList = test.getChildren(parentPath, true);
if(childrenList!=null && !childrenList.isEmpty()){
for(String children : childrenList){
System.out.println("子节点:" + children);
}
}
} Thread.sleep(1000);
//创建临时有序子节点
test.createNode(childrenPath, String.valueOf(System.currentTimeMillis()), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT_SEQUENTIAL);
test.createNode(childrenPath, String.valueOf(System.currentTimeMillis()), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT_SEQUENTIAL);
test.createNode(childrenPath, String.valueOf(System.currentTimeMillis()), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT_SEQUENTIAL); // 读取子节点,并删除
List<String> childrenList = test.getChildren(parentPath, true);
if (childrenList != null && !childrenList.isEmpty()) {
for (String children : childrenList) {
System.out.println("子节点:" + children);
test.deleteNode(parentPath + "/" + children);
}
} //删除父节点
if (test.exists(childrenPath, false) != null) {
test.deleteNode(childrenPath);
} //释放链接
Thread.sleep(1000);
test.releaseConnection();
} /**
* 创建节点
*
* @param path 节点路径
* @param data 数据内容
* @param acl 访问控制列表
* @param createMode znode创建类型
* @return
*/
public boolean createNode(String path, String data, List<ACL> acl, CreateMode createMode) {
try {
//设置监控(由于zookeeper的监控都是一次性的,所以每次必须设置监控)
exists(path, true);
String resultPath = this.zk.create(path, data.getBytes(), acl, createMode);
logger.info(String.format("节点创建成功,path: %s,data: %s", resultPath, data));
} catch (Exception e) {
logger.error("节点创建失败", e);
return false;
} return true;
} /**
* 更新指定节点数据内容
*
* @param path 节点路径
* @param data 数据内容
* @return
*/
public boolean updateNode(String path, String data) {
try {
Stat stat = this.zk.setData(path, data.getBytes(), -1);
logger.info("更新节点数据成功,path:" + path + ", stat: " + stat);
} catch (Exception e) {
logger.error("更新节点数据失败", e);
return false;
} return true;
} /**
* 删除指定节点
*
* @param path
* 节点path
*/
public void deleteNode(String path) {
try {
this.zk.delete(path, -1);
logger.info("删除节点成功,path:" + path);
} catch (Exception e) {
logger.error("删除节点失败", e);
}
} /**
* 读取节点数据
*
* @param path 节点路径
* @param needWatch 是否监控这个目录节点,这里的 watcher是在创建ZooKeeper实例时指定的watcher
* @return
*/
public String getNodeData(String path, boolean needWatch) {
try {
Stat stat = exists(path, needWatch);
if(stat != null){
return new String(this.zk.getData(path, needWatch, stat));
}
} catch (Exception e) {
logger.error("读取节点数据内容失败", e);
} return null;
} /**
* 获取子节点
*
* @param path 节点路径
* @param needWatch 是否监控这个目录节点,这里的 watcher是在创建ZooKeeper实例时指定的watcher
* @return
*/
public List<String> getChildren(String path, boolean needWatch) {
try {
return this.zk.getChildren(path, needWatch);
} catch (Exception e) {
logger.error("获取子节点失败", e);
return null;
}
} /**
* 判断znode节点是否存在
*
* @param path 节点路径
* @param needWatch 是否监控这个目录节点,这里的 watcher是在创建ZooKeeper实例时指定的watcher
* @return
*/
public Stat exists(String path, boolean needWatch) {
try {
return this.zk.exists(path, needWatch);
} catch (Exception e) {
logger.error("判断znode节点是否存在发生异常", e);
} return null;
} /**
* 创建ZK连接
*
* @param connectAddr
* @param sessionTimeout
*/
public void createConnection() {
this.releaseConnection(); try {
zk = new ZooKeeper(ZOOKEEPER_ADDRESS, SESSION_TIMEOUT, this);
logger.info("开始连接ZK服务器..."); //zk连接未创建成功进行阻塞
connectedSemaphore.await();
} catch (Exception e) {
logger.error("ZK连接创建失败", e);
}
} /**
* 关闭ZK连接
*/
public void releaseConnection() {
if (this.zk != null) {
try {
this.zk.close();
logger.info("ZK连接关闭成功");
} catch (InterruptedException e) {
logger.error("ZK连接关闭失败", e);
}
}
} @Override
public void process(WatchedEvent event) {
logger.info("进入process()方法...event = " + event); if (event == null) {
return;
} KeeperState keeperState = event.getState(); // 连接状态
EventType eventType = event.getType(); // 事件类型
String path = event.getPath(); // 受影响的path String logPrefix = "【Watcher-" + this.seq.incrementAndGet() + "】";
logger.info(String.format("%s收到Watcher通知...", logPrefix));
logger.info(String.format("%s连接状态:%s", logPrefix, keeperState));
logger.info(String.format("%s事件类型:%s", logPrefix, eventType));
logger.info(String.format("%s受影响的path:%s", logPrefix, path)); if (KeeperState.SyncConnected == keeperState) {
if (EventType.None == eventType) {
// 成功连接上ZK服务器
logger.info(logPrefix + "成功连接上ZK服务器...");
connectedSemaphore.countDown(); } else if (EventType.NodeCreated == eventType) {
// 创建节点
logger.info(logPrefix + "节点创建");
this.exists(path, true); } else if (EventType.NodeDataChanged == eventType) {
// 更新节点
logger.info(logPrefix + "节点数据更新");
logger.info(logPrefix + "数据内容: " + this.getNodeData(path, true)); } else if (EventType.NodeChildrenChanged == eventType) {
// 更新子节点
logger.info(logPrefix + "子节点变更");
logger.info(logPrefix + "子节点列表:" + this.getChildren(path, true)); } else if (EventType.NodeDeleted == eventType) {
// 删除节点
logger.info(logPrefix + "节点 " + path + " 被删除");
} } else if (KeeperState.Disconnected == keeperState) {
logger.info(logPrefix + "与ZK服务器断开连接");
} else if (KeeperState.AuthFailed == keeperState) {
logger.info(logPrefix + "权限检查失败");
} else if (KeeperState.Expired == keeperState) {
logger.info(logPrefix + "会话失效");
} } }

Zookeeper API for JAVA实战与应用的更多相关文章

  1. zookeeper原生API做java客户端

    简介 本文是使用apache提供的原生api做zookeeper客户端 jar包 zookeeper-3.4.5.jar   Demo package bjsxt.zookeeper.base; im ...

  2. 「小程序JAVA实战」springboot的后台搭建(31)

    转自:https://idig8.com/2018/08/29/xiaochengxujavashizhanspringbootdehoutaidajian31/ 根据下面的图,我们来建立下对应的sp ...

  3. 《Netty Zookeeper Redis 高并发实战》 图书简介

    <Netty Zookeeper Redis 高并发实战> 图书简介 本书为 高并发社群 -- 疯狂创客圈 倾力编著, 高度剖析底层原理,深度解读面试难题 疯狂创客圈 Java 高并发[ ...

  4. ZooKeeper系列4:ZooKeeper API简介及编程

    问题导读: 1.ZooKeeper API 共包含几个包? 2.如何使用ZooKeeper API 创建zookeeper应用程序? 1)ZooKeeper API 简介   ZooKeeper AP ...

  5. 面向服务的体系架构 SOA(三) --- Zookeeper API、zkClient API的使用

    zookeeper简单介绍及API使用 1.1 zookeeper简介 zookeeper是一个针对大型分布式系统的可靠的协调系统,提供的功能包括配置维护.名字服务.分布式同步.组服务等.zookee ...

  6. Zookeeper+Kafka完全分布式实战部署

    Zookeeper+Kafka完全分布式实战部署 作者:尹正杰 版权声明:原创作品,谢绝转载!否则将追究法律责任. 其实我之前部署过kafak和zookeeper的完全分布式,集群是可以正常使用没错, ...

  7. Hbase记录-ZooKeeper API

    Zookeeper API ZooKeeper有一个Java和C绑定的官方API.ZooKeeper社区提供了对于大多数语言(.NET,Python等)的非官方API.使用ZooKeeper的API, ...

  8. Restful API设计规范及实战【说的比较清楚了】

    Restful API设计规范及实战   Restful API的概念在此就不费口舌了,博友们网上查哈定义文章很多,直入正题吧: 首先抛出一个问题:判断id为 用户下,名称为 使命召唤14(COD14 ...

  9. Zookeeper 系列(三)Zookeeper API

    Zookeeper 系列(三)Zookeeper API 本节首先介绍 Zookeeper 的 Shell 命令,再对 Java 操作 Zookeeper 的三种方式进行讲解,本节先介绍 Zookee ...

随机推荐

  1. KBEngine 学习笔记

    最近开始学习 KBE 扩展技能点>_<!所以建一个随笔记录一下遇到的小问题: 问题1 :DBMgr找不到LibMysql32.dll 解决:VS 中KBE源码 默认的是Win32 ,Win ...

  2. MySQL服务 - MySQL程序的配置文件、参数、变量查看

    查看配置文件及读取顺序 MySQL的配置文件以.cnf结尾,可能会有多个,而不同版本的MySQL程序的读取配置文件的路径也都不同,要想获取MySQL读取配置文件的顺序可以通过以下指令查看: shell ...

  3. Ubuntu进阶学习,指令迅速查询,Bug迅速查询(Ctrl+F)

    There is some notes while I am learning Ubuntu Operate System! (Ask Ubuntu) 1-- Hard link : ln comma ...

  4. [Locked] Unique Word Abbreviation

    Unique Word Abbreviation An abbreviation of a word follows the form <first letter><number&g ...

  5. Mac中如何写NTFS的移动硬盘

    Mac中如何写NTFS的移动硬盘 按 照网上的说法,简单的就是购买Paragon NTFS或者Tuxera NTFS之类的收费软件,不想花钱的就用NTFS-3G.作为一个开发人员,既然能够自己解决,当 ...

  6. Start to write blogs 【开始写博客】

    I used to be lazy and wrote no blogs. I used to live at leisure and wasted opportunity. Time flies, ...

  7. maven 在clean package时,出现:找不到符号 [ERROR] 符号: 方法 sqlDdlFilter(java.lang.String) 解决办法

    另一个项目中增加了,sqlDdlFilter 在调用的项目中clean package时,出现 找不到符号[ERROR] 符号: 方法 sqlDdlFilter(java.lang.String) 原 ...

  8. H5 边框:带border的百分比布局

    响应式Web设计经常需要我们通过百分比设置组件宽度.如果我们不考虑边框,那么很容易就可以实现,但如果你给每一列以及总宽度都采用百分比设置,那这个时候固定的边框大小就会出来捣乱.下面我们将看到一组方法去 ...

  9. 1 什么是virtual Machine

    1.所有的虚拟机以文件的形式存放在存储上. 2.虚拟机的文件构成: swap files: <vm_name>.vswp  虚拟机的内存文件,vmx-<vm_name>.vsw ...

  10. 短址服务 api

    1  is.gd 他这个api简单: http://is.gd/api.php?longurl= 后面加网址就可以返回短址 2 Google URL Shortener API api地址: http ...