Leader Election

Curator在选主方式上除了提供Leader Latch，还有一个更为经典的方式：Leader Election，一种基于选举而非抢占的选主方式。

1. 关键API

org.apache.curator.framework.recipes.leader.LeaderSelector

主API

org.apache.curator.framework.recipes.leader.LeaderSelectorListener

选主监听器
继承自org.apache.curator.framework.state.ConnectionStateListener

通知主节点选主成功

org.apache.curator.framework.recipes.leader.LeaderSelectorListenerAdapter

选主监听器的一个抽象类
增加了对链接状态监听的默认实现
用以处理zk链接问题引发的选主状态不同步

org.apache.curator.framework.recipes.leader.CancelLeadershipException

在 ConnectionStateListener.stateChanged()抛出
会引发LeaderSelector.interruptLeadership()调用
主身份被打断

2. 机制说明

Leader Election内部通过一个分布式锁来实现选主；

并且选主结果是公平的，zk会按照各节点请求的次序成为主节点。
3. 用法
3.1 创建

创建LeaderSelector：
方法1

public LeaderSelector(CuratorFramework client,String mutexPath,LeaderSelectorListener listener)

参数说明：

 client：zk客户端链接mutexPath：分组路径（zk中的path）（名字就带有mutex，所以和锁有关）listener：选主结果监听器

方法2

public LeaderSelector(CuratorFramework client,String mutexPath,ThreadFactory threadFactory,Executor executor,LeaderSelectorListener listener)

参数说明：

   client：zk客户端链接mutexPath：分组路径（zk中的path）（名字就带有mutex，所以和锁有关）threadFactory：内部线程工厂，可以对工作线程做一些标记executor：执行器，或者说线程池listener：选主结果监听器

3.2 使用

LeaderSelector创建好之后，必须执行：

leaderSelector.start();

启动后，如果当选为主则会触发监听器中的takeLeadership()方法。

和Leader Latch一样，无论结果如何最终应该调用：

leaderSelector.close();

4. 错误处理

   LeaderSelectorListener继承自ConnectionStateListener。 当LeaderSelector启动后，会自动添加监听。 使用LeaderSelector时，必须关注链接状态的变化。 如果当选为主，应该处理链接中断：SUSPENDED，以及链接丢失：LOST。 当遇到SUSPENDED状态时，实例必须认为自己不再是主了，直到链接恢复到RECONNECTED状态。 当遇到LOST状态，实例不再是主了，并且应该退出takeLeadership方法。

   重要：建议当遇到SUSPENDED以及LOST时，直接抛出CancelLeadershipException异常 这样，会让LeaderSelector尝试中断任务执行并取消执行线程对takeLeadership 的执行。 正是因为这样，才提供了一个LeaderSelectorListenerAdapter,来处理上述逻辑。 所以，在实际使用中最好继承LeaderSelectorListenerAdapter使用。

5. 源码分析

5.1 类定义

还是先来看看类的定义
5.1.1 LeaderSelector

import java.io.Closeable;public class LeaderSelector implements Closeable {}

同样，实现了java.io.Closeable，所以你懂的，try()…catch{}。

5.1.2 LeaderSelectorListener

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.state.ConnectionStateListener;public interface LeaderSelectorListener extends ConnectionStateListener
{public void takeLeadership(CuratorFramework client) throws Exception;
}

继承自ConnectionStateListener,选主的有效性有链接状态密切关注

   takeLeadership方法当选为主后，此方法被调用此方法用于执行任务不用立即返回直到打算放弃主时，才应该结束此方法

5.1.3 LeaderSelectorListenerAdapter

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.state.ConnectionState;public abstract class LeaderSelectorListenerAdapter implements LeaderSelectorListener
{@Overridepublic void stateChanged(CuratorFramework client, ConnectionState newState){if ( (newState == ConnectionState.SUSPENDED) || (newState == ConnectionState.LOST) ){throw new CancelLeadershipException();}}
}

抽象类
带有ConnectionStateListener的stateChanged默认实现
参见：4. 错误处理

