八curator recipes之选举主节点LeaderSelector

Posted 2021-02-03 lay2017

简介

前面我们看到LeaderLatch对于选举的实现：https://www.cnblogs.com/lay2017/p/10264300.html

节点在加入选举以后，除非程序结束或者close()退出选举，否则加点自加入选举以后将持续持有或者保持对主节点的竞争。

recipes的另外一个实现Leader Election则不同，被选为主节点的节点任务如果执行完就会放弃主节点，然后由剩下的节点进行主节点竞争。如果你希望已经执行完的主节点再次加入主节点选举那么你需要调用autoRequeue()方法去自动加入。

官方文档：http://curator.apache.org/curator-recipes/leader-election.html

javaDoc：http://curator.apache.org/apidocs/org/apache/curator/framework/recipes/leader/LeaderSelector.html

代码示例

以下代码你可以通过是否调用autoRequeue来看看console的输出有什么不同

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;

public class LeaderElectionDemo {
    private static CuratorFramework client = CuratorFrameworkFactory.newClient("localhost:2181", new ExponentialBackoffRetry(3000, 3));
    private static String path = "/mutex/path/0001";
    static {
        client.start();
    }

    public static void main(String[] args) throws InterruptedException {
        startThread0();
        Thread.sleep(10);
        startThread1();
        Thread.sleep(50000);
        client.close();
    }

    public static void startThread0() {
        new Thread(() -> {
            LeaderSelector selector = new LeaderSelector(client, path, new LeaderSelectorListener() {
                @Override
                public void takeLeadership(CuratorFramework curatorFramework) throws Exception {
                    System.out.println("thread0 is leader");
                }

                @Override
                public void stateChanged(CuratorFramework curatorFramework, ConnectionState connectionState) {
                    System.out.println("thread0 state changed，state is " + connectionState.isConnected());
                }
            });
            selector.autoRequeue();
            selector.start();
        }).start();
    }

    public static void startThread1() {
        new Thread(() -> {
            LeaderSelector selector = new LeaderSelector(client, path, new LeaderSelectorListener() {
                @Override
                public void takeLeadership(CuratorFramework curatorFramework) throws Exception {
                    System.out.println("thread1 is leader");
                }

                @Override
                public void stateChanged(CuratorFramework curatorFramework, ConnectionState connectionState) {
                    System.out.println("thread1 state changed，state is " + connectionState.isConnected());
                }
            });
            selector.autoRequeue();
            selector.start();
        }).start();
    }
}