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
{
    @Override
    public 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 = "";

    @VisibleForTesting
    volatile CountDownLatch debugLeadershipLatch = null;
    volatile CountDownLatch debugLeadershipWaitLatch = null;

    private enum State { LATENT, STARTED, CLOSED }

    // guarded by synchronization
    private boolean isQueued = false;

    private static final ThreadFactory defaultThreadFactory = ThreadUtils.newThreadFactory("LeaderSelector");
}

   log : slf4j
    client : zk客户端（curator-framework提供）
    listener : 监听选主成功，并被回调
    executorService ： 线程池，同样实现了java.io.Closeable
    mutex : 分布式锁对象
    state :
        内部枚举
            LATENT 休眠
            STARTED 已启动
            CLOSED 已关闭
        状态
        原子化引用包装
    autoRequeue :
        是否自动重新参与选主
        原子化对象
    ourTask ：
        任务的异步Future持有
        原子化引用包装
    hasLeadership
        volatile可见性
        不同于LeaderLatch
            没有采用AtomicBoolean
            任务采用线程池
            选主回调更新状态
            所以对于hasLeadership的并发竞争少
    id : 参与者id
    debugLeadershipLatch : 测试时使用
    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)
        {
            @Override
            protected 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>()
        {
            @Override
            public 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();
        synchronized
        isQueued = 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()

@VisibleForTesting
    void 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;
        }

        @Override
        public void takeLeadership(CuratorFramework client) throws Exception
        {
            listener.takeLeadership(client);
        }

        @Override
        public 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 = id
        leaderSelector.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