5.1.4 CancelLeadershipException

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.state.ConnectionState;public class CancelLeadershipException extends RuntimeException
{public CancelLeadershipException(){}public CancelLeadershipException(String message){super(message);}public CancelLeadershipException(String message, Throwable cause){super(message, cause);}public CancelLeadershipException(Throwable cause){super(cause);}
}

运行时异常
注意：
只有LeaderSelectorListener#stateChanged方法中抛出才能引发LeaderSelector#interruptLeadership()

5.2 成员变量

public class LeaderSelector implements Closeable
{private final Logger log = LoggerFactory.getLogger(getClass());private final CuratorFramework client;private final LeaderSelectorListener listener;private final CloseableExecutorService executorService;private final InterProcessMutex mutex;private final AtomicReference<State> state = new AtomicReference<State>(State.LATENT);private final AtomicBoolean autoRequeue = new AtomicBoolean(false);private final AtomicReference<Future<?>> ourTask = new AtomicReference<Future<?>>(null);private volatile boolean hasLeadership;private volatile String id = "";@VisibleForTestingvolatile CountDownLatch debugLeadershipLatch = null;volatile CountDownLatch debugLeadershipWaitLatch = null;private enum State { LATENT, STARTED, CLOSED }// guarded by synchronizationprivate boolean isQueued = false;private static final ThreadFactory defaultThreadFactory = ThreadUtils.newThreadFactory("LeaderSelector");
}

   log : slf4jclient : zk客户端（curator-framework提供）listener : 监听选主成功，并被回调executorService ： 线程池，同样实现了java.io.Closeablemutex : 分布式锁对象state :内部枚举LATENT 休眠STARTED 已启动CLOSED 已关闭状态原子化引用包装autoRequeue :是否自动重新参与选主原子化对象ourTask ：任务的异步Future持有原子化引用包装hasLeadershipvolatile可见性不同于LeaderLatch没有采用AtomicBoolean任务采用线程池选主回调更新状态所以对于hasLeadership的并发竞争少id : 参与者iddebugLeadershipLatch : 测试时使用debugLeadershipWaitLatch ： 试时使用isQueued :是否已经在排队中安全性由synchronized保障defaultThreadFactory私有常量默认线程工厂选主线程带有"Curator-LeaderSelector"前缀

注意：

和LeaderLatch不同，
虽然大部分变量采用了final,并采用Atom*进行包装
但是有些只采用volatile,只是保证了可见性
可见，同样是选主，但是LeaderLatch 和 Leader Election 还是有很多不一样的

5.3 构造器

常规套路，多个构造器模板，最终由下面这个完成：

public LeaderSelector(CuratorFramework client, String leaderPath, CloseableExecutorService executorService, LeaderSelectorListener listener){Preconditions.checkNotNull(client, "client cannot be null");PathUtils.validatePath(leaderPath);Preconditions.checkNotNull(listener, "listener cannot be null");this.client = client;this.listener = new WrappedListener(this, listener);hasLeadership = false;this.executorService = executorService;mutex = new InterProcessMutex(client, leaderPath){@Overrideprotected byte[] getLockNodeBytes(){return (id.length() > 0) ? getIdBytes(id) : null;}};}

可以发现：

与LeaderLatch不同，需要指定一个线程池异步任务去选主选主成功后执行listener的回调
client，leaderPath，listener不能为空
对listener进行了一层包装，参见 5.5 监听器包装
构造的过程中，初始化了对leaderPath进行了加锁

5.4 启动

第3节，介绍过。和LeaderLatch一样，Leader Election是由start()启动选主过程：

