Leader Latch

Zookeeper在分布式系统中，常常被用于选主。在执行某个任务时，让所有的节点都知道有一个特别的，唯一的节点是任务的主节点，由主节点进行任务的执行，其他节点作为备用节点。通过这种热备方式，为分布式系统中任务执行的可控性，以及系统高可用性。

而Curator提供了两种选主机制，可以根据实际情况进行选用。

1. 关键API

org.apache.curator.framework.recipes.leader.LeaderLatch

2. 机制说明

LeaderLatch的方式，就是以一种抢占的方式来决定选主。比较简单粗暴，逻辑相对简单。类似非公平锁的抢占，所以，多节点是一个随机产生主节点的过程。基本就是，谁抢到就算谁的。

多个参与者（如：逻辑节点；某个线程等），指定在一个分组之下，每个分组内进行主节点抢占。

3. 用法

3.1 创建

方法1

public LeaderLatch(CuratorFramework client,String latchPath)

参数说明：

• client : zk客户端链接
• latchPath : 分组路径（zk中的path）

方法2

public LeaderLatch(CuratorFramework client,String latchPath,String id)

参数说明：

• client : zk客户端链接
• latchPath : 分组路径（zk中的path）
• id : 参与者ID

3.2 使用

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

leaderLatch.start();

这样，才能让leaderLatch开始参与选主过程。

由于LeaderLatch是一个不断抢占的过程，所以需要调用：

public boolean hasLeadership()

来检测当前参与者是否选主成功。这个方法是非阻塞的（立即返回），其结果只代表调用时的选主结果。所以，可以轮询此方法，或者当执行完本地逻辑后，需要执行分布式任务前检擦此方法。

不过，类似JDK中的CountDownLatch,LeaderLatch也提供了阻塞方法：

1. 方法1

public void await()throws InterruptedException,EOFException

这个方法，会阻塞，直到选主成功。

1. 方法2 为了避免方法1的长时间选主失败

public boolean await(long timeout,TimeUnit unit)throws InterruptedException

这个方法会根据参数中指定的时间，作为等待的期限。到期后，返回选主结果。

对于LeaderLatch实例，无论是否轩主成功，最后都应该调用：

leaderLatch.close();

这样，才会把当前参与者的信息从选主分组中移除出去。如果，当前参与者是主，还会释放主的资格。避免死锁。

4. 错误处理

在实际使用中，必须考虑链接问题引起的主身份丢失问题。 例如：当hasLeadership()返回true，之后链接出问题。 强烈建议：使用LeaderLatch时为其添加一个ConnectionStateListener

LeaderLatch实例会添加一个ConnectionStateListener来监听当前zk链接。 如果，链接不可用（SUSPENDED）则LeaderLatch会认为自己不在是主，等到链接恢复可用时，才可继续。 如果，链接断开（LOST），则LeaderLatch会认为自己不在是主，等到链接重新建立后，删除之前的参与者信息，然后重新参与选主。

5. 源码分析

5.1 LeaderLatch

5.1.1 类定义

先来看看类定义：

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

注意：实现了java.io.Closeable，所以你懂的， try()...catch{}。（3.2中的leaderLatch.close();）

5.1.2 成员变量

public class LeaderLatch implements Closeable
{private final Logger log = LoggerFactory.getLogger(getClass());private final WatcherRemoveCuratorFramework client;private final String latchPath;private final String id;private final AtomicReference<State> state = new AtomicReference<State>(State.LATENT);private final AtomicBoolean hasLeadership = new AtomicBoolean(false);private final AtomicReference<String> ourPath = new AtomicReference<String>();private final ListenerContainer<LeaderLatchListener> listeners = new ListenerContainer<LeaderLatchListener>();private final CloseMode closeMode;private final AtomicReference<Future<?>> startTask = new AtomicReference<Future<?>>();private final ConnectionStateListener listener = new ConnectionStateListener(){@Overridepublic void stateChanged(CuratorFramework client, ConnectionState newState){handleStateChange(newState);}};private static final String LOCK_NAME = "latch-";private static final LockInternalsSorter sorter = new LockInternalsSorter(){@Overridepublic String fixForSorting(String str, String lockName){return StandardLockInternalsDriver.standardFixForSorting(str, lockName);}};public enum State { LATENT, STARTED, CLOSED }public enum CloseMode { SILENT, NOTIFY_LEADER }@VisibleForTestingvolatile CountDownLatch debugResetWaitLatch = null;

• log : caurtor依赖slf4j
• client : zk客户端（curator-framework提供）
• latchPath : 分组路径（zk中的path）
• id : 参与者ID
• state
  • 内部枚举
  • 状态
    • LATENT 休眠
    • STARTED 已启动
    • CLOSED 已关闭
  • 使用AtomicReference原子化包装
• hasLeadership
  • 是否为主
  • 使用AtomicBoolean原子化包装
• ourPath
  • 使用AtomicReference原子化包装
• listeners
  • 一组LeaderLatchListener监听器
• closeMode
  • 内部枚举
  • LeaderLatch关闭方式
    • SILENT : 静默关闭，不触发相关监听器
    • NOTIFY_LEADER ：关闭时触发监听器
• startTask
  • 异步Future
  • 使用AtomicReference原子化包装
• listener
  • 链接状态监听器
  • 参见 ： 4. 错误处理
• LOCK_NAME
  • 私有常量
• sorter
  • 私有常量
  • 用于锁处理时，规范path
  • 对参与者进行排序
• debugResetWaitLatch
  • volatile 可见性
  • reset()使用
  • 在测试时控制启动的时机，防止环境未初始化完成就处理了启动逻辑

注意：

这些成员变量都是`final`类型。
并且，对于引用类型都进行原子化包装，避免并发问题

5.1.3 构造器

提供多个构造器模板，最终都是调用：

