zookeeper分布式集群Curator的LeaderSelector节点选举Leader

红太狼 2024-04-18 07:31 78阅读 0赞

## ZooKeeper分布式集群Curator的LeaderSelector节点选举Leader ##

import org.apache.curator.RetryPolicy;
    import org.apache.curator.framework.CuratorFramework;
    import org.apache.curator.framework.CuratorFrameworkFactory;
    import org.apache.curator.framework.recipes.leader.LeaderSelector;
    import org.apache.curator.framework.recipes.leader.LeaderSelectorListener;
    import org.apache.curator.framework.state.ConnectionState;
    import org.apache.curator.retry.ExponentialBackoffRetry;
    import org.apache.zookeeper.CreateMode;
    import org.apache.zookeeper.data.Stat;
    
    import java.util.ArrayList;
    import java.util.List;
    import java.util.concurrent.TimeUnit;
    
    public class Main {
        private final String TAG = "调试==>> ";
        private int COUNT = 5;
    
        public static void main(String[] args) {
            //初始化log4j，zookeeper否则报错。
            //org.apache.log4j.BasicConfigurator.configure();
    
            try {
                Main app = new Main();
                app.zk();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    
        private void zk() throws Exception {
            String path = "/path/leader";
    
            List<CuratorFramework> clients = new ArrayList<>();
            List<LeaderSelector> leaderSelectors = new ArrayList<>();
    
            for (int i = 0; i < COUNT; i++) {
                CuratorFramework client = makeClient();
                client.start();
                clients.add(client);
    
    
                //检测是否存在该路径。
                Stat stat = client.checkExists().forPath(path);
    
                //如果不存在这个路径，stat为null，创建新的节点路径。
                if (stat == null) {
                    String s = client.create()
                            .creatingParentsIfNeeded()
                            .withMode(CreateMode.PERSISTENT)
                            .forPath(path);
    
                    System.out.println(TAG + "创建-" + s);
                } else {
                    System.out.println(TAG + "已存在:" + path + ",不需重复创建");
                }
    
    
                String leaderId = "#" + i;
    
                //不再使用LeaderSelector时，应该调用它的close方法。
                LeaderSelector leaderSelector = new LeaderSelector(client, path, new LeaderSelectorListener() {
                    @Override
                    public void takeLeadership(CuratorFramework c) throws Exception {
                        //打印成为leader的信息。
                        info(leaderSelectors);
                    }
    
                    @Override
                    public void stateChanged(CuratorFramework client, ConnectionState newState) {
                        String s = newState.isConnected() ? "连接" : "断连";
                        System.out.println(TAG + s);
                    }
                });
                leaderSelector.setId(leaderId);
    
                leaderSelectors.add(leaderSelector);
    
                //启动竞选leader
                leaderSelector.autoRequeue();
                leaderSelector.start();
            }
    
            TimeUnit.SECONDS.sleep(3);
        }
    
        private void info(List<LeaderSelector> leaderSelectors) {
            for (LeaderSelector l : leaderSelectors) {
                if (l.hasLeadership()) {
                    System.out.println(TAG + "现任Leader:" + l.getId()+"@"+System.currentTimeMillis());
                }
            }
        }
    
        private CuratorFramework makeClient() {
            RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 5);
            CuratorFramework client = CuratorFrameworkFactory.builder()
                    .connectString(getAddress())
                    .sessionTimeoutMs(10 * 1000)
                    .connectionTimeoutMs(20 * 1000)
                    .retryPolicy(retryPolicy)
                    .build();
    
            return client;
        }
    
        private String getAddress() {
            String ip = "127.0.0.1";
            return ip + ":2181," + ip + ":2182," + ip + ":2183";
        }
    }

输出：

调试==>> 创建-/path/leader
    调试==>> 已存在:/path/leader,不需重复创建
    调试==>> 现任Leader:#0@1567501371544
    调试==>> 现任Leader:#1@1567501371610
    调试==>> 已存在:/path/leader,不需重复创建
    调试==>> 现任Leader:#0@1567501371716
    调试==>> 已存在:/path/leader,不需重复创建
    调试==>> 现任Leader:#2@1567501371856
    调试==>> 已存在:/path/leader,不需重复创建
    调试==>> 现任Leader:#1@1567501372007
    调试==>> 现任Leader:#3@1567501372091
    调试==>> 现任Leader:#0@1567501372133
    调试==>> 现任Leader:#4@1567501372263
    调试==>> 现任Leader:#2@1567501372350
    调试==>> 现任Leader:#1@1567501372433
    调试==>> 现任Leader:#3@1567501372475
    调试==>> 现任Leader:#0@1567501372584
    调试==>> 现任Leader:#4@1567501372648
    调试==>> 现任Leader:#2@1567501372717
    调试==>> 现任Leader:#1@1567501372792
    调试==>> 现任Leader:#3@1567501372889
    调试==>> 现任Leader:#0@1567501372958
    调试==>> 现任Leader:#4@1567501373033
    调试==>> 现任Leader:#2@1567501373132
    调试==>> 现任Leader:#1@1567501373201
    调试==>> 现任Leader:#3@1567501373298
    调试==>> 现任Leader:#0@1567501373417
    调试==>> 现任Leader:#4@1567501373500
    调试==>> 现任Leader:#2@1567501373583
    调试==>> 现任Leader:#1@1567501373685
    调试==>> 现任Leader:#3@1567501373849
    调试==>> 现任Leader:#0@1567501373892
    调试==>> 现任Leader:#4@1567501374021
    调试==>> 现任Leader:#2@1567501374061
    调试==>> 现任Leader:#1@1567501374157
    调试==>> 现任Leader:#3@1567501374225
    调试==>> 现任Leader:#0@1567501374323
    调试==>> 现任Leader:#4@1567501374407
    调试==>> 现任Leader:#2@1567501374525
    调试==>> 现任Leader:#1@1567501374620
    调试==>> 现任Leader:#3@1567501374701