public void start()
{Preconditions.checkState(state.compareAndSet(State.LATENT, State.STARTED), "Cannot be started more than once");Preconditions.checkState(!executorService.isShutdown(), "Already started");Preconditions.checkState(!hasLeadership, "Already has leadership");client.getConnectionStateListenable().addListener(listener);requeue();
}
public boolean requeue()
{Preconditions.checkState(state.get() == State.STARTED, "close() has already been called");return internalRequeue();
}
private synchronized boolean internalRequeue()
{if ( !isQueued && (state.get() == State.STARTED) ){isQueued = true;Future<Void> task = executorService.submit(new Callable<Void>(){@Overridepublic Void call() throws Exception{try{doWorkLoop();}finally{clearIsQueued();if ( autoRequeue.get() ){internalRequeue();}}return null;}});ourTask.set(task);return true;}return false;
}

可以发现：

CAS操作，更新状态从休眠到已启动
如果线程池已关闭已经关闭则认定:"Already started"
坚持是否已经当选成功
在链接上添加选主监听器
执行requeue()方法，重新排队选主确认当前状态是已启动调用internalRequeue()方法synchronized同步互斥如果当前已经在排队中，则返回false否则更新isQueued = true由于synchronized，所以isQueued是安全更新向线程池提交一个异步任务由ourTask持有此任务的Future返回true

继续看看提交的异步任务，做了哪些事：

调用doWorkLoop()
通过finally调用clearIsQueued();synchronizedisQueued = false如果开启自动重新排队，则再次调用internalRequeue()有点类似递归但不是递归不断重新排队在当前任务结束时，通过重新调用所在方法重新向线程池中建立一个同样的任务

那么先来看看干活的doWorkLoop()方法：

