Raft算法的Java实现

自己这几天在看Redis的Sentinel高可用解决方案，Sentinel选主使用的是著名的Raft一致性算法，本文对Raft的选主作了介绍，具体的算法内容，请参考 Raft 论文

Raft的整体结构

Raft 通过选举一个高贵的领导人，然后给予他全部的管理复制日志的责任来实现一致性。

而每个 server 都可能会在 3 个身份之间切换：

• 领导者
• 候选者
• 跟随者

而影响他们身份变化的则是 选举。
当所有服务器初始化的时候，都是 跟随者，这个时候需要一个 领导者，所有人都变成 候选者，直到有人成功当选 领导者。

角色轮换如下图：

而领导者也有宕机的时候，宕机后引发新的 选举，所以，整个集群在选举和正常运行之间切换，具体如下图：

从上图可以看出，选举和正常运行之间切换，但请注意， 上图中的 term 3 有一个地方，后面没有跟着 正常运行 阶段，为什么呢?

答：当一次选举失败（比如正巧每个人都投了自己），就执行一次 加时赛，每个 Server 会在一个随机的时间里重新投票，这样就能保证不冲突了。所以，当 term 3 选举失败，等了几十毫秒，执行 term 4 选举，并成功选举出领导人。

接着，领导者周期性的向所有跟随者发送心跳包来维持自己的权威。如果一个跟随者在一段时间里没有接收到任何消息，也就是选举超时，那么他就会认为系统中没有可用的领导者,并且发起选举以选出新的领导者。

要开始一次选举过程，跟随者先要增加自己的当前任期号并且转换到候选人状态。然后请求其他服务器为自己投票。那么会产生 3 种结果：

a. 自己成功当选
b. 其他的服务器成为领导者
c. 僵住，没有任何一个人成为领导者

注意：

1. 每一个 server 最多在一个任期内投出一张选票（有任期号约束），先到先得。
2. 要求最多只能有一个人赢得选票。
3. 一旦成功，立即成为领导人，然后广播所有服务器停止投票阻止新得领导产生。

僵住怎么办？ Raft 通过使用随机选举超时时间（例如 150 - 300 毫秒）的方法将服务器打散投票。每个候选人在僵住的时候会随机从一个时间开始重新选举。

以上，就是 Raft 算法对选主的介绍，非常简短，具体的还是要看一边论文才能搞清楚，直接看代码基本看不懂。

实现代码

我在网上看到的都是golang语言的实现，作为一个Java程序员想找到一份Java的实现真难啊，蚂蚁金服开源了他们的JRaft实现，可是毕竟实现的非常完美，代码就有点难看懂了。

启动类就是做一下对自身节点的ip port的包装，

public class RaftNodeBootStrap {
   

    public static void main(String[] args) throws Throwable {
   
        main0();
    }

    public static void main0() throws Throwable {
   
        String[] peerAddr = {
   "localhost:8775","localhost:8776","localhost:8777", "localhost:8778", "localhost:8779"};

        NodeConfig config = new NodeConfig();

        // 自身节点
        config.setSelfPort(Integer.valueOf(System.getProperty("serverPort")));

        // 其他节点地址
        config.setPeerAddrs(Arrays.asList(peerAddr));

        Node node = DefaultNode.getInstance();
        node.setConfig(config);

        node.init();

        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
   
            try {
   
                node.destroy();
            } catch (Throwable throwable) {
   
                throwable.printStackTrace();
            }
        }));
    }    

重点在node.setConfig() 和init()两个方法

 @Override
    public void setConfig(NodeConfig config) {
   
        this.config = config;
        stateMachine = DefaultStateMachine.getInstance();
        logModule = DefaultLogModule.getInstance();

        peerSet = PeerSet.getInstance();
        for (String s : config.getPeerAddrs()) {
   
            Peer peer = new Peer(s);
            peerSet.addPeer(peer);
            if (s.equals("localhost:" + config.getSelfPort())) {
   
                peerSet.setSelf(peer);
            }
        }

        RPC_SERVER = new DefaultRpcServer(config.selfPort, this);
    }

init对自己的同伴节点做初始化，最后一行，开启了一个RPCServer，RPCServer的功能看过Raft论文的应该都清楚，接收投票RPCVoteRequest, 和附加日志 RPCLogAppendRequest （心跳也是日志附加Request，只是日志内容为null）

public void init() throws Throwable {
   
        if (started) {
   
            return;
        }
        synchronized (this) {
   
            if (started) {
   
                return;
            }
            RPC_SERVER.start();

            consensus = new DefaultConsensus(this);
            delegate = new ClusterMembershipChangesImpl(this);

            RaftThreadPool.scheduleWithFixedDelay(heartBeatTask, 500);
            RaftThreadPool.scheduleAtFixedRate(electionTask, 6000, 500);
            RaftThreadPool.execute(replicationFailQueueConsumer);

            LogEntry logEntry = logModule.getLast();
            if (logEntry != n