Zookeeper客户端状态自我监听

缺乏、安全感 2022-09-11 13:13 184阅读 0赞

### ConnectionStateListener接口 ###

这个接口用于实现客户端与zk集群连接的状态发生变化时执行回调监听功能；当然，功能的前提是客户端服务处于正常运行状态。

假设我们要在服务启动时，手动创建一个zk临时节点，用于某些场景下的注册功能。

但是当客户端因为网络等原因断开连接，当网络恢复后，zk客户端与集群会重新连接，但是恢复连接后，临时节点并不能再次自动创建。

这种情况下我们可以监听重连事件，在重连后手动创建临时节点。

@Service
    @Slf4j
    public class ZkMonitorService implements CommandLineRunner {
    
        @Resource
        private CuratorFramework zkClient;
    
        private static String NODE = "/mainDataNode";
    
        @PostConstruct
        public void monitorRegister() {
            String ip = NetUtil.localIpv4s().stream().findFirst().get();
            try {
                //注册主节点
                Stat stat = zkClient.checkExists().forPath(NODE);
                if (ObjectUtil.isEmpty(stat)) {
                    zkClient.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT).withACL(ZooDefs.Ids.OPEN_ACL_UNSAFE).forPath(NODE, NODE.getBytes("UTF-8"));
                }
                //注册临时节点
                String subNode = NODE + "/" + ip + "_";
                //验证
                Stat stat1 = zkClient.checkExists().forPath(subNode);
                if (Objects.isNull(stat1)) {
                    zkClient.create().withMode(CreateMode.EPHEMERAL_SEQUENTIAL).forPath(subNode, ip.getBytes("UTF-8"));
                    log.info("=======服务节点启动,注册临时节点：{}", subNode);
                }
            } catch (Exception e) {
                throw new BizException(String.format("%s创建子节点失败，请重新尝试", ip));
            }
        }
    
        /**
         * 创建监听器实例,监听状态变更事件
         */
        @Override
        public void run(String... args) throws Exception {
            //注册监听
            OrderMonitorListener stateListener = new OrderMonitorListener(zkClient);
            zkClient.getConnectionStateListenable().addListener(stateListener);
        }
    }

监听实现

@Slf4j
    public class OrderMonitorListener implements ConnectionStateListener {
    
        private CuratorFramework zkClient;
    
        public OrderMonitorListener(CuratorFramework zkClient) {
            this.zkClient = zkClient;
        }
    
        @Override
        public void stateChanged(CuratorFramework curatorFramework, ConnectionState connectionState) {
            if (connectionState == ConnectionState.LOST) {
                log.info("===========监听到zk节点发生LOST事件");
            } else if (connectionState == ConnectionState.CONNECTED) {
                log.info("===========监听到zk节点发生CONNECTED事件");
            } else if (connectionState == ConnectionState.RECONNECTED) {
                log.info("===========监听到zk节点发生RECONNECTED事件:【RECONNECTED不会重新创建临时节点,需要手动创建】");
                String ip = NetUtil.localIpv4s().stream().findFirst().get();
                String node = "/mainDataNode/" + ip + "_";
                try {
                    zkClient.create().withMode(CreateMode.EPHEMERAL_SEQUENTIAL).forPath(node, ip.getBytes("UTF-8"));
                    log.info("=======服务节点RECONNECTED重连,重新创建临时节点：{}", node);
                } catch (Exception e) {
                    e.printStackTrace();
                }
    
            }
        }
    
        @Override
        public boolean doNotDecorate() {
            return false;
        }
    
    }