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

淡淡的烟草味﹌ 2024-04-18 07:09 58阅读 0赞

## zookeeper分布式集群Curator的LeaderLatch节点选举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.LeaderLatch;
    import org.apache.curator.framework.recipes.leader.LeaderLatchListener;
    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<LeaderLatch> leaders = 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 + ",不需重复创建");
                }
    
                LeaderLatch leader = new LeaderLatch(client, path, "leaderId#" + i);
                leader.start();
                leaders.add(leader);
    
                leader.addListener(new LeaderLatchListener() {
                    @Override
                    public void isLeader() {
                        try {
                            System.out.println(TAG + "isLeader ? Leader:" + leader.getId());
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                    }
    
                    @Override
                    public void notLeader() {
                        System.out.println(TAG + "不是Leader");
                    }
                });
            }
    
            TimeUnit.SECONDS.sleep(10);
    
            leaderStatus(clients, leaders);
        }
    
        private void leaderStatus(List<CuratorFramework> clients, List<LeaderLatch> leaders) throws Exception {
            LeaderLatch currentLeader = null;
            for (LeaderLatch l : leaders) {
                if (l.hasLeadership()) {
                    currentLeader = l;
                    break;
                }
            }
    
            System.out.println(TAG + "现任Leader:" + currentLeader.getId());
    
            System.out.println(TAG + currentLeader.getId() + "卸任");
            //leader释放，卸任。
            currentLeader.close();
    
            TimeUnit.SECONDS.sleep(10);
    
            for (LeaderLatch l : leaders) {
                if (l.hasLeadership()) {
                    currentLeader = l;
                    break;
                }
            }
    
            System.out.println(TAG + "继任Leader:" + currentLeader.getId());
        }
    
        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
    调试# isLeader ? Leader:leaderId#0
    调试# 已存在:/path/leader,不需重复创建
    调试# 已存在:/path/leader,不需重复创建
    调试# 已存在:/path/leader,不需重复创建
    调试# 已存在:/path/leader,不需重复创建
    调试# 现任Leader:leaderId#0
    调试# leaderId#0卸任
    调试# isLeader ? Leader:leaderId#1
    调试# 继任Leader:leaderId#1

在构造LeaderLatch时候需要传入一个自己设置的id，否则在后面的程序代码中，LeaderLatch的getId为空。

运行该程序前，需要把zookeeper集群启动起来，我的例子中，我事先启动了三个集群节点。

LeaderLatch通过close()释放、卸任leader，否则将一直持有leader权限。