一、搭建ZooKeeper环境
1、参考Linux下ZooKeeper集群安装
二、添加zooKeeper驱动,以gradle添加为例
compile group: 'org.apache.zookeeper', name: 'zookeeper', version: '3.4.5'
compile group: 'junit', name: 'junit', version: '4.12'
三、测试,新建SimpleZkClient类,测试增删改查
![](https://images.cnblogs.com/OutliningIndicators/ContractedBlock.gif)
package com.moy.zookeeper; import org.apache.zookeeper.CreateMode; import org.apache.zookeeper.KeeperException; import org.apache.zookeeper.ZooDefs; import org.apache.zookeeper.ZooKeeper; import org.apache.zookeeper.data.Stat; import org.junit.Before; import org.junit.Test; import java.io.IOException; import java.util.List; import java.util.Objects; /** * [Project]:moy-gradle-project <br/> * [Email]:moy25@foxmail.com <br/> * [Date]:2018/4/3 <br/> * [Description]: <br/> * * @author YeXiangYang */ public class SimpleZkClient { private static String connectString = "node3:2181,node4:2181,node5:2181"; private static int sessionTimeout = 2000; private ZooKeeper zkClient; private static String testPath = "/hello world"; @Before public void before() throws IOException { zkClient = new ZooKeeper(connectString, sessionTimeout, (event) -> { System.out.printf("%s ---> %s ", event.getType(), event.getPath()); try { zkClient.getChildren("/", true); } catch (Exception e) { throw new RuntimeException(e); } }); } @Test public void addNode() throws KeeperException, InterruptedException { String node = zkClient.create(testPath, "hello zk".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); System.out.printf(node); } @Test public void getAllNode() throws KeeperException, InterruptedException { List<String> childrenNode = zkClient.getChildren("/", true); for (String nodePath : childrenNode) { System.out.println(nodePath); } } @Test public void getData() throws KeeperException, InterruptedException { byte[] data = zkClient.getData(testPath, true, null); System.out.printf("%s ---> %s ", testPath, new String(data)); } @Test public void setData() throws KeeperException, InterruptedException { if (Objects.nonNull(zkClient.exists(testPath, false))) { Stat stat = zkClient.setData(testPath, "hello world".getBytes(), -1); System.out.println(stat); } } @Test public void deleteNode() throws KeeperException, InterruptedException { if (Objects.nonNull(zkClient.exists(testPath, false))) { zkClient.delete(testPath, -1); } } }
四、测试zooKeeper注册监听
1、新建ZkHelper帮助类
![](https://images.cnblogs.com/OutliningIndicators/ContractedBlock.gif)
package com.moy.zookeeper; import org.apache.zookeeper.ZooKeeper; import java.io.IOException; import java.util.Objects; /** * [Project]:moy-gradle-project <br/> * [Email]:moy25@foxmail.com <br/> * [Date]:2018/4/5 <br/> * [Description]: <br/> * * @author YeXiangYang */ public abstract class ZkHelper { private static String connectString = "node3:2181,node4:2181,node5:2181"; private static int sessionTimeout = 2000; private static ThreadLocal<ZooKeeper> zkClient = new ThreadLocal<>(); public static ZooKeeper getZkClient() { ZooKeeper zooKeeper = zkClient.get(); if (Objects.isNull(zooKeeper)) { zooKeeper = createZkConnect(); zkClient.set(zooKeeper); } return zooKeeper; } public static void removeZkClient() { if (Objects.nonNull(zkClient.get())) { zkClient.remove(); } } private static ZooKeeper createZkConnect() { try { return new ZooKeeper(connectString, sessionTimeout, (event) -> { System.out.printf("zk发生了变化事件: %s ---> %s ", event.getPath(), event.getType()); }); } catch (IOException e) { throw new RuntimeException(e); } } }
2、新建服务端测试注册类DistributedServer
![](https://images.cnblogs.com/OutliningIndicators/ContractedBlock.gif)
package com.moy.zookeeper.demo; import com.moy.zookeeper.ZkHelper; import org.apache.zookeeper.CreateMode; import org.apache.zookeeper.KeeperException; import org.apache.zookeeper.ZooDefs; import org.apache.zookeeper.ZooKeeper; import org.apache.zookeeper.data.Stat; import java.util.Objects; import java.util.Random; /** * [Project]:moy-gradle-project <br/> * [Email]:moy25@foxmail.com <br/> * [Date]:2018/4/5 <br/> * [Description]: <br/> * * @author YeXiangYang */ public class DistributedServer { public static final String PARENT_NODE = "/servers"; public static void main(String[] args) throws Exception { // 获取zk连接 ZooKeeper zkClient = ZkHelper.getZkClient(); // 注册服务信息到zk上 registerServer(zkClient); // 执行业务 businessHandler(); } public static void businessHandler() throws InterruptedException { System.out.printf("开始执行业务... "); Thread.sleep(Long.MAX_VALUE); } private static void registerServer(ZooKeeper zkClient) { try { checkParentNode(zkClient); zkClient.create(PARENT_NODE + "/server", createServerName().getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL); } catch (Exception e) { throw new RuntimeException(e); } } private static String createServerName() { return "node-" + new Random().nextInt(10); } private static void checkParentNode(ZooKeeper zkClient) throws KeeperException, InterruptedException { Stat stat = zkClient.exists(PARENT_NODE, true); if (Objects.isNull(stat)) { zkClient.create(PARENT_NODE, PARENT_NODE.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); } } }
3、新建客户端测试监听类DistributedClient
![](https://images.cnblogs.com/OutliningIndicators/ContractedBlock.gif)
package com.moy.zookeeper.demo; import com.moy.zookeeper.ZkHelper; import org.apache.zookeeper.KeeperException; import org.apache.zookeeper.ZooKeeper; import java.util.List; import java.util.Objects; /** * [Project]:moy-gradle-project <br/> * [Email]:moy25@foxmail.com <br/> * [Date]:2018/4/5 <br/> * [Description]: <br/> * * @author YeXiangYang */ public class DistributedClient { public static void main(String[] args) throws Exception { // 获取zk连接 ZooKeeper zkClient = ZkHelper.getZkClient(); // 获取zk服务节点信息 listServerInfo(zkClient); // 处理业务 DistributedServer.businessHandler(); } private static void listServerInfo(ZooKeeper zkClient) { try { List<String> serverList = zkClient.getChildren(DistributedServer.PARENT_NODE, event -> { listServerInfo(zkClient); }); printServerInfo(zkClient, serverList); } catch (Exception e) { e.printStackTrace(); } } private static void printServerInfo(ZooKeeper zkClient, List<String> serverList) throws KeeperException, InterruptedException { System.out.println(System.currentTimeMillis() + "节点信息为:"); if (Objects.isNull(serverList) || serverList.size() <= 0) { System.out.println(" []"); return; } for (String server : serverList) { String nodePath = DistributedServer.PARENT_NODE + "/" + server; if (Objects.isNull(zkClient.exists(nodePath, false))) { continue; } byte[] data = zkClient.getData(nodePath, false, null); String serverData = new String(data); System.out.printf(" [server-%s 拥有的信息为: %s] ", server, serverData); } } }
五、运行DistributedServer,注册服务信息,在运行DistributedClient,监听服务信息。
1、需要将zookeeper配置信息改为本身的
yexiangyang
moyyexy@gmail.com