【转载】http://www.voidcn.com/article/p-ornzyvyy-bqr.html

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创建好之后，必须执行：

leaderLatch.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 = new AtomicReference(State.LATENT);private final AtomicBoolean autoRequeue = new AtomicBoolean(false);private final AtomicReference> ourTask = new AtomicReference>(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 : 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){@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 task = executorService.submit(new Callable(){@Overridepublic Void call() throws Exception{try{doWorkLoop();}finally{clearIsQueued();if ( autoRequeue.get() ){internalRequeue();}}return null;}});ourTask.set(task);return true;}return false;
}

可以发现：

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

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

1. 调用doWorkLoop()
2. 通过finally
   1. 调用clearIsQueued();
      • synchronized
      • isQueued = false
   2. 如果开启自动重新排队，则再次调用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;}}

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

继续看看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}}}}

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

1. CAS操作，将状态从已启动更新为已关闭
2. 清除掉链接上的监听
3. 关闭线程池
4. 制空异步任务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();}}}

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

5.7 中断

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

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

1. synchronized
2. 就是拿到线程池中正在执行的任务的Future
3. 调用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

本文来自互联网用户投稿，文章观点仅代表作者本人，不代表本站立场，不承担相关法律责任。如若转载，请注明出处。 如若内容造成侵权/违法违规/事实不符，请点击【内容举报】进行投诉反馈！