基于Apache Curator框架的两种分布式Leader选举策略详解

2,722 阅读15分钟
原文链接: www.zifangsky.cn

在分布式环境中,一个应用通常都会部署在多个服务器节点上。如果这些应用节点的运行模式是一主多从或者多主多从,这时就需要用到Leader选举策略,从多个节点中选举出Master节点。另外,当某个Master节点意外宕机,这时也需要用到Leader选举策略从它的多个Slave节点中选举出新的Master节点

对于Leader选举策略,Apache Curator框架提供了两种策略,开发者可以根据实际需求具体选择

(1)添加依赖:

首先需要在pom.xml中添加如下依赖:

        <!-- Apache Curator -->
        <dependency>
            <groupId>org.apache.curator</groupId>
            <artifactId>curator-framework</artifactId>
            <version>4.0.0</version>
        </dependency>
        <dependency>
            <groupId>org.apache.curator</groupId>
            <artifactId>curator-recipes</artifactId>
            <version>4.0.0</version>
        </dependency>
        <dependency>
            <groupId>org.apache.curator</groupId>
            <artifactId>curator-x-discovery</artifactId>
            <version>4.0.0</version>
        </dependency>
        <dependency>
            <groupId>org.apache.curator</groupId>
            <artifactId>curator-test</artifactId>
            <version>4.0.0</version>
            <scope>test</scope>
        </dependency>

(2)Leader Latch:

Apache Curator框架提供的第一种Leader选举策略是Leader Latch。这种选举策略,其核心思想是初始化多个LeaderLatch,然后在等待几秒钟后,Curator会自动从中选举出Leader。示例代码如下:

package cn.zifangsky.kafkademo.zookeeper;
 
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;
 
import org.apache.curator.RetryPolicy;
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.apache.curator.utils.CloseableUtils;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
 
/**
 * 测试Apache Curator框架的两种选举方案
 * @author zifangsky
 *
 */
public class TestLeaderLatch {
    //会话超时时间
    private final int SESSION_TIMEOUT = 30 * 1000;
    
    //连接超时时间
    private final int CONNECTION_TIMEOUT = 3 * 1000;
    
    //客户端数量
    private final int CLIENT_NUMBER = 10;
    
    //ZooKeeper服务地址
    private static final String SERVER = "192.168.1.159:2100,192.168.1.159:2101,192.168.1.159:2102";
    
    private final String PATH = "/curator/latchPath";
    
    //创建连接实例
    private CuratorFramework client = null;
    
    /**
     * baseSleepTimeMs:初始的重试等待时间
     * maxRetries:最多重试次数
     */
    RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
 
    //LeaderLatch实例集合
    List<LeaderLatch> leaderLatchList = new ArrayList<LeaderLatch>(CLIENT_NUMBER);
    
    /**
     * 初始化
     * @throws Exception
     */
    @Before
    public void init() throws Exception{
        //创建 CuratorFrameworkImpl实例
        client = CuratorFrameworkFactory.newClient(SERVER, SESSION_TIMEOUT, CONNECTION_TIMEOUT, retryPolicy);
        client.start();
        
        for(int i=0;i<CLIENT_NUMBER;i++){
            //创建LeaderLatch实例
            LeaderLatch leaderLatch = new LeaderLatch(client, PATH, "Client #" + i);
            leaderLatchList.add(leaderLatch);
            leaderLatch.start();
        }
        
        //等待Leader选举完成
        TimeUnit.SECONDS.sleep(5);
        System.out.println("**********LeaderLatch初始化完成**********");
    }
    
