文档章节

[Curator] Leader Election 的使用与分析

秋雨霏霏
 秋雨霏霏
发布于 2017/05/26 19:46
字数 2595
阅读 679
收藏 1

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
{
    @Override
    public void stateChanged(CuratorFramework client, ConnectionState newState)
    {
        if ( (newState == ConnectionState.SUSPENDED) || (newState == ConnectionState.LOST) )
        {
            throw new CancelLeadershipException();
        }
    }
}

  • 抽象类
  • 带有ConnectionStateListenerstateChanged默认实现
  • 参见: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的回调
  • clientleaderPathlistener不能为空
  • 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;
}

可以发现:

  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()

@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
                }
            }
        }
    }
  1. 初始hasLeadership=false
  2. 申请对leaderPath加锁,阻塞
  3. 加锁成功,则当选为主hasLeadership = true
  4. 回调org.apache.curator.framework.recipes.leader.LeaderSelectorListenertakeLeadership方法
    • 此回调是在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;
        }

        @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();
            }
        }
    }
  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 = 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

© 著作权归作者所有

秋雨霏霏
粉丝 148
博文 94
码字总数 168411
作品 0
杭州
CTO(技术副总裁)
私信 提问
Apache Curator 2.8.0 发布,简化 zookeeper 客户端

Apache Curator 2.8.0 发布,此版本的 Apache Curator Java 库让 Apache Zookeeper 使用起来更方便,详细更新信息如下: Bug 修复: [CURATOR-154] - PersistentEphemeralNode May Fail to ...

oschina
2015/05/12
3.2K
2
Apache Curator 2.4.2 发布

Apache Curator 就是为了简化 zookeeper客户端调用而生,利用它,可以更好的使用zookeeper。 Apache Curator 2.4.2 发布了,改进内容包括: Bug [CURATOR-74] - Documentation on InterProce...

oschina
2014/04/25
1K
0
聊聊curator recipes的LeaderLatch

序 本文主要研究一下curator recipes的LeaderLatch 实例 zkCli查询 LeaderLatch.start curator-recipes-4.0.1-sources.jar!/org/apache/curator/framework/recipes/leader/LeaderLatch.java ......

go4it
2018/10/11
0
0
使用Curator实现的zookeeper分布式锁出现的Unimplemented for {root.path}

问题描述 Curator使用 ZooKeeper 作为分布式锁,启动时发生该异常。 Curator 客户端版本:curator-recipes-2.10.0 ZooKeeper 服务器版本:3.4.13 异常日志 以及 问题分析 UnimplementedExcep...

loubobooo
02/27
0
0
Apache Curator 2.7.0 发布

Apache Curator 2.7.0 发布,此版本现已提供下载,二进制包提供在 Maven 库。更新内容请看这里: Bug 修复 [CURATOR-42] - Background guaranteed delete considers NoNode to be a failed ...

oschina
2014/11/04
2.5K
0

没有更多内容

加载失败,请刷新页面

加载更多

Jenkins系列_插件安装及报错处理

进入Jenkins之后我们可以进行插件的安装,插件管理位于以下模块: 发现上面报了一堆错误,是因为插件的依赖没有安装好,那么这一节,就先把这些错误解决掉吧。解决完成后,也就基本会使用插件...

shzwork
今天
2
0
mysql mysql的所有查询语句和聚合函数(整理一下,忘记了可以随时看看)

查询所有字段 select * from 表名; 查询自定字段 select 字段名 from 表名; 查询指定数据 select * from 表名 where 条件; 带关键字IN的查询 select * from 表名 where 条件 [not] in(元素...

edison_kwok
昨天
9
0
解决多线程并行加载缓存问题(利用guava实现)

依赖 com.google.guava:guava:20.0 import com.google.common.cache.Cache;import com.google.common.cache.CacheBuilder;import java.util.concurrent.ExecutionException;import j......

暗中观察
昨天
3
0
利用VisualVM 内存查看

准备工作,建几个测试类。等下就是要查看这几个类里面的属性 package visualvm;public class MultiObject { private String str; private int i; MultiObject(String str...

冷基
昨天
3
0
组装一台工作游戏两用机

一、配置清单如下: 分类 项目 价格(元) 主板 华硕(ASUS)TUF Z370-PLUS GAMING II 电竞特工 Z370二代 支持9代CPU 1049 CPU 英特尔(Intel) i7 8700K 酷睿六核 盒装CPU处理器 2640 风扇 九...

mbzhong
昨天
4
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部