zookeeper：master选举

• 模拟选举机器类

package com.karat.cn.zookeeperAchieveLock.zkclient;

import java.io.Serializable;

/**
 * 选举的机器
 */
public class UserCenter implements Serializable{

    private static final long serialVersionUID = -1776114173857775665L;
    private int id; //机器信息

    private String name;//机器名称

    public int getId() {
        return id;
    }

    public void setId(int id) {
        this.id = id;
    }

    public String getName() {
        return name;
    }

    public void setName(String name) {
        this.name = name;
    }

    @Override
    public String toString() {
        return "UserCenter [id=" + id + ", name=" + name + "]";
    }
 
}

• 选举服务

package com.karat.cn.zookeeperAchieveLock.zkclient;

import org.I0Itec.zkclient.IZkDataListener;
import org.I0Itec.zkclient.ZkClient;
import org.I0Itec.zkclient.exception.ZkNodeExistsException;

import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/**
 * 选举的服务
 */
public class MasterSelector {

    private ZkClient zkClient;

    private final static String MASTER_PATH="/master"; //需要争抢的节点

    private IZkDataListener dataListener; //注册节点内容变化

    private UserCenter server;  //其他服务器

    private UserCenter master;  //master节点

    private boolean isRunning=false;

    ScheduledExecutorService scheduledExecutorService= Executors.newScheduledThreadPool(1);//定时任务

    public MasterSelector(UserCenter server,ZkClient zkClient) {
        System.out.println("["+server+"] 去争抢master权限");
        this.server = server;
        this.zkClient=zkClient;

        this.dataListener= new IZkDataListener() {
            @Override
            public void handleDataChange(String s, Object o) throws Exception {

            }

            @Override
            public void handleDataDeleted(String s) throws Exception {
                //节点如果被删除, 发起选主操作
                chooseMaster();
            }
        };
    }

    public void start(){
        //开始选举
        if(!isRunning){
            isRunning=true;
            zkClient.subscribeDataChanges(MASTER_PATH,dataListener); //注册节点事件
            chooseMaster();
        }
    }


    public void stop(){
        //停止
        if(isRunning){
            isRunning=false;
            scheduledExecutorService.shutdown();
            zkClient.unsubscribeDataChanges(MASTER_PATH,dataListener);
            releaseMaster();
        }
    }


    //具体选master的实现逻辑
    private void chooseMaster(){
        if(!isRunning){
            System.out.println("当前服务没有启动");
            return ;
        }
        try {
            zkClient.createEphemeral(MASTER_PATH, server);//创建一个临时节点
            master=server; //把server节点赋值给master
            System.out.println(master+"->我现在已经是master，你们要听我的");

            //定时器
            //master释放(master 出现故障）,没2秒钟释放一次
            scheduledExecutorService.schedule(()->{
                releaseMaster();//释放锁
            },2, TimeUnit.SECONDS);
        }catch (ZkNodeExistsException e){
            //创建一个临时节点抛出异常
            //表示master已经存在
            UserCenter userCenter=zkClient.readData(MASTER_PATH,true);
            if(userCenter==null) {
                System.out.println("启动操作：");
                chooseMaster(); //再次获取master
            }else{
                master=userCenter;
            }
        }
    }

    private void releaseMaster(){
        //释放锁(故障模拟过程)
        //判断当前是不是master，只有master才需要释放
        if(checkIsMaster()){
            zkClient.delete(MASTER_PATH); //删除
        }
    }


    private boolean checkIsMaster(){
        //判断当前的server是不是master
        UserCenter userCenter=zkClient.readData(MASTER_PATH);
        if(userCenter.getName().equals(server.getName())){
            master=userCenter;
            return true;
        }
        return false;
    }

}

• 选举测试

package com.karat.cn.zookeeperAchieveLock.zkclient;

import org.I0Itec.zkclient.ZkClient;
import org.I0Itec.zkclient.serialize.SerializableSerializer;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;

/**
 * master选举测试
 */
public class MasterChooseTest {

    private final static String CONNECTSTRING="47.107.121.215:2181";


    public static void main(String[] args) throws IOException {
        List<MasterSelector> selectorLists=new ArrayList<>();
        try {
            for(int i=0;i<10;i++) {
                ZkClient zkClient = new ZkClient(CONNECTSTRING, 5000,
                        5000,
                        new SerializableSerializer());
                UserCenter userCenter = new UserCenter();
                userCenter.setId(i);
                userCenter.setName("客户端：" + i);

                MasterSelector selector = new MasterSelector(userCenter,zkClient);
                selectorLists.add(selector);
                selector.start();//触发选举操作
                TimeUnit.SECONDS.sleep(1);
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            for(MasterSelector selector:selectorLists){
                selector.stop();
            }
        }
    }
}

通过zookeeper进行master选举，就是利用zookeeper的节点特性，通过是否能够创建临时节点来判断是否选举成功，如果不能创建临时节点，则表明已有线程创建成功，那么创建成功的线程就为选举的master，当网络发生故障或其它问题导致该线程挂掉，那么zookeeper中的该临时节点也会删除，通过监听节点是否有过删除动作，重新选举新的master。

• 有关leader选举文章：http://www.cnblogs.com/leesf456/p/6107600.html#undefined