Zookeeper遇到的BUG

Myth丶恋晨 2024-04-06 12:53 148阅读 0赞

Zookeeper遇到的BUG-不要在回调函数中阻塞线程

在zookeeper api实现配置中心时，遇到watch失效问题，watch到父级节点下发生变更时，再次getChildren(),最终因其没有调用原先我们预料到的回调函数，导致问题出现。当然很少有人会使用原生的zookeeper api，使用zookeeper客户端直接封装自然可以解决问题。

最终解决方案watch回调函数再次去getData时开一个线程去做。

整体源码:

  1. public class ConfigServer implements AsyncCallback.StatCallback, Watcher, AsyncCallback.ChildrenCallback,
  2.         AsyncCallback.Create2Callback {
  4.     private String path;//path是配置数据节点的父级目录，表示当前集群需要从zk集群的哪个目录下取配置信息
  5.     private ZooKeeper zk;
  7.     private String name;
  8.     private byte[] data;
  10.     private CountDownLatch publishCountDownLatch;
  11.     private CountDownLatch getDateCountDownLatch;
  13.     private CountDownLatch count;
  14.     private HashMap<String, byte[]> res;
  15.     public ConfigServer(String path, ZooKeeper zk) {
  17.         this.path = path;
  18.         this.zk = zk;
  19.         res=new HashMap<>();
  20.     }
  21.     //zk集群是二进制安全的，实际上内部存储的是字节流数据
  22.     public void publish(String name,byte[] data) throws InterruptedException {
  24.         publishCountDownLatch=new CountDownLatch(1);
  25.         this.name=name;
  26.         this.data=data;
  27.         String realPath=path+"/"+name;
  28.         zk.create(realPath,data, ZooDefs.Ids.OPEN_ACL_UNSAFE,CreateMode.PERSISTENT,this,"kuailexingqiu");
  29.         publishCountDownLatch.await();
  30.     }
  31.     public void delete(String name) throws InterruptedException, KeeperException {
  33.         Stat stat = zk.exists(path + "/" + name, false);
  34.         if(stat!=null){
  36.             zk.delete(path+"/"+name,stat.getVersion()+1);
  37.         }
  38.     }
  39.     public HashMap<String, byte[]> getData() throws InterruptedException {
  40.     //把path父级目录下所有内容查出来
  41.         System.out.println("--------------------"+Thread.currentThread().getName()+"-----------------------");
  42.         getDateCountDownLatch=new CountDownLatch(1);
  43.         res=new HashMap<>();
  45.         new Thread(()->{
  47.             zk.getChildren(path,this,this,"kuailexingqiu");
  48.         }).start();
  50.         getDateCountDownLatch.await();
  51.         System.out.println("------------------进来了-----------------");
  52.         //其实在这步要将map中的数据写入配置文件中，在主动触发配置文件自动刷新
  53.         if(res!=null){
  55.             Set<String> keySet = res.keySet();
  56.             for (String s : keySet) {
  58.                 System.out.println("----------------------------------------");
  59.                 System.out.println(s+"===="+new String(res.get(s)));
  60.                 System.out.println("----------------------------------------");
  61.             }
  62.             return res;
  63.         }
  64.         return null;
  65.     }
  66.     public String getPath() {
  68.         return path;
  69.     }
  70.     public void setPath(String path) {
  72.         this.path = path;
  73.     }
  75.     @Override
  76.     public void processResult(int rc, String path, Object ctx, Stat stat) {
  78.         if(stat==null){
  80.             zk.setData(path,data,0,this,"kuailexingqiu");
  81.         }else{
  83.             publishCountDownLatch.countDown();
  84.         }
  85.     }
  87.     @Override
  88.     public void process(WatchedEvent event) {
  90.         Event.EventType type = event.getType();
  91.         System.out.println("--------------------"+Thread.currentThread().getName()+"-----------------------");
  92.         switch (type) {
  94.             case None:
  95.                 break;
  96.             case NodeCreated:
  98.                 break;
  99.             case NodeDeleted:
  100.                 try {
  102.                     getData();
  103.                 } catch (InterruptedException e) {
  105.                     e.printStackTrace();
  106.                 }
  107.                 break;
  108.             case NodeDataChanged:
  109.                 break;
  110.             case NodeChildrenChanged:
  111. //                new Thread(()->{
  113. //                    try {
  115.                 try {
  117.                     getData();
  118.                 } catch (InterruptedException e) {
  120.                     e.printStackTrace();
  121.                 }
  122. //                    } catch (InterruptedException e) {
  124. //                        e.printStackTrace();
  125. //                    }
  126. //                }).start();
  127.                 break;
  128.             case DataWatchRemoved:
  129.                 break;
  130.             case ChildWatchRemoved:
  131.                 break;
  132.             case PersistentWatchRemoved:
  133.                 break;
  134.         }
  135.     }
  137.     @Override
  138.     public void processResult(int rc, String path, Object ctx, List<String> children) {
  140.         System.out.println("--------------------"+Thread.currentThread().getName()+"-----------------------");
  141.         System.out.println("------------------哈哈哈哈------------------------- ");
  142.         if(children.isEmpty()){
  144.             getDateCountDownLatch.countDown();
  145.             return;
  146.         }
  147. //        count=new CountDownLatch(children.size());
  148.         for (String child : children) {
  150.             try {
  152.                 byte[] bytes = zk.getData(path + "/" + child, false, null);
  153.                 res.put(child,bytes);
  154.             } catch (KeeperException e) {
  156.                 e.printStackTrace();
  157.             } catch (InterruptedException e) {
  159.                 e.printStackTrace();
  160.             }
  161.         }
  162.         //上面的getData是异步的很可能发出children.len个请求后，就直接执行后面代码，countDown导致程序返回结果
  163. //        try {
  165. //            count.await();
  166. //        } catch (InterruptedException e) {
  168. //            e.printStackTrace();
  169. //        }
  170.         getDateCountDownLatch.countDown();
  171.     }
  173.     @Override
  174.     public void processResult(int rc, String path, Object ctx, String name, Stat stat) {
  176.         if(stat==null){
  178.             throw new RuntimeException("创建新的配置项失败!");
  179.         }else{
  181.             publishCountDownLatch.countDown();
  182.         }
  183.     }
  184. }
  186. @Override
  187.     public void process(WatchedEvent event) {
  189.         Event.EventType type = event.getType();
  190.         System.out.println(Thread.currentThread().getName());
  191.         switch (type) {
  193.             case None:
  194.                 break;
  195.             case NodeCreated:
  197.                 break;
  198.             case NodeDeleted:
  199.                 try {
  201.                     getData();
  202.                 } catch (InterruptedException e) {
  204.                     e.printStackTrace();
  205.                 }
  206.                 break;
  207.             case NodeDataChanged:
  208.                 break;
  209.             case NodeChildrenChanged:
  210.                 new Thread(()->{
  212.                     try {
  214.                         getData();
  215.                     } catch (InterruptedException e) {
  217.                         e.printStackTrace();
  218.                     }
  219.                 }).start();
  220.                 break;
  221.             case DataWatchRemoved:
  222.                 break;
  223.             case ChildWatchRemoved:
  224.                 break;
  225.             case PersistentWatchRemoved:
  226.                 break;
  227.         }
  228.     }

