每日算法 — 使用java实现Raft算法：分布式共识与领导者选举

概述

Raft是斯坦福大学的Diego Ongaro和John Ousterhout于2014年提出的分布式共识算法，旨在替代复杂的Paxos算法，同时提供等价的安全性和容错能力。Raft将共识问题分解为三个相对独立的子问题：领导者选举、日志复制和安全性。本文将用Java实现一个完整的Raft节点，帮助读者理解分布式系统中的一致性原理。

核心概念

Raft集群中的每个节点处于以下三种状态之一：

• Follower（跟随者）：被动响应来自Leader和Candidate的请求
• Candidate（候选者）：发起选举，争取成为Leader
• Leader（领导者）：处理所有客户端请求，复制日志到Follower

任期（Term）

Raft将时间划分为任意长度的任期，每个任期以一次选举开始。任期号是单调递增的整数，用于检测过期的领导者和选举。

Java实现

节点状态与数据结构

import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;

public class RaftNode {
    // 节点状态枚举
    public enum State { FOLLOWER, CANDIDATE, LEADER }

    private final int nodeId;
    private final List<String> peerAddresses;
    private final AtomicReference<State> state = new AtomicReference<>(State.FOLLOWER);

    // 持久化状态（需要持久化到磁盘）
    private final AtomicLong currentTerm = new AtomicLong(0);  // 当前任期
    private volatile Integer votedFor = null;  // 当前任期投票给谁
    private final List<LogEntry> log = new CopyOnWriteArrayList<>();  // 日志条目

    // 易失状态
    private volatile int commitIndex = 0;  // 已知已提交的最高日志索引
    private volatile int lastApplied = 0;  // 已应用到状态机的最高日志索引

    // Leader易失状态（仅在Leader时有效）
    private final Map<Integer, Integer> nextIndex = new ConcurrentHashMap<>();   // 每个Follower的下一条日志索引
    private final Map<Integer, Integer> matchIndex = new ConcurrentHashMap<>();  // 每个Follower已复制的最高日志索引

    // 定时器与线程
    private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(3);
    private volatile ScheduledFuture<?> electionTimer;
    private volatile ScheduledFuture<?> heartbeatTimer;

    // 随机选举超时：150-300ms
    private final int electionTimeoutBase = 150;
    private final Random random = new Random();

    //  RPC客户端（模拟）
    private final RaftRpcClient rpcClient;

    public RaftNode(int nodeId, List<String> peers, RaftRpcClient rpcClient) {
        this.nodeId = nodeId;
        this.peerAddresses = peers;
        this.rpcClient = rpcClient;

        // 初始化Leader状态
        for (int i = 0; i < peers.size(); i++) {
            if (i != nodeId) {
                nextIndex.put(i, 1);
                matchIndex.put(i, 0);
            }
        }

        // 启动选举定时器
        resetElectionTimer();
    }

日志条目结构

    /**
     * 日志条目：每条日志包含任期号和操作命令
     */
    public static class LogEntry {
        public final long term;
        public final int index;
        public final String command;

        public LogEntry(long term, int index, String command) {
            this.term = term;
            this.index = index;
            this.command = command;
        }

        @Override
        public String toString() {
            return String.format("LogEntry{term=%d, index=%d, cmd='%s'}", term, index, command);
        }
    }

领导者选举

    /**
     * 重置选举定时器：Follower在每次收到有效RPC时调用
     */
    private void resetElectionTimer() {
        if (electionTimer != null) {
            electionTimer.cancel(false);
        }

        int timeout = electionTimeoutBase + random.nextInt(150);
        electionTimer = scheduler.schedule(this::startElection, timeout, TimeUnit.MILLISECONDS);
    }

    /**
     * 发起选举：转换为Candidate状态，增加任期，向所有节点请求投票
     */
    private synchronized void startElection() {
        if (state.get() == State.LEADER) return;

        state.set(State.CANDIDATE);
        currentTerm.incrementAndGet();
        votedFor = nodeId;
        int votesGranted = 1;  // 自己投自己

        System.out.println("Node " + nodeId + " 开始选举，任期=" + currentTerm.get());

        // 向所有其他节点发送RequestVote RPC
        for (int i = 0; i < peerAddresses.size(); i++) {
            if (i == nodeId) continue;

            final int peerId = i;
            CompletableFuture.runAsync(() -> {
                RequestVoteRequest req = new RequestVoteRequest(
                    currentTerm.get(),
                    nodeId,
                    getLastLogIndex(),
                    getLastLogTerm()
                );

                try {
                    RequestVoteResponse resp = rpcClient.requestVote(peerId, req);
                    handleVoteResponse(resp);
                } catch (Exception e) {
                    // RPC失败，不计入投票
                }
            });
        }

        // 选举超时内统计票数
        scheduler.schedule(() -> {
            if (state.get() == State.CANDIDATE && votesGranted > peerAddresses.size() / 2) {
                becomeLeader();
            }
        }, electionTimeoutBase, TimeUnit.MILLISECONDS);
    }

