Apache Curator Leader选举 简单示例

<dependency>
<groupId>org.apache.curator</groupId>
<artifactId>curator-recipes</artifactId>
<version>2.10.0</version>
</dependency>

package com.zk.leader;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.leader.LeaderLatch;
import org.apache.curator.retry.ExponentialBackoffRetry;

/**
* 必须启动LeaderLatch: leaderLatch.start();
* 一旦启动， LeaderLatch会和其它使用相同latch path的其它LeaderLatch交涉，然后随机的选择其中一个作为leader。
* 你可以随时查看一个给定的实例是否是leader:
*/
public class Example1
{
private static final String PATH = "/leader";
private static final int COUNT = 5;

public static void main(String[] args) throws Exception
{
CuratorFramework client = CuratorFrameworkFactory.newClient("192.168.1.100:2181", new ExponentialBackoffRetry(1000, 3));
client.start();

List<LeaderLatch> list = new ArrayList<>();

for (int i = 1; i <= COUNT; i++)
{
LeaderLatch example = new LeaderLatch(client, PATH, "Client #" + i);
example.start();
list.add(example);
}

TimeUnit.SECONDS.sleep(2);

LeaderLatch leader = null;
for (LeaderLatch ll : list)
{
if (ll.hasLeadership())
{
leader = ll;
}
System.out.println(ll.getId() + "\t" + ll.hasLeadership());
}
;

TimeUnit.SECONDS.sleep(2);

list.remove(leader);
leader.close();

TimeUnit.SECONDS.sleep(2);

System.out.println("========================");

for (LeaderLatch ll : list)
{
System.out.println(ll.getId() + "\t" + ll.hasLeadership());
}

for (LeaderLatch ll : list)
{
ll.close();
}

client.close();
}
}

Client #1	false
Client #2	false
Client #3	true
Client #4	false
Client #5	false
========================
Client #1	false
Client #2	false
Client #4	false
Client #5	true

package com.zk.leader;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;

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.LeaderSelectorListenerAdapter;
import org.apache.curator.retry.ExponentialBackoffRetry;

/**
* 与LeaderLatch相比， 通过LeaderSelectorListener可以对领导权进行控制， 在适当的时候释放领导权，这样每个节点都有可能获得领导权。
* 而LeaderLatch一根筋到死， 除非调用close方法，否则它不会释放领导权。
*/
public class Example2
{
private static final String PATH = "/leader";
private static final int COUNT = 5;

public static void main(String[] args) throws Exception
{
CuratorFramework client = CuratorFrameworkFactory.newClient("192.168.1.100:2181", new ExponentialBackoffRetry(1000, 3));
client.start();

List<MyLeaderSelectorListenerAdapter> list = new ArrayList<>();

for (int i = 1; i <= COUNT; i++)
{
MyLeaderSelectorListenerAdapter example = new MyLeaderSelectorListenerAdapter(client, PATH, "Client #" + i);
list.add(example);
}

TimeUnit.SECONDS.sleep(20);

for (MyLeaderSelectorListenerAdapter listener : list)
{
listener.close();
}

client.close();
}
}
class MyLeaderSelectorListenerAdapter extends LeaderSelectorListenerAdapter
{
private final String name;
private final LeaderSelector leaderSelector;

public MyLeaderSelectorListenerAdapter(CuratorFramework client, String path, String name)
{
this.name = name;
leaderSelector = new LeaderSelector(client, path, this);

//保证在此实例释放领导权之后还可能获得领导权。
leaderSelector.autoRequeue();

leaderSelector.start();
}

public void close()
{
leaderSelector.close();
}

/**
* 你可以在takeLeadership进行任务的分配等等，并且不要返回，如果你想要要此实例一直是leader的话可以加一个死循环。
* 一旦此方法执行完毕之后，就会重新选举
*/
public void takeLeadership(CuratorFramework client) throws Exception
{
System.out.println(name + " 当选为leader");
TimeUnit.SECONDS.sleep(2);
}
}

Client #1 当选为leader
Client #4 当选为leader
Client #2 当选为leader
Client #3 当选为leader
Client #5 当选为leader
Client #1 当选为leader
Client #4 当选为leader
Client #2 当选为leader
Client #3 当选为leader
Client #5 当选为leader