    /**
     * 测试获取当前选举出来的leader,以及手动尝试获取领导权
     * @throws Exception 
     */
    @Test
    public void testCheckLeader() throws Exception{
        LeaderLatch currentLeader = null;
        
        for(LeaderLatch tmp : leaderLatchList){
            if(tmp.hasLeadership()){ //判断是否是leader
                currentLeader = tmp;
                break;
            }
        }        
        
        System.out.println("当前leader是: " + currentLeader.getId());
//        System.out.println("当前leader是: " + leaderLatchList.get(0).getLeader().getId());
        
        /**
         * 从List中移除当前主节点,并从剩下的节点中继续选举leader
         */
        currentLeader.close(); //关闭当前主节点
        leaderLatchList.remove(currentLeader); //从List中移除
        TimeUnit.SECONDS.sleep(5); //等待再次选举
 
        //再次获取当前leader
        for(LeaderLatch tmp : leaderLatchList){
            if(tmp.hasLeadership()){
                currentLeader = tmp;
                break;
            }
        }
        
        System.out.println("新leader是: " + currentLeader.getId());
        currentLeader.close(); //关闭当前主节点
        leaderLatchList.remove(currentLeader); //从List中移除
        
        LeaderLatch firstNode = leaderLatchList.get(0); //获取此时第一个节点
        System.out.println("删除leader后,当前第一个节点: " + firstNode.getId());
        
        firstNode.await(10, TimeUnit.SECONDS); //阻塞并尝试获取领导权,可能失败
        
        //再次获取当前leader
        for(LeaderLatch tmp : leaderLatchList){
            if(tmp.hasLeadership()){
                currentLeader = tmp;
                break;
            }
        }
        
        System.out.println("最终实际leader是: " + currentLeader.getId());
        
    }
 
    /**
     * 测试完毕关闭连接
     */
    @After
    public void close(){
        for(LeaderLatch tmp : leaderLatchList){
            CloseableUtils.closeQuietly(tmp);
        }
        
        CloseableUtils.closeQuietly(client);
    }
 
}

关于上述代码,有以下几点需要简单说明:

i)在初始化LeaderLatch的时候,因为这里只是简单测试,因此直接在一个for循环里完成了整个初始化过程。然而在实际的分布式环境中,每个LeaderLatch的初始化过程应该在每个应用节点内部完成(PS:可以使用多个单元测试模拟)

ii)LeaderLatch的await() 方法的含义是阻塞当前线程,直到当前LeaderLatch实例获取领导权。当然有可能当前LeaderLatch实例直到等待时间结束也没有获取领导权,原因可能是:其他线程在某一时刻中断此线程、当前LeaderLatch实例在某一时刻被关闭、其他某个LeaderLatch实例一直没有释放领导权等等

iii)Leader Latch选举的本质是连接ZooKeeper,然后在“/curator/latchPath”路径为每个LeaderLatch创建临时有序节点:

在创建临时节点时,org.apache.curator.framework.recipes.leader.LeaderLatch 的 checkLeadership(List<String> children) 方法会将选举路径(/curator/latchPath)下面的所有节点按照序列号排序,如果当前节点的序列号最小,则将该节点设置为leader。反之则监听比当前节点序列号小一位的节点的状态(PS:因为每次都会选择序列号最小的节点为leader,所以在比当前节点序列号小一位的节点未被删除前,当前节点是不可能变成leader的)。如果监听的节点被删除,则会触发重新选举方法——reset()

注:上图使用的工具是ZK UI,具体可以参考我之前的这篇文章:www.zifangsky.cn/1126.html

上面示例代码输出如下:

**********LeaderLatch初始化完成**********
当前leader是: Client #0
新leader是: Client #8
删除leader后,当前第一个节点: Client #1
最终实际leader是: Client #9

上面的输出结果很好理解,Client #0的序列号最小(0000000120),其次是Client #8(0000000121)、Client #9(0000000122)……