    /**
     * 处理投票响应
     */
    private synchronized void handleVoteResponse(RequestVoteResponse resp) {
        if (state.get() != State.CANDIDATE) return;

        if (resp.term > currentTerm.get()) {
            // 发现更高任期，退回Follower
            currentTerm.set(resp.term);
            state.set(State.FOLLOWER);
            votedFor = null;
            resetElectionTimer();
            return;
        }

        if (resp.voteGranted) {
            // 收到赞成票，统计逻辑应在发起选举时异步处理
        }
    }

    /**
     * 处理收到的投票请求
     */
    public synchronized RequestVoteResponse handleRequestVote(RequestVoteRequest req) {
        if (req.term < currentTerm.get()) {
            return new RequestVoteResponse(currentTerm.get(), false);
        }

        if (req.term > currentTerm.get()) {
            currentTerm.set(req.term);
            state.set(State.FOLLOWER);
            votedFor = null;
        }

        // 检查日志是否至少一样新
        boolean logOk = req.lastLogTerm > getLastLogTerm() ||
            (req.lastLogTerm == getLastLogTerm() && req.lastLogIndex >= getLastLogIndex());

        if ((votedFor == null || votedFor == req.candidateId) && logOk) {
            votedFor = req.candidateId;
            resetElectionTimer();
            return new RequestVoteResponse(currentTerm.get(), true);
        }

        return new RequestVoteResponse(currentTerm.get(), false);
    }

    /**
     * 成为Leader：初始化Leader状态，开始发送心跳
     */
    private synchronized void becomeLeader() {
        if (state.compareAndSet(State.CANDIDATE, State.LEADER)) {
            System.out.println("Node " + nodeId + " 成为Leader，任期=" + currentTerm.get());

            // 初始化Leader状态
            for (int i = 0; i < peerAddresses.size(); i++) {
                if (i != nodeId) {
                    nextIndex.put(i, log.size() + 1);
                    matchIndex.put(i, 0);
                }
            }

            // 取消选举定时器，启动心跳定时器
            if (electionTimer != null) {
                electionTimer.cancel(false);
            }

            // 立即发送心跳并建立周期性心跳
            sendHeartbeats();
            heartbeatTimer = scheduler.scheduleAtFixedRate(
                this::sendHeartbeats, 50, 50, TimeUnit.MILLISECONDS);
        }
    }

日志复制与心跳

    /**
     * 发送心跳（空AppendEntries）或日志复制
     */
    private void sendHeartbeats() {
        if (state.get() != State.LEADER) return;

        for (int i = 0; i < peerAddresses.size(); i++) {
            if (i == nodeId) continue;

            final int peerId = i;
            CompletableFuture.runAsync(() -> {
                int nextIdx = nextIndex.getOrDefault(peerId, 1);
                int prevLogIndex = nextIdx - 1;
                long prevLogTerm = prevLogIndex > 0 ? log.get(prevLogIndex - 1).term : 0;

                // 获取待发送的日志条目
                List<LogEntry> entries = new ArrayList<>();
                for (int j = nextIdx - 1; j < log.size(); j++) {
                    entries.add(log.get(j));
                }

                AppendEntriesRequest req = new AppendEntriesRequest(
                    currentTerm.get(),
                    nodeId,
                    prevLogIndex,
                    prevLogTerm,
                    entries,
                    commitIndex
                );

                try {
                    AppendEntriesResponse resp = rpcClient.appendEntries(peerId, req);
                    handleAppendEntriesResponse(peerId, resp, nextIdx);
                } catch (Exception e) {
                    // 网络故障，下次重试
                }
            });
        }
    }

    /**
     * 处理日志复制响应
     */
    private synchronized void handleAppendEntriesResponse(int peerId, 
            AppendEntriesResponse resp, int sentNextIndex) {
        if (state.get() != State.LEADER) return;

        if (resp.term > currentTerm.get()) {
            currentTerm.set(resp.term);
            state.set(State.FOLLOWER);
            votedFor = null;
            if (heartbeatTimer != null) heartbeatTimer.cancel(false);
            resetElectionTimer();
            return;
        }

        if (resp.success) {
            // 更新匹配索引
            matchIndex.put(peerId, sentNextIndex + log.size() - sentNextIndex);
            nextIndex.put(peerId, matchIndex.get(peerId) + 1);

            // 检查是否可以提交
            tryAdvanceCommitIndex();
        } else {
            // 日志不匹配，递减nextIndex重试
            nextIndex.put(peerId, Math.max(1, nextIndex.get(peerId) - 1));
        }
    }