    public LeaderLatch(CuratorFramework client, String latchPath){this(client, latchPath, "", CloseMode.SILENT);}public LeaderLatch(CuratorFramework client, String latchPath, String id){this(client, latchPath, id, CloseMode.SILENT);}public LeaderLatch(CuratorFramework client, String latchPath, String id, CloseMode closeMode){this.client = Preconditions.checkNotNull(client, "client cannot be null").newWatcherRemoveCuratorFramework();this.latchPath = PathUtils.validatePath(latchPath);this.id = Preconditions.checkNotNull(id, "id cannot be null");this.closeMode = Preconditions.checkNotNull(closeMode, "closeMode cannot be null");}

可以发现：

1. 默认是采用CloseMode.SILENT方式关闭
2. 默认id是空字符串
3. client、latchPath、id、closeMode不能为空

5.1.4 启动

第3节，介绍过LeaderLatch是由start()启动选主过程：

public void start() throws Exception {Preconditions.checkState(state.compareAndSet(State.LATENT, State.STARTED), "Cannot be started more than once");startTask.set(AfterConnectionEstablished.execute(client, new Runnable(){@Overridepublic void run(){try{internalStart();}finally{startTask.set(null);}}}));
}

可以发现

1. 调用原子性CAS方法，将状态由休眠更新到已启动
2. 执行了一个异步任务来完成启动过程
   1. 使用一个链接可用后回调方式
      • AfterConnectionEstablished.execute()
        • 内部使用了一个ThreadUtils.newSingleThreadExecutor
        • 单线程的线程池
        • 所以本地多个LeaderLatch实例的启动过程是序列化方式执行的
   2. 使用成员变量startTask持有异步Future
   3. 启动完成后会制空startTask
      • 说明启动过程可能会有状态变化
3. 启动的过程实际是由internalStart()方法来完成

private synchronized void internalStart() {if ( state.get() == State.STARTED ){client.getConnectionStateListenable().addListener(listener);try{reset();}catch ( Exception e ){ThreadUtils.checkInterrupted(e);log.error("An error occurred checking resetting leadership.", e);}}
}

1. internalStart()使用synchronized
   • 同步调用
   • 使用this进行互斥锁对象
   • 同一个LeaderLatch对象的多次启动同样序列化执行
     • 即便绕过第2布，也同样可以保证不会重复启动
2. 进行状态判断
   • synchronized内部，再次判断
   • 相当于Double check
3. 在当前连接上注册自带的监听器
4. 调用reset()完成启动逻辑
5. 处理了异常
   • 触发线程中断
     • internalStart()是异步执行，通过中断可以进行更细节的控制
   • 避免粗暴的抛出异常
     • internalStart()是异步执行
     • 避免当前线程意外中断
     • 同时也避免了第2.1步骤中那个单线程的线程池频繁的进行线程开/关所带来的额外开销

@VisibleForTesting
void reset() throws Exception {setLeadership(false);setNode(null);BackgroundCallback callback = new BackgroundCallback(){@Overridepublic void processResult(CuratorFramework client, CuratorEvent event) throws Exception{if ( debugResetWaitLatch != null ){debugResetWaitLatch.await();debugResetWaitLatch = null;}if ( event.getResultCode() == KeeperException.Code.OK.intValue() ){setNode(event.getName());if ( state.get() == State.CLOSED ){setNode(null);}else{getChildren();}}else{log.error("getChildren() failed. rc = " + event.getResultCode());}}};client.create().creatingParentContainersIfNeeded().withProtection().withMode(CreateMode.EPHEMERAL_SEQUENTIAL).inBackground(callback).forPath(ZKPaths.makePath(latchPath, LOCK_NAME), LeaderSelector.getIdBytes(id));
}
private synchronized void setLeadership(boolean newValue)
{boolean oldValue = hasLeadership.getAndSet(newValue);if ( oldValue && !newValue ){ // Lost leadership, was true, now falselisteners.forEach(new Function<LeaderLatchListener, Void>(){@Overridepublic Void apply(LeaderLatchListener listener){listener.notLeader();return null;}});}else if ( !oldValue && newValue ){ // Gained leadership, was false, now truelisteners.forEach(new Function<LeaderLatchListener, Void>(){@Overridepublic Void apply(LeaderLatchListener input){input.isLeader();return null;}});}notifyAll();
}
private void setNode(String newValue) throws Exception{String oldPath = ourPath.getAndSet(newValue);if ( oldPath != null ){client.delete().guaranteed().inBackground().forPath(oldPath);}}

1. reset()的可见范围

   • 利于测试
   • 使用了com.google.common.annotations.VisibleForTesting

2. 初始化选主状态false

   1. getAndSet设置
   2. 根据不同的情况触发不同的监听器
      1. 得到
      2. 失去
   3. notifyAll()
      • 唤醒所有的synchronized等待

3. 制空上次path

   • 如果上一次path有残留，则delete服务器上的信息

4. 在latchPath下创建一个EPHEMERAL_SEQUENTIAL节点

   • 临时顺序节点
   • 并注册了回调
     • 回掉获取latchPath的子节点
     • 并判断自身是否为主

5.1.5 选主

LeaderLatch的选主判断逻辑，是由上一节中第12步中注册的回调方法来触发。 实际由checkLeadership()方法处理：

private void checkLeadership(List<String> children) throws Exception{final String localOurPath = ourPath.get();List<String> sortedChildren = LockInternals.getSortedChildren(LOCK_NAME, sorter, children);int ourIndex = (localOurPath != null) ? sortedChildren.indexOf(ZKPaths.getNodeFromPath(localOurPath)) : -1;if ( ourIndex < 0 ){log.error("Can't find our node. Resetting. Index: " + ourIndex);reset();}else if ( ourIndex == 0 ){setLeadership(true);}else{String watchPath = sortedChildren.get(ourIndex - 1);Watcher watcher = new Watcher(){@Overridepublic void process(WatchedEvent event){if ( (state.get() == State.STARTED) && (event.getType() == Event.EventType.NodeDeleted) && (localOurPath != null) ){try{getChildren();}catch ( Exception ex ){ThreadUtils.checkInterrupted(ex);log.error("An error occurred checking the leadership.", ex);}}}};BackgroundCallback callback = new BackgroundCallback(){@Overridepublic void processResult(CuratorFramework client, CuratorEvent event) throws Exception{if ( event.getResultCode() == KeeperException.Code.NONODE.intValue() ){// previous node is gone - resetreset();}}};// use getData() instead of exists() to avoid leaving unneeded watchers which is a type of resource leakclient.getData().usingWatcher(watcher).inBackground(callback).forPath(ZKPaths.makePath(latchPath, watchPath));}}

当获取到最新的参与者列表后：

1. 对列表进行排序
2. 如果自身处于列表第一位，则当选为主
3. 否则，在latchPath上增加监听/回调
   1. 监听列表中上一位参与者
      • 当上一位参与者退出（节点被删除时）
      • 重新getChildren()再次进行选主
   2. 当latchPath发生变动（如：删除）
      • 调用reset(),重新进行启动过程
        • 即可导致hasLeadership()失效

6. 测试

package com.roc.curator.demo.leader.latchimport org.apache.commons.lang3.RandomStringUtils
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
import org.junit.Before
import org.junit.Test
import java.util.*
import java.util.concurrent.TimeUnit/*** Created by roc on 2017/5/25.*/
class LatchParticipant() {val LATCH_PATH: String = "/test/leader/latch"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()var latch: LeaderLatch = LeaderLatch(client, LATCH_PATH, id)latch.start()println("$id 开始竞选 $time")while(!latch.await(3, TimeUnit.SECONDS)){println("$id 选主失败 ： $time")println("当前主是：${latch.leader.id}")println("参与者：${latch.participants}")}println("$id 选主成功 $time")while (latch.hasLeadership()) {println("$id 执行 $time")TimeUnit.SECONDS.sleep(2)if (Math.random() > 0.89) {break;}}println("$id 结束此轮： $time")latch.close()}
}

zookeeper节点：

get /test/leader/latch/_c_9b313527-e0ed-410f-9510-30e5fd92b5c6-latch-0000000208
zNillKMfuB
cZxid = 0x1db19
ctime = Thu May 25 20:53:22 CST 2017
mZxid = 0x1db19
mtime = Thu May 25 20:53:22 CST 2017
pZxid = 0x1db19
cversion = 0
dataVersion = 0
aclVersion = 0
ephemeralOwner = 0x15156529fae07e9
dataLength = 10
numChildren = 0