private void doWorkLoop() throws Exception
{KeeperException exception = null;try{doWork();}catch ( KeeperException.ConnectionLossException e ){exception = e;}catch ( KeeperException.SessionExpiredException e ){exception = e;}catch ( InterruptedException ignore ){Thread.currentThread().interrupt();}if ( (exception != null) && !autoRequeue.get() )   // autoRequeue should ignore connection loss or session expired and just keep trying{throw exception;}}

异常持有，常规套路
可以发现干活的是doWork()
当遇到java.lang.InterruptedException，则线程重新进行锁竞争（synchronized）
如果开启了自动重新排队，则不抛出异常任务正常结束重做任务

继续看看doWork()

@VisibleForTestingvoid doWork() throws Exception{hasLeadership = false;try{mutex.acquire();hasLeadership = true;try{if ( debugLeadershipLatch != null ){debugLeadershipLatch.countDown();}if ( debugLeadershipWaitLatch != null ){debugLeadershipWaitLatch.await();}listener.takeLeadership(client);}catch ( InterruptedException e ){Thread.currentThread().interrupt();throw e;}catch ( Throwable e ){ThreadUtils.checkInterrupted(e);}finally{clearIsQueued();}}catch ( InterruptedException e ){Thread.currentThread().interrupt();throw e;}finally{if ( hasLeadership ){hasLeadership = false;try{mutex.release();}catch ( Exception e ){ThreadUtils.checkInterrupted(e);log.error("The leader threw an exception", e);// ignore errors - this is just a safety}}}}

初始hasLeadership=false
申请对leaderPath加锁，阻塞
加锁成功，则当选为主hasLeadership = true
回调org.apache.curator.framework.recipes.leader.LeaderSelectorListener的takeLeadership方法此回调是在doWork()中同步调用也即是在doWorkLoop()中也就是在executorService线程池中的一个异步任务中调用的
对中断异常进行处理
内层finally清理排队中标识
外层finally进行着主任务执行完成后的清理工作如果是主还原主标识对leaderPath解锁

5.5 关闭

public synchronized void close()
{Preconditions.checkState(state.compareAndSet(State.STARTED, State.CLOSED), "Already closed or has not been started");client.getConnectionStateListenable().removeListener(listener);executorService.close();ourTask.set(null);
}

CAS操作，将状态从已启动更新为已关闭
清除掉链接上的监听
关闭线程池
制空异步任务Future

5.6 监听器包装

在构造器中，可以发现，对于监听器，LeaderSelector是做了一层包装的：

private static class WrappedListener implements LeaderSelectorListener{private final LeaderSelector leaderSelector;private final LeaderSelectorListener listener;public WrappedListener(LeaderSelector leaderSelector, LeaderSelectorListener listener){this.leaderSelector = leaderSelector;this.listener = listener;}@Overridepublic void takeLeadership(CuratorFramework client) throws Exception{listener.takeLeadership(client);}@Overridepublic void stateChanged(CuratorFramework client, ConnectionState newState){try{listener.stateChanged(client, newState);}catch ( CancelLeadershipException dummy ){leaderSelector.interruptLeadership();}}}

一个手工代理的套路
关键的对listener.stateChanged进行了代理增强
处理了CancelLeadershipException
进行了leaderSelector.interruptLeadership()
这就是上文说在takeLeadership中抛出CancelLeadershipException,才会得到妥善的清理

5.7 中断

那再来看看leaderSelector是如何处理中断的：

public synchronized void interruptLeadership()
{Future<?> task = ourTask.get();if ( task != null ){task.cancel(true);}
}

synchronized
就是拿到线程池中正在执行的任务的Future
调用Future的cancel取消执行

可以看出这里对任务的处理要好于Leader Latch的方式。在Leader Latch需要自行处理中断
6. 小结

通过源码发现LeaderSelector选主，完全是通过一个分布式公平锁来实现的。

内部使用线程池来执行选主任务，以及业务逻辑。而且，是可以重新排队的。

可以和Leader Latch制作一个小对比

方式 | 任务调用 | 重新选主 | 公平性 | 适用场景 —|--- Leader Election | 异步 | 自动 | 公平 | 分布式任务 Leader Latch | 同步 | 手工实现 | 非公平· | 热备·

Leader Latch 采用有序临时节点，重入时顺序控制可能缺失公平性

Leader Latch 对于节点的监听，为避免惊群效应，采用的对参与者排序后，逐个监听上一位参与者

7. 测试

import org.apache.commons.lang3.RandomStringUtils
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
import org.junit.Before
import org.junit.Test
import java.util.*
import java.util.concurrent.CountDownLatch
import java.util.concurrent.TimeUnit/*** Created by roc on 2017/5/26.*/
class LeaderElectionTest {val LATCH_PATH: String = "/test/leader/election"var client: CuratorFramework = CuratorFrameworkFactory.builder().connectString("0.0.0.0:8888").connectionTimeoutMs(5000).retryPolicy(ExponentialBackoffRetry(1000, 10)).sessionTimeoutMs(3000).build()@Before fun init() {client.start()}@Test fun runTest() {var id: String = RandomStringUtils.randomAlphabetic(10)println("id : $id ")val time = Date()val latch: CountDownLatch = CountDownLatch(1)val leaderSelector: LeaderSelector = LeaderSelector(client, LATCH_PATH, object : LeaderSelectorListenerAdapter() {override fun takeLeadership(cc: CuratorFramework) {println("$id 当选 $time")while (true) {println("$id 执行中 $time")if (Math.random() > 0.89) {println("$id 退出此轮任务 $time")break;}TimeUnit.SECONDS.sleep(2)}}})leaderSelector.id = idleaderSelector.autoRequeue()leaderSelector.start()latch.await()leaderSelector.close()}}

zookeeper：

ls /test/leader/election
[_c_670dad30-2a82-4f93-81cf-02ef905c9a51-lock-0000000361,
_c_a4fd8057-c970-4dd4-aa25-fbf40e2f2151-lock-0000000360, _c_44e04a6d-d75b-480d-a46e-245612fec815-lock-0000000362]

get /test/leader/election/_c_4c1a24ff-e707-4c2e-a444-02ab8954f38b-lock-0000000364
djOWbQflak
cZxid = 0x1ddfe
ctime = Fri May 26 19:38:29 CST 2017
mZxid = 0x1ddfe
mtime = Fri May 26 19:38:29 CST 2017
pZxid = 0x1ddfe
cversion = 0
dataVersion = 0
aclVersion = 0
ephemeralOwner = 0x15156529fae07f0
dataLength = 10
numChildren = 0