    /**
     * 处理收到的AppendEntries请求
     */
    public synchronized AppendEntriesResponse handleAppendEntries(AppendEntriesRequest req) {
        if (req.term < currentTerm.get()) {
            return new AppendEntriesResponse(currentTerm.get(), false);
        }

        // 收到Leader心跳，重置选举定时器
        resetElectionTimer();

        if (req.term > currentTerm.get()) {
            currentTerm.set(req.term);
            state.set(State.FOLLOWER);
            votedFor = null;
        }

        if (state.get() == State.CANDIDATE) {
            state.set(State.FOLLOWER);
        }

        // 日志一致性检查
        if (req.prevLogIndex > 0) {
            if (req.prevLogIndex > log.size()) {
                return new AppendEntriesResponse(currentTerm.get(), false);
            }
            if (log.get(req.prevLogIndex - 1).term != req.prevLogTerm) {
                return new AppendEntriesResponse(currentTerm.get(), false);
            }
        }

        // 追加新日志条目
        int insertIndex = req.prevLogIndex;
        for (LogEntry entry : req.entries) {
            if (insertIndex < log.size()) {
                if (log.get(insertIndex).term != entry.term) {
                    // 删除冲突条目及之后的所有条目
                    while (log.size() > insertIndex) {
                        log.remove(log.size() - 1);
                    }
                    log.add(entry);
                }
            } else {
                log.add(entry);
            }
            insertIndex++;
        }

        // 更新commitIndex
        if (req.leaderCommit > commitIndex) {
            commitIndex = Math.min(req.leaderCommit, log.size());
        }

        return new AppendEntriesResponse(currentTerm.get(), true);
    }

    /**
     * 尝试推进commitIndex：找到满足多数派复制的最高索引
     */
    private synchronized void tryAdvanceCommitIndex() {
        for (int n = log.size(); n > commitIndex; n--) {
            if (log.get(n - 1).term != currentTerm.get()) break;

            int matchCount = 1;  // Leader自己
            for (int i = 0; i < peerAddresses.size(); i++) {
                if (i != nodeId && matchIndex.getOrDefault(i, 0) >= n) {
                    matchCount++;
                }
            }

            if (matchCount > peerAddresses.size() / 2) {
                commitIndex = n;
                break;
            }
        }
    }

客户端请求处理

    /**
     * 客户端向Leader提交命令
     */
    public synchronized boolean submitCommand(String command) {
        if (state.get() != State.LEADER) {
            // 重定向到Leader
            return false;
        }

        LogEntry entry = new LogEntry(currentTerm.get(), log.size() + 1, command);
        log.add(entry);

        // 立即触发日志复制
        sendHeartbeats();

        return true;
    }

    private int getLastLogIndex() {
        return log.size();
    }

    private long getLastLogTerm() {
        return log.isEmpty() ? 0 : log.get(log.size() - 1).term;
    }

RPC数据结构

    // RequestVote请求
    public static class RequestVoteRequest {
        public final long term;
        public final int candidateId;
        public final int lastLogIndex;
        public final long lastLogTerm;

        public RequestVoteRequest(long term, int candidateId, int lastLogIndex, long lastLogTerm) {
            this.term = term; this.candidateId = candidateId;
            this.lastLogIndex = lastLogIndex; this.lastLogTerm = lastLogTerm;
        }
    }

    public static class RequestVoteResponse {
        public final long term;
        public final boolean voteGranted;

        public RequestVoteResponse(long term, boolean voteGranted) {
            this.term = term; this.voteGranted = voteGranted;
        }
    }

    // AppendEntries请求
    public static class AppendEntriesRequest {
        public final long term;
        public final int leaderId;
        public final int prevLogIndex;
        public final long prevLogTerm;
        public final List<LogEntry> entries;
        public final int leaderCommit;

        public AppendEntriesRequest(long term, int leaderId, int prevLogIndex,
                                   long prevLogTerm, List<LogEntry> entries, int leaderCommit) {
            this.term = term; this.leaderId = leaderId;
            this.prevLogIndex = prevLogIndex; this.prevLogTerm = prevLogTerm;
            this.entries = entries; this.leaderCommit = leaderCommit;
        }
    }

    public static class AppendEntriesResponse {
        public final long term;
        public final boolean success;

        public AppendEntriesResponse(long term, boolean success) {
            this.term = term; this.success = success;
        }
    }

    // RPC接口（模拟）
    public interface RaftRpcClient {
        RequestVoteResponse requestVote(int peerId, RequestVoteRequest req);
        AppendEntriesResponse appendEntries(int peerId, AppendEntriesRequest req);
    }
}

算法分析

安全性保证

Raft通过以下机制保证安全性：

1. 选举限制：Candidate的日志必须至少和多数派一样新，才能当选
2. 提交规则：Leader只能提交当前任期的日志条目
3. 日志匹配：如果两条日志在相同索引处具有相同任期，则它们之前的所有条目都相同

容错能力

时间复杂度

总结

Raft通过将复杂的共识问题分解为领导者选举和日志复制两个子问题，大大降低了理解和实现的难度。本文实现的Raft核心包含完整的任期管理、投票逻辑、日志复制和提交机制。在实际生产环境中，还需考虑日志压缩（Snapshot）、成员变更（Membership Change）和性能优化等扩展功能。