思考:如果上面测试代码的init()方法是如下示例,最终选举出来的leader顺序是什么样的,为什么?

    /**
     * 初始化
     * @throws Exception
     */
    @Before
    public void init() throws Exception{
        //创建 CuratorFrameworkImpl实例
        client = CuratorFrameworkFactory.newClient(SERVER, SESSION_TIMEOUT, CONNECTION_TIMEOUT, retryPolicy);
        client.start();
        
        for(int i=0;i<CLIENT_NUMBER;i++){
            //创建LeaderLatch实例
            LeaderLatch leaderLatch = new LeaderLatch(client, PATH, "Client #" + i);
            leaderLatchList.add(leaderLatch);
            leaderLatch.start();
            
            TimeUnit.SECONDS.sleep(5);
        }
        
        //等待Leader选举完成
        TimeUnit.SECONDS.sleep(5);
        System.out.println("**********LeaderLatch初始化完成**********");
    }

iv)Leader Latch选举策略在选举出leader后,该LeaderLatch实例会一直占有领导权,直到调用 close() 方法关闭当前主节点,然后其他LeaderLatch实例才会再次选举leader。这种策略适合主备应用,当主节点意外宕机之后,多个从节点会自动选举其中一个为新的主节点(Master节点)

(3)Leader Election:

Apache Curator框架提供的另一种Leader选举策略是Leader Election。这种选举策略跟Leader Latch选举策略不同之处在于每个实例都能公平获取领导权,而且当获取领导权的实例在释放领导权之后,该实例还有机会再次获取领导权。另外,选举出来的leader不会一直占有领导权,当 takeLeadership(CuratorFramework client) 方法执行结束之后会自动释放领导权。示例代码如下:

i)继承LeaderSelectorListenerAdapter,用于定义获取领导权后的业务逻辑:

package cn.zifangsky.kafkademo.zookeeper;
 
import java.io.Closeable;
import java.io.IOException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
 
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.leader.LeaderSelector;
import org.apache.curator.framework.recipes.leader.LeaderSelectorListenerAdapter;
 
public class CustomLeaderSelectorListenerAdapter extends
        LeaderSelectorListenerAdapter implements Closeable {
 
    private String name;
    private LeaderSelector leaderSelector;
    public AtomicInteger leaderCount = new AtomicInteger();
 
    public CustomLeaderSelectorListenerAdapter(CuratorFramework client,String path,String name
            ) {
        this.name = name;
        this.leaderSelector = new LeaderSelector(client, path, this);
 
        /**
         * 自动重新排队
         * 该方法的调用可以确保此实例在释放领导权后还可能获得领导权
         */
        leaderSelector.autoRequeue();
    }
 
    public void start() throws IOException {
        leaderSelector.start();
    }
    
    @Override
    public void close() throws IOException {
        leaderSelector.close();
    }
    
    /**
     * 获取领导权
     */
    @Override
    public void takeLeadership(CuratorFramework client) throws Exception {
        final int waitSeconds = 2;
        System.out.println(name + "成为当前leader");
        System.out.println(name + " 之前成为leader的次数:" + leaderCount.getAndIncrement() + "次");
 
        //TODO 其他业务代码
        try{
            //等待2秒后放弃领导权(模拟业务执行过程)
            Thread.sleep(TimeUnit.SECONDS.toMillis(waitSeconds));
        }catch ( InterruptedException e ){
            System.err.println(name + "已被中断");
            Thread.currentThread().interrupt();
        }finally{
            System.out.println(name + "放弃领导权\n");
        }
        
    }
 
}
 

ii)测试:

package cn.zifangsky.kafkademo.zookeeper;
 
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;
 
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.curator.utils.CloseableUtils;
import org.junit.After;
import org.junit.Test;
 
/**
 * 测试Apache Curator框架的两种选举方案
 * @author zifangsky
 *
 */
public class TestLeaderElection {
    //会话超时时间
    private final int SESSION_TIMEOUT = 30 * 1000;
    
    //连接超时时间
    private final int CONNECTION_TIMEOUT = 3 * 1000;
    
    //客户端数量
    private final int CLIENT_NUMBER = 10;
    
    //ZooKeeper服务地址
    private static final String SERVER = "192.168.1.159:2100,192.168.1.159:2101,192.168.1.159:2102";
    
