Zookeeper做主从选举

超、凢脫俗 2022-09-01 08:46 107阅读 0赞

## 相关命令 ##

create [-s] [-e] [-c] [-t ttl] path [data] [acl]
    ls [-s] [-w] [-R] path

## 实现思路 ##

*  在指定目录下，创建有序且临时节点，客户端根据自身创建的节点是否是 “序号最小”的节点进行判断
 *  如果是 “序号最小” 的节点，则认为当前客户端是 master
 *  如果非 “序号最小” 的节点，则认为当前客户端是 slaver

## 考虑点 ##

*  网络断线重连后（Event.KeeperState.SyncConnected），要检查自己是否为 master
 *  网络断线过久，重连后Session过期(Event.KeeperState.Expired)，要重新建立 zookeeper 实例，并重新建立节点，判断 master
 *  其他KeeperState=Disconnected事件触发时，最好先走一遍 退出master的逻辑，防止出现2个 master
 *  不能完全依赖 watcher事件，要建立主动校正机制；例如原master宕机后，服务器触发watcher事件，但此时到其他client的网络均异常，没有重新进行 选举

## 代码示例实现 ##

@Slf4j
    public class Zookeeper357Test { 
        public static String currentPath;
        public static ZooKeeper zooKeeper;
    
        public static void main(String[] args) throws IOException, InterruptedException, KeeperException { 
            reConnect();
            while (true) { 
    
            }
        }
        public static void reConnect() throws IOException, KeeperException, InterruptedException { 
            zooKeeper = new ZooKeeper("127.0.0.1:2181", 30000, new ConnectWatcher(), new ZKClientConfig());
    
            // 3.6.0 以上版本，添加一次 watcher 即可
           // zooKeeper.addWatch("/zk_temp",new LeaderWatcher(zooKeeper), AddWatchMode.PERSISTENT_RECURSIVE);
    
            LeaderWatcher leaderWatcher = new LeaderWatcher(zooKeeper);
            zooKeeper.getChildren("/zk_temp", leaderWatcher, null);
    
            // 启动的时候，注册下临时节点
            currentPath = zooKeeper.create("/zk_temp/", null, ZooDefs.Ids.OPEN_ACL_UNSAFE,
                    CreateMode.EPHEMERAL_SEQUENTIAL);
        }
    }
    
    
    
    @Slf4j
    public class LeaderWatcher implements Watcher { 
    
        private ZooKeeper zooKeeper;
    
        public LeaderWatcher() { 
        }
    
        public LeaderWatcher(ZooKeeper zooKeeper) { 
            this.zooKeeper = zooKeeper;
        }
    
        @SneakyThrows
        @Override
        public void process(WatchedEvent event) { 
    
            if (Objects.equals(event.getType(), Event.EventType.NodeChildrenChanged)) { 
                checkMaster();
            }
    
            if (Objects.equals(event.getType(), Event.EventType.None)) { 
                if (Objects.equals(event.getState(), Event.KeeperState.SyncConnected)) { 
                    log.info("网络恢复连接，检查下我是不是master :{}", Zookeeper357Test.currentPath);
                    checkMaster();
                } else if (Objects.equals(event.getState(), Event.KeeperState.Expired)) { 
                    log.info("- session过期了，重新建立连接 - :{}", Zookeeper357Test.currentPath);
                    Zookeeper357Test.reConnect();
                } else if (!Objects.equals(event.getState(), Event.KeeperState.SyncConnected)) { 
                    log.info("- 可能网络出现了问题，安全起见，我先不当master了 - :{}", Zookeeper357Test.currentPath);
                }
            }
    
        }
    
        private void checkMaster() throws KeeperException, InterruptedException { 
            List<String> children = zooKeeper.getChildren("/zk_temp", new LeaderWatcher(zooKeeper), null);
            List<String> orderedList = children.stream().sorted().collect(Collectors.toList());
            if (Objects.equals(Zookeeper357Test.currentPath, "/zk_temp/" + orderedList.get(0))) { 
                log.info("***************我是master :{} **********", Zookeeper357Test.currentPath);
            } else { 
                log.info("- 我是slaver - :{}", Zookeeper357Test.currentPath);
            }
        }
    }