问题原因:zookeeper在响应式编程下，会定义回调方法，而客户端是通过main-Event线程来执行回调方法的，且默认情况下只使用main-Thread来进行回调。这也就意味着如果使用CountDownLatch将main-Event阻塞住，且这个阻塞必须要main-Event线程执行完回调方法后使计时器count==0时才会取消，这就会导致main-Event死锁。

而且如果有多个回调方法同时需要执行，那么会放入队列中，逐个执行。但也只会是单线程执行。

所以以后利用zk进行异步回调时，记住一定不要用countDownLatch阻塞住main-Event线程。

使用zk实现配置中心最终代码

  1. public class ConfigServer implements AsyncCallback.StatCallback, Watcher, AsyncCallback.ChildrenCallback,
  2.         AsyncCallback.Create2Callback {
  4.     private String path;//path是配置数据节点的父级目录，表示当前集群需要从zk集群的哪个目录下取配置信息
  5.     private ZooKeeper zk;
  7.     private String name;
  8.     private byte[] data;
  10.     private CountDownLatch publishCountDownLatch;
  11.     private CountDownLatch getDateCountDownLatch;
  13.     private CountDownLatch count;
  14.     private HashMap<String, byte[]> res;
  15.     public ConfigServer(String path, ZooKeeper zk) {
  17.         this.path = path;
  18.         this.zk = zk;
  19.         res=new HashMap<>();
  20.     }
  21.     //zk集群是二进制安全的，实际上内部存储的是字节流数据
  22.     public void publish(String name,byte[] data) throws InterruptedException {
  24.         publishCountDownLatch=new CountDownLatch(1);
  25.         this.name=name;
  26.         this.data=data;
  27.         String realPath=path+"/"+name;
  28.         zk.create(realPath,data, ZooDefs.Ids.OPEN_ACL_UNSAFE,CreateMode.PERSISTENT,this,"kuailexingqiu");
  29.         publishCountDownLatch.await();
  30.     }
  31.     public void delete(String name) throws InterruptedException, KeeperException {
  33.         Stat stat = zk.exists(path + "/" + name, false);
  34.         if(stat!=null){
  36.             zk.delete(path+"/"+name,stat.getVersion()+1);
  37.         }
  38.     }
  39.     public HashMap<String, byte[]> getData() throws InterruptedException {
  40.     //把path父级目录下所有内容查出来
  41.         System.out.println("--------------------"+Thread.currentThread().getName()+"-----------------------");
  42.         res=new HashMap<>();
  44.         getDateCountDownLatch=new CountDownLatch(1);
  45.         zk.getChildren(path,this,this,"kuailexingqiu");
  46.         getDateCountDownLatch.await();
  48.         System.out.println("------------------进来了-----------------");
  49.         //其实在这步要将map中的数据写入配置文件中，在主动触发配置文件自动刷新
  50.         if(res!=null){
  52.             Set<String> keySet = res.keySet();
  53.             for (String s : keySet) {
  55.                 System.out.println("----------------------------------------");
  56.                 System.out.println(s+"===="+new String(res.get(s)));
  57.                 System.out.println("----------------------------------------");
  58.             }
  59.             return res;
  60.         }
  61.         return null;
  62.     }
  63.     public String getPath() {
  65.         return path;
  66.     }
  67.     public void setPath(String path) {
  69.         this.path = path;
  70.     }
  72.     @Override
  73.     public void processResult(int rc, String path, Object ctx, Stat stat) {
  75.         if(stat==null){
  77.             zk.setData(path,data,0,this,"kuailexingqiu");
  78.         }else{
  80.             publishCountDownLatch.countDown();
  81.         }
  82.     }
  84.     @Override
  85.     public void process(WatchedEvent event) {
  87.         Event.EventType type = event.getType();
  88.         System.out.println("--------------------"+Thread.currentThread().getName()+"-----------------------");
  89.         switch (type) {
  91.             case None:
  92.                 break;
  93.             case NodeCreated:
  95.                 break;
  96.             case NodeDeleted:
  97.                 try {
  99.                     getData();
  100.                 } catch (InterruptedException e) {
  102.                     e.printStackTrace();
  103.                 }
  104.                 break;
  105.             case NodeDataChanged:
  106.                 break;
  107.             case NodeChildrenChanged:
  108.                 new Thread(()->{
  110. //                    try {
  112.                 try {
  114.                     getData();
  115.                 } catch (InterruptedException e) {
  117.                     e.printStackTrace();
  118.                 }
  119. //                    } catch (InterruptedException e) {
  121. //                        e.printStackTrace();
  122. //                    }
  123.                 }).start();
  124.                 break;
  125.             case DataWatchRemoved:
  126.                 break;
  127.             case ChildWatchRemoved:
  128.                 break;
  129.             case PersistentWatchRemoved:
  130.                 break;
  131.         }
  132.     }
  134.     @Override
  135.     public void processResult(int rc, String path, Object ctx, List<String> children) {
  137.         System.out.println("--------------------"+Thread.currentThread().getName()+"-----------------------");
  138.         System.out.println("------------------哈哈哈哈------------------------- ");
  139.         if(children.isEmpty()){
  141.             getDateCountDownLatch.countDown();
  142.             return;
  143.         }
  144. //        count=new CountDownLatch(children.size());
  145.         for (String child : children) {
  147.             try {
  149.                 byte[] bytes = zk.getData(path + "/" + child, false, null);
  150.                 res.put(child,bytes);
  151.             } catch (KeeperException e) {
  153.                 e.printStackTrace();
  154.             } catch (InterruptedException e) {
  156.                 e.printStackTrace();
  157.             }
  158.         }
  159.         //上面的getData是异步的很可能发出children.len个请求后，就直接执行后面代码，countDown导致程序返回结果
  160. //        try {
  162. //            count.await();
  163. //        } catch (InterruptedException e) {
  165. //            e.printStackTrace();
  166. //        }
  167.         getDateCountDownLatch.countDown();
  168.     }
  170.     @Override
  171.     public void processResult(int rc, String path, Object ctx, String name, Stat stat) {
  173.         if(stat==null){
  175.             throw new RuntimeException("创建新的配置项失败!");
  176.         }else{
  178.             publishCountDownLatch.countDown();
  179.         }
  180.     }
  181. }

发表评论

表情：
评论列表 （有 0 条评论，148人围观）

还没有评论，来说两句吧...

相关阅读

    相关 程序员遇到bug怎么面对?

            这几天项目在做性能测试,做渗透测试的时候,给我提了两个安全方面的bug,之前写的时候没有考虑这个问题,bug提出来以后找了个有经验的人问了问怎么做,然后就开始动

    快来打我*快来打我*/ 0 赞/ 253 阅读

    相关 遇到bug怎么办？

    感觉作为一名程序员，如果用二分之一的时间写代码，那另外二分之一的时间就是在改bug。 作为一名前端，我记录一下自己遇到bug时候的解决情况。让自己日后不要犯下重复的错误。

    绝地灬酷狼绝地灬酷狼/ 0 赞/ 542 阅读