分布式学习之zookeeper3

　　分布式学习之zookeeper-3
　　分布式很多地方都会用到zk，虽然这个技术出了很久，但是作为应用开发工程师可能这方面接触的还是比较少。
　　我打算从浅入深的学习下zk的使用。 Java操作zk的几种方式
　　java操作zk可以使用3种方式 使用原生 zkClinet curator
　　开始写代码前请务必启动启动 zookeeper(虽然是废话，但是还要提醒一下)。
　　另外最好开一个客户端用来观察调用代码后的信息。 原生
　　maven引用      org.apache.zookeeper     zookeeper     3.4.8 
　　代码： /**  * @Author: jimmy  * @Date: 2022/1/2 21:32  * @Description:  */ public class ZookeeperOri {      public static void main(String[] args) throws IOException, KeeperException, InterruptedException {         // 连接zk服务器         ZooKeeper zk = new ZooKeeper(＂127.0.0.1＂, 2000, new Watcher() {             @Override             public void process(WatchedEvent watchedEvent) {                 System.out.println(＂--zookeeper连接---＂);             }         });          // ---- 节点基本操作 ----         // 获取子节点         List children = zk.getChildren(＂/＂, null);         children.forEach(e -> System.out.println(e));          // 判断是否存在节点、删除节点 注意删除节点的时候必须不存在子节点         if (null != zk.exists(＂/my1＂, null)) {             zk.delete(＂/my1＂, -1);         }         // 原生客户端创建节点         // CreateMode PERSISTENT 持久  PERSISTENT_SEQUENTIAL 有序持久         //            EPHEMERAL 临时   EPHEMERAL_SEQUENTIAL  有序临时         // 这里需要设置权限  不然会报错 -> 【KeeperErrorCode = MarshallingError】         String c_rs = zk.create(＂/my1＂,＂test-my＂.getBytes(),                 ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);         System.out.println(＂----＂ + c_rs + ＂-----＂);          byte[] rs = zk.getData(＂/my1＂, null, new Stat());         System.out.println(＂----＂ + new String(rs) + ＂-----＂);          // 非顺序节点不能重复创建         try {             String c_rs_tmp = zk.create(＂/my1＂,＂test-my＂.getBytes(),                     ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);         } catch (Exception e) {             System.out.println(＂ex :＂ + e.getMessage());         }          // 设置值         Stat stat = zk.setData(＂/my1＂, ＂my-test＂.getBytes(), -1);         System.out.println(JSONObject.toJSONString(stat));          // 获取值         rs = zk.getData(＂/my1＂, null, new Stat());         System.out.println(＂----＂ + new String(rs) + ＂-----＂);          // 创建有序节点         UUID uuid = new UUID(12,1);         c_rs = zk.create(＂/my/sequen＂,(＂test-mysequen＂ + uuid.toString()).getBytes(),                 ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT_SEQUENTIAL);         System.out.println(＂----＂ + c_rs + ＂-----＂);          List children_s = zk.getChildren(＂/my＂, null);         children_s.forEach(e -> System.out.println(e));          // 有序节点的获取，不能直接使用节点名称，这里会报错         try{             rs = zk.getData(＂/my/sequen＂, null, new Stat());             System.out.println(＂----＂ + new String(rs) + ＂-----＂);         } catch (Exception e) {             System.out.println(＂ex :＂ + e.getMessage());         }          System.out.println(＂有序节点的获取-----＂);         for (String nodeKey : children_s) {             byte[] tmp = zk.getData(＂/my/＂ + nodeKey, null, new Stat());             System.out.println(＂nodeKey:＂ + nodeKey + ＂, value:＂ + new String(tmp));         }          // ---- 监听 ----         String epNode = zk.create(＂/myEp＂,＂test-myEp＂.getBytes(),                 ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);          zk.getData(＂/myEp＂, new WatcherDefine(zk), new Stat());         // 触发watch         zk.setData(＂/myEp＂, ＂uu＂.getBytes(), -1);         zk.close();     }       static class WatcherDefine implements Watcher{            private  ZooKeeper  zookeeper;          public WatcherDefine(ZooKeeper zookeeper) {             this.zookeeper = zookeeper;         }          @Override         public void process(WatchedEvent watchedEvent) {             Stat stat = new Stat();             //如果当前的连接状态是连接成功的，那么通过计数器去控制             if(watchedEvent.getState()== Watcher.Event.KeeperState.SyncConnected){                 if(Watcher.Event.EventType.None==watchedEvent.getType()&&null==watchedEvent.getPath()){                     System.out.println(watchedEvent.getState()+＂-->＂+watchedEvent.getType());                 }else if(watchedEvent.getType()== Watcher.Event.EventType.NodeDataChanged){                     try {                         System.out.println(＂数据变更触发路径：＂+watchedEvent.getPath()+＂->改变后的值：＂+                                 new String(zookeeper.getData(watchedEvent.getPath(),true,stat)));                     } catch (KeeperException e) {                         e.printStackTrace();                     } catch (InterruptedException e) {                         e.printStackTrace();                     }                 }else if(watchedEvent.getType()== Watcher.Event.EventType.NodeChildrenChanged){                     //子节点的数据变化会触发                     try {                         System.out.println(＂子节点数据变更路径：＂+watchedEvent.getPath()+＂->节点的值：＂+                                 zookeeper.getData(watchedEvent.getPath(),true,stat));                     } catch (KeeperException e) {                         e.printStackTrace();                     } catch (InterruptedException e) {                         e.printStackTrace();                     }                 }else if(watchedEvent.getType()== Watcher.Event.EventType.NodeCreated){                     //创建子节点的时候会触发                     try {                         System.out.println(＂节点创建路径：＂+watchedEvent.getPath()+＂->节点的值：＂+                                 zookeeper.getData(watchedEvent.getPath(),true, stat));                     } catch (KeeperException e) {                         e.printStackTrace();                     } catch (InterruptedException e) {                         e.printStackTrace();                     }                 }else if(watchedEvent.getType()== Watcher.Event.EventType.NodeDeleted){                     //子节点删除会触发                     System.out.println(＂节点删除路径：＂+watchedEvent.getPath());                 }             }         }     } }zkClinet
　　maven依赖       com.101tec     zkclient     0.10 
　　我在下面测试的时候遇到了乱码问题分为2部分（我是windows测试环境） 无论是否是中文写入的时候都是乱码实现 ZkSerializer来解决。 上面问题解决后中文仍然是乱码windows的 cmd/powershell 的锅，需要设置为UTF-8临时设置：CHCP 65001这个仍然会存在乱码的问题，因此我在练习类种后面把数据打出来了。
　　练习类： /**  * @Author: jimmy  * @Date: 2022/1/2 12:00  * @Description:  */ public class ZkClientPp {      public static void main(String[] args) throws InterruptedException, UnsupportedEncodingException {         ZkClient zkClient = new ZkClient(＂127.0.0.1:2181＂);         // 解决乱码问题         zkClient.setZkSerializer(new ZkCodeSerializer());          if(zkClient.exists(＂/prc＂)){             zkClient.delete(＂/prc＂);         }         // 创建持久化节点         zkClient.createPersistent(＂/prc＂);         zkClient.writeData(＂/prc＂,＂handsome-boy＂);          // 读取值         String data = zkClient.readData(＂/prc＂);         System.out.println(data);          // --------- 获取子节点 ----------         List children = zkClient.getChildren(＂/＂);         children.forEach(c -> System.out.println(c + ＂＂));          System.out.println(＂……………· ·…οΟ  の  Οο…· ·…………… ＂);          // 监听         zkClient.subscribeDataChanges(＂/prc＂, new IZkDataListener() {             @Override             public void handleDataChange(String s, Object o) throws Exception {                 System.out.println(＂……………· ·…οΟ  の  Οο…· ·…………… ＂);                 System.out.println(＂节点名称：＂ + s + ＂->节点修改后的值＂+o);             }              @Override             public void handleDataDeleted(String s) throws Exception {                 System.out.println(＂……………· ·…οΟ  の  Οο…· ·…………… ＂);                 System.out.println(＂节点名称：＂ + s + ＂-> 删除了＂);             }         });          Thread.sleep(2000);          //new String(＂老子叫工藤新一＂.getBytes(＂utf-8＂)         zkClient.writeData(＂/prc＂, ＂老子叫工藤新一！＂);         String s = zkClient.readData(＂/prc＂);         System.out.println(＂第一次变更……………· ·…οΟ  の  Οο…· ·…………… prc＂);         System.out.println(s);          Thread.sleep(5000);          zkClient.writeData(＂/prc＂, ＂变成了一个小屁孩＂);         data = zkClient.readData(＂/prc＂);         System.out.println(＂第二次变更……………· ·…οΟ  の  Οο…· ·…………… prc＂);         System.out.println(data);          // 修改完后，若程序立即结束，则无法看到watch的信息。         Thread.sleep(Integer.MAX_VALUE);          zkClient.close();     }      // 乱码解决类     static class ZkCodeSerializer implements ZkSerializer {         @Override         public byte[] serialize(Object object) throws ZkMarshallingError {             return ((String)object).getBytes(Charset.forName(＂UTF-8＂));         }          @Override         public Object deserialize(byte[] bytes) throws ZkMarshallingError {             return new String(bytes, Charset.forName(＂UTF-8＂));         }     } }curator
　　maven依赖      org.apache.curator     curator-framework     2.11.0 
　　代码： /**  * @Author: jimmy  * @Date: 2022/1/3 15:25  * @Description:  */ public class CuratorClinet {      public static void main(String[] args) throws InterruptedException {         CuratorFramework curatorFramework =                 CuratorFrameworkFactory.newClient(＂127.0.0.1:2181＂,5000,2000,                         new ExponentialBackoffRetry(1000,3));          curatorFramework.start();          // 创建节点         try {             String result=curatorFramework.create().                     creatingParentsIfNeeded().                     withMode(CreateMode.PERSISTENT).                     forPath(＂/curator/C1/C101＂,＂curator＂.getBytes());              System.out.println(result);         } catch (Exception e) {             e.printStackTrace();         }          // 删除节点         try {             // 默认情况下，version为-1 递归删除             curatorFramework.delete().deletingChildrenIfNeeded().forPath(＂/curator＂);         } catch (Exception e) {             e.printStackTrace();         }          // 查询         Stat stat=new Stat();         try {             byte[] bytes=curatorFramework.getData().storingStatIn(stat).forPath(＂/my＂);             System.out.println(new String(bytes));         } catch (Exception e) {             e.printStackTrace();         }          // 更新         try {             Stat s = curatorFramework.setData().forPath(＂/my＂,＂123＂.getBytes());             System.out.println(s);         } catch (Exception e) {             e.printStackTrace();         }          // 异步操作         ExecutorService service= Executors.newFixedThreadPool(1);         CountDownLatch countDownLatch=new CountDownLatch(1);         try {             curatorFramework.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL).                     inBackground(new BackgroundCallback() {                         @Override                         public void processResult(CuratorFramework curatorFramework, CuratorEvent curatorEvent) throws Exception {                             System.out.println(Thread.currentThread().getName() + ＂->resultCode:＂                                     +curatorEvent.getResultCode()+＂->＂                                     +curatorEvent.getType());                             countDownLatch.countDown();                         }                     },service).forPath(＂/prc＂,＂be happy＂.getBytes());         } catch (Exception e) {             e.printStackTrace();         }         countDownLatch.await();         service.shutdown();          // 事务操作（curator独有的）         try {             Collection resultCollections=curatorFramework.inTransaction().create().forPath(＂/demo1＂,＂111＂.getBytes()).and().                     setData().forPath(＂/abc＂,＂trunsaction：test＂.getBytes()).and().commit();             for (CuratorTransactionResult result:resultCollections){                 System.out.println(result.getForPath()+＂->＂+result.getType());             }         } catch (Exception e) {             e.printStackTrace();         }     } }