Raft 算法详解#

public class RaftNode {
    // 持久化状态（必须写入磁盘）
    private long currentTerm = 0;           // 当前任期
    private String votedFor = null;         // 当前任期投票给的候选人
    private List<LogEntry> log = new ArrayList<>(); // 日志条目

    // 节点状态
    private NodeState state = NodeState.FOLLOWER;
    private String nodeId;

    // 选举相关计时器
    private long electionTimeout;          // 选举超时时间
    private long lastHeartbeat = 0;         // 上次收到心跳的时间

    // 投票规则
    public boolean canVoteFor(String candidateId, long candidateTerm, int candidateLogLength) {
        // 规则1：任期必须 >= 当前任期
        if (candidateTerm < currentTerm) return false;

        // 规则2：当前任期还没投过票，或投给了同一个候选人
        if (currentTerm > votedFor && votedFor != null) return false;

        // 规则3：候选人日志必须比本地日志「更新」
        // 「更新」的定义：term 更大，或 term 相同但 index 更大
        LogEntry lastLocalLog = log.get(log.size() - 1);
        LogEntry candidateLastLog = new LogEntry(candidateLogLength, 0, null);

        return candidateLastLog.isNewerThan(lastLocalLog);
    }
}

public enum NodeState {
    FOLLOWER,   // 跟随者：被动接收心跳
    CANDIDATE,  // 候选人：发起选举
    LEADER      // 领导者：处理所有客户端请求
}

// 选举超时逻辑
public void onHeartbeatTimeout() {
    state = NodeState.CANDIDATE;
    currentTerm++;                          // 新任期
    votedFor = nodeId;                      // 给自己投票
    electionTimeout = RANDOM_RANGE(150, 300); // 随机超时

    // 向所有节点发送 RequestVote
    for (String peer : peers) {
        sendRequestVote(peer, currentTerm, nodeId, getLastLogInfo());
    }
}

public class RaftNode {
    public AppendEntriesResponse handleAppendEntries(AppendEntriesRequest req) {
        // 一致性检查：prevLogIndex 之前的日志必须匹配
        if (req.prevLogIndex > 0 &&
            (log.size() < req.prevLogIndex ||
             log.get(req.prevLogIndex - 1).term != req.prevLogTerm)) {
            // 不匹配：拒绝并返回本地日志长度
            return new AppendEntriesResponse(false, log.size());
        }

        // 匹配：追加新日志
        for (LogEntry entry : req.entries) {
            appendLog(entry);
        }

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

        return new AppendEntriesResponse(true, log.size());
    }
}