Zookeeper在实际使用场景很多，比如
配置中心，分布式锁，注册中心等。先简单用原生api搞个zookeeper的配置中心demo,原理是利用watcher 和 znode，客户端监听服务的znode变化
简单流程
image.png
相关代码
package com.huey.demo.configdemo;
/**
* @author huey China.
* @Description : zk 配置相关
* @Date Created in 2018/11/18 上午10:18
*/
public class ZookeeperConfig {
protected String connectString="192.168.59.2:2181,192.168.59.3:2181,192.168.59.4:2181";
}
package com.huey.demo.configdemo;
import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;
/**
* @author huey China.
* @Description : 配置中心服务端
* @Date Created in 2018/11/18 上午10:24
*/
public class ZookeeperConfigServerHandler extends ZookeeperConfig{
private ZooKeeper zookeeper;
public ZookeeperConfigServerHandler() {
try {
this.zookeeper = new ZooKeeper(connectString, 5000,null);
} catch (Exception e) {
e.printStackTrace();
}
}
/***
* 创建持久节点
* @param path
* @param data
* @return
*/
public String createPersistent(String path,String data){
try {
return zookeeper.create(path, data.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
} catch (KeeperException e) {
e.printStackTrace();
} catch (InterruptedException e) {
e.printStackTrace();
}
return null;
}
/***
* 创建临时节点
* @param path
* @param data
* @return
*/
public String createEphemeral(String path,String data){
try {
return zookeeper.create(path, data.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
} catch (KeeperException e) {
e.printStackTrace();
} catch (InterruptedException e) {
e.printStackTrace();
}
return null;
}
/***
* 更新信息
* @param path
* @return
* @throws KeeperException
* @throws InterruptedException
*/
public String getData(String path,boolean watcher) throws KeeperException, InterruptedException {
byte data[] = zookeeper.getData(path,watcher,null);
data = (data == null)? "null".getBytes() : data;
return new String(data);
}
/***
* 更新信息
* @param path
* @param data
* @return
* @throws KeeperException
* @throws InterruptedException
*/
public Stat setData(String path, String data) throws KeeperException, InterruptedException {
return zookeeper.setData(path, data.getBytes(), -1);
}
/***
* 是否存在
* @param path
* @return
* @throws KeeperException
* @throws InterruptedException
*/
public Stat exists(String path, boolean watcher) throws KeeperException, InterruptedException {
return zookeeper.exists(path,watcher);
}
/***
* 删除
* @param path
* @return
* @throws KeeperException
* @throws InterruptedException
*/
public void delete(String path) throws KeeperException, InterruptedException {
zookeeper.delete(path,-1);
}
/***
* 删除
* @param path
* @return
* @throws KeeperException
* @throws InterruptedException
*/
public void deleteRecursive(String path) throws KeeperException, InterruptedException {
ZKUtil.deleteRecursive(zookeeper, path);
}
public void close() throws InterruptedException {
zookeeper.close();
}
}
package com.huey.demo.configdemo;
import org.apache.zookeeper.KeeperException;
import org.junit.Before;
import org.junit.Test;
import java.util.UUID;
/**
* @author huey China.
* @Description : 模拟服务端 znode存储与修改
* @Date Created in 2018/11/18 上午11:00
*/
public class ZookeeperConfigServer {
/**
* 模拟某个节点存储机制
*
*/
private String path = null;
private String pathValue = null;
private ZookeeperConfigServerHandler zookeeperConfigServerHandler;
/**
* @author huey China.
* @Description : 节点生成
* @Date Created in 2018/11/18 上午11:01
*/
@Before
public void front() {
String i = "1";
this.path = new StringBuilder().append("/config_node").toString();
/**
*
*sell_number_version
*/
this.pathValue = new StringBuilder().append("/sell_").append(i).append("_").append(2).toString();
System.out.println(this.path+ " -----> " + this.pathValue);
}
/**
* @author huey China.
* @Description : 启动并创建临时节点
* @Date Created in 2018/11/18 上午11:31 异常为事件未watch可忽略
*/
@Test
public void testCreate() throws InterruptedException {
//启动服务端
zookeeperConfigServerHandler = new ZookeeperConfigServerHandler();
zookeeperConfigServerHandler.createEphemeral(this.path,this.pathValue);
Thread.sleep(Integer.MAX_VALUE);
}
/**
* @author huey China.
* @Description : 模拟服务端修改节点数据
* @Date Created in 2018/11/18 上午11:32
*/
@Test
public void testUpdate() throws InterruptedException, KeeperException {
ZookeeperConfigServerHandler zookeeperConfigServerHandler = new ZookeeperConfigServerHandler();
zookeeperConfigServerHandler.setData(this.path, new StringBuilder().append("/sell_").append(UUID.randomUUID()).append("_").append(System.currentTimeMillis()).toString());
}
/**
* @author huey China.
* @Description : 服务端获取该临时即诶单数据
* @Date Created in 2018/11/18 上午11:33
*/
@Test
public void testGetData() throws KeeperException, InterruptedException {
System.out.println("zkconfig new value is " + zookeeperConfigServerHandler.getData(this.path,true));
}
}
package com.huey.demo.configdemo;
/**
* @author huey China.
* @Description : zk watcher
* @Date Created in 2018/11/18 下午3:48
*/
import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;
import java.util.concurrent.ConcurrentHashMap;
public class ZookeeperConfigClientHandler extends ZookeeperConfig implements Watcher{
private ZooKeeper zookeeper;
/**
*
*该节点jvm内存存储
*/
private ConcurrentHashMap<String, String> nodes = new ConcurrentHashMap<String, String>();
public ConcurrentHashMap<String, String> getNodes() {
return nodes;
}
public ZookeeperConfigClientHandler() {
try {
this.zookeeper = new ZooKeeper(connectString, 5000,this);
} catch (Exception e) {
e.printStackTrace();
}
}
public ZookeeperConfigClientHandler(String path) {
try {
this.zookeeper = new ZooKeeper(connectString, 5000,this);
} catch (Exception e) {
e.printStackTrace();
}
}
/***
* 创建临时节点
* @param path
* @param data
* @return
*/
public String createEphemeral(String path,String data){
try {
return zookeeper.create(path, data.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
} catch (KeeperException e) {
e.printStackTrace();
} catch (InterruptedException e) {
e.printStackTrace();
}
return null;
}
/***
* 获取信息
* @param path
* @return
* @throws KeeperException
* @throws InterruptedException
*/
public String getData(String path,boolean watcher) throws KeeperException, InterruptedException {
byte data[] = zookeeper.getData(path,watcher,null);
data = (data == null)? "null".getBytes() : data;
return new String(data);
}
/***
* 更新信息
* @param path
* @param data
* @return
* @throws KeeperException
* @throws InterruptedException
*/
public Stat setData(String path, String data) throws KeeperException, InterruptedException {
return zookeeper.setData(path, data.getBytes(), -1);
}
/***
* 是否存在
* @param path
* @return
* @throws KeeperException
* @throws InterruptedException
*/
public Stat exists(String path, boolean watcher) throws KeeperException, InterruptedException {
return zookeeper.exists(path,watcher);
}
/***
* 删除
* @param path
* @return
* @throws KeeperException
* @throws InterruptedException
*/
public void delete(String path) throws KeeperException, InterruptedException {
zookeeper.delete(path,-1);
}
/***
* 删除
* @param path
* @return
* @throws KeeperException
* @throws InterruptedException
*/
public void deleteRecursive(String path) throws KeeperException, InterruptedException {
ZKUtil.deleteRecursive(zookeeper, path);
}
public void close() throws InterruptedException {
zookeeper.close();
}
/**
* @author huey China.
* @Description : TODO 节点未动态化
* @Date Created in 2018/11/18 上午11:44
*/
public void process(WatchedEvent event) {
// 连接状态
Event.KeeperState keeperState = event.getState();
// 事件类型
Event.EventType eventType = event.getType();
// 受影响的path
String pathNode = event.getPath();
System.out.println("连接状态:"+keeperState+",事件类型："+eventType+",受影响的path:" + pathNode);
try {
String path = "/config_node";
String data = this.getData(path, true);
if(null!=this.exists(path,true) ) {
nodes.put(path,data);
System.out.println("zk内容:"+ data + ",内存内容:"+ nodes.get(path));
}
if (this.isNodeDataChanged(eventType, true)) {
System.out.println("更改后zk内容:"+ this.getData(path, true) + ",内存内容:"+ nodes.get(path));
}
} catch (Exception e) {
e.printStackTrace();
}
System.out.println("--------------------");
}
/**
* @author huey China.
* @Description : 是否需要节点数据改变事件
* @Date Created in 2018/11/18 上午11:46
*/
private boolean isNodeDataChanged(Watcher.Event.EventType eventType,boolean flag){
if (Event.EventType.NodeDataChanged == eventType && flag == true){
return true;
}
return false;
}
}
package com.huey.demo.configdemo;
import org.apache.zookeeper.KeeperException;
import org.junit.Test;
/**
* @author huey China.
* @Description : 模拟服务端 znode存储与修改
* @Date Created in 2018/11/18 上午11:00
*/
public class ZookeeperConfigClient {
/**
* 模拟某个节点存储机制
*/
private String path = "/config_node";
private String pathValue = null;
private ZookeeperConfigClientHandler zookeeperConfigClientHandler = null;
/**
* @author huey China.
* @Description : 启动客户端，监听服务端
* @Date Created in 2018/11/18 上午11:34
*/
@Test
public void testStart() throws InterruptedException {
zookeeperConfigClientHandler = new ZookeeperConfigClientHandler();
Thread.sleep(Integer.MAX_VALUE);
}
/**
* @author huey China.
* @Description : 客户端获取数据
* @Date Created in 2018/11/18 上午11:34
*/
@Test
public void testGetData() throws KeeperException, InterruptedException {
System.out.println("zkconfig new value is " + zookeeperConfigClientHandler.getData(this.path, true));
System.out.println("内存 value is " + zookeeperConfigClientHandler.getNodes().get(this.path));
}
}
核心代码
image.png
服务端与客户端启动前
image.png
启动后
image.png
服务端运行testUpdate后，客户端收到监听，并更新最新配置数据（其他UT类似）
image.png
总结
利用了zk的监听原理，底层是netty（juc+nio+序列化），使用上还是比较简单的，其他场景待续！
.参考
官网：http://zookeeper.apache.org
书籍：从Paxos到Zookeeper
网课： 推荐 慕课网 图灵学院 谷粒学院