    private final String PATH = "/curator/latchPath";
    
    //创建连接实例
    private CuratorFramework client = null;
    
    /**
     * baseSleepTimeMs:初始的重试等待时间
     * maxRetries:最多重试次数
     */
    RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
 
    //自定义LeaderSelectorListenerAdapter实例集合
    List<CustomLeaderSelectorListenerAdapter> leaderSelectorListenerList
        = new ArrayList<CustomLeaderSelectorListenerAdapter>();
    
    @Test
    public void test() throws Exception{
        //创建 CuratorFrameworkImpl实例
        client = CuratorFrameworkFactory.newClient(SERVER, SESSION_TIMEOUT, CONNECTION_TIMEOUT, retryPolicy);
        client.start();
        
        for(int i=0;i<CLIENT_NUMBER;i++){        
            //创建LeaderSelectorListenerAdapter实例
            CustomLeaderSelectorListenerAdapter leaderSelectorListener = 
                    new CustomLeaderSelectorListenerAdapter(client, PATH, "Client #" + i);
            
            leaderSelectorListener.start();
            leaderSelectorListenerList.add(leaderSelectorListener);            
        }
        
        //暂停当前线程,防止单元测试结束,可以让leader选举过程持续进行
        TimeUnit.SECONDS.sleep(600);
    }
    
    /**
     * 测试完毕关闭连接
     */
    @After
    public void close(){
        for(CustomLeaderSelectorListenerAdapter tmp : leaderSelectorListenerList){
            CloseableUtils.closeQuietly(tmp);
        }
        
        CloseableUtils.closeQuietly(client);
    }    
    
}

以上代码需要注意的是:

  • 上面只是简单测试,为了使模拟过程更加真实,可以在多个单元测试中实例化并测试选举过程
  • 每个实例在获取领导权后,如果 takeLeadership(CuratorFramework client) 方法执行结束,将会释放其领导权

上面输出示例如下:


Client #1成为当前leader
Client #1 之前成为leader的次数:0次
Client #1放弃领导权

Client #4成为当前leader
Client #4 之前成为leader的次数:0次
Client #4放弃领导权

Client #6成为当前leader
Client #6 之前成为leader的次数:0次
Client #6放弃领导权

Client #7成为当前leader
Client #7 之前成为leader的次数:0次
Client #7放弃领导权

Client #5成为当前leader
Client #5 之前成为leader的次数:0次
Client #5放弃领导权

Client #9成为当前leader
Client #9 之前成为leader的次数:0次
Client #9放弃领导权

Client #3成为当前leader
Client #3 之前成为leader的次数:0次
Client #3放弃领导权

Client #2成为当前leader
Client #2 之前成为leader的次数:0次
Client #2放弃领导权

Client #8成为当前leader
Client #8 之前成为leader的次数:0次
Client #8放弃领导权

Client #0成为当前leader
Client #0 之前成为leader的次数:2次
Client #0放弃领导权

Client #1成为当前leader
Client #1 之前成为leader的次数:1次
Client #1放弃领导权

Client #4成为当前leader
Client #4 之前成为leader的次数:1次
Client #4放弃领导权

Client #6成为当前leader
Client #6 之前成为leader的次数:1次
Client #6放弃领导权

Client #7成为当前leader
Client #7 之前成为leader的次数:1次
Client #7放弃领导权

Client #5成为当前leader
Client #5 之前成为leader的次数:1次
Client #5放弃领导权

Client #9成为当前leader
Client #9 之前成为leader的次数:1次
Client #9放弃领导权

Client #3成为当前leader
Client #3 之前成为leader的次数:1次
Client #3放弃领导权

Client #2成为当前leader
Client #2 之前成为leader的次数:1次
Client #2放弃领导权

Client #8成为当前leader
Client #8 之前成为leader的次数:1次
Client #8放弃领导权

Client #0成为当前leader
Client #0 之前成为leader的次数:3次
Client #0放弃领导权

参考: