SOFAJRaft核心源碼解析 - 高性能Java實(shí)現(xiàn)的Raft算法

2024-12-27 13:55 更新

大家好，我是 V 哥，SOFAJRaft 是螞蟻金服開(kāi)源的一個(gè)基于 Raft 共識(shí)算法的 Java 實(shí)現(xiàn)，它特別適合高負(fù)載、低延遲的分布式系統(tǒng)場(chǎng)景。SOFAJRaft 支持 Multi-Raft-Group，能夠同時(shí)處理多個(gè) Raft 集群，具有擴(kuò)展性和強(qiáng)一致性保障。這個(gè)項(xiàng)目是從百度的 braft 移植而來(lái)的，并且在性能和功能上做了多項(xiàng)優(yōu)化。今天的文章，V 哥來(lái)聊一聊SOFAJRaft的核心源碼實(shí)現(xiàn)。

打開(kāi)全球最大的基友網(wǎng)站 Github，搜索 sofa-jraft，可以找到SOFAJRaft庫(kù)的源碼實(shí)現(xiàn)：

Github

SOFAJRaft 是一個(gè)基于 RAFT 一致性算法的生產(chǎn)級(jí)高性能 Java 實(shí)現(xiàn)，支持 MULTI-RAFT-GROUP，適用于高負(fù)載低延遲的場(chǎng)景。 使用 SOFAJRaft 你可以專注于自己的業(yè)務(wù)領(lǐng)域，由 SOFAJRaft 負(fù)責(zé)處理所有與 RAFT 相關(guān)的技術(shù)難題，并且 SOFAJRaft 非常易于使用，你可以通過(guò)幾個(gè)示例在很短的時(shí)間內(nèi)掌握它。

V哥要介紹的不是基礎(chǔ)應(yīng)用，而是通過(guò)SOFAJRaft庫(kù)的實(shí)現(xiàn)原理，幫助兄弟們來(lái)理解Raft算法。

SOFAJRaft 核心概念

SOFAJRaft 的核心是 Raft 算法，它主要的組件包括：

  • Leader 選舉：用于在集群中選出唯一的 Leader。
  • 日志復(fù)制：Leader 將客戶端的請(qǐng)求日志復(fù)制到所有的 Follower。
  • 日志一致性：通過(guò)多數(shù)派機(jī)制確保集群中的日志是一致的。
  • 日志應(yīng)用：日志經(jīng)過(guò)多數(shù)派確認(rèn)后應(yīng)用到狀態(tài)機(jī)中。

核心源碼分析

1. Raft 節(jié)點(diǎn)啟動(dòng)與初始化

SOFAJRaft 中的 Raft 節(jié)點(diǎn)通過(guò) NodeImpl 類進(jìn)行管理，它是 Raft 節(jié)點(diǎn)的核心實(shí)現(xiàn)。

  1. public class NodeImpl implements Node, Lifecycle<NodeOptions>, Replicator.ReplicatorStateListener, StateMachineCaller.RaftStateMachineListener {
  2.     // Raft 節(jié)點(diǎn)狀態(tài)
  3.     private volatile State state;
  4.     private final RaftGroupId groupId; // Raft group ID
  5.     private final PeerId serverId;     // 當(dāng)前節(jié)點(diǎn) ID
  6.     private final NodeOptions options; // 節(jié)點(diǎn)選項(xiàng)配置
  8.     
  9.     // 構(gòu)造函數(shù)
  10.     public NodeImpl(final String groupId, final PeerId serverId) {
  11.         this.groupId = new RaftGroupId(groupId);
  12.         this.serverId = serverId;
  13.         this.options = new NodeOptions();
  14.     }
  17.     @Override
  18.     public synchronized boolean init(final NodeOptions opts) {
  19.         // 初始化配置
  20.         this.options = opts;
  21.         // 啟動(dòng)選舉定時(shí)器等邏輯
  22.     }
  23. }

在這里，NodeImpl 類的 init 方法用于初始化 Raft 節(jié)點(diǎn)，它會(huì)設(shè)置 Raft 節(jié)點(diǎn)的配置并啟動(dòng)選舉定時(shí)器等機(jī)制。

2. Leader 選舉

Raft 的 Leader 選舉是通過(guò)定時(shí)器和心跳機(jī)制來(lái)實(shí)現(xiàn)的。當(dāng) Follower 沒(méi)有在一段時(shí)間內(nèi)收到 Leader 的心跳時(shí)，它會(huì)進(jìn)入選舉狀態(tài)。

  1. public class ElectionTimer extends Timer {
  2.     private final NodeImpl node;
  5.     public ElectionTimer(NodeImpl node) {
  6.         this.node = node;
  7.     }
  10.     @Override
  11.     public void run() {
  12.         // 處理選舉超時(shí)
  13.         this.node.handleElectionTimeout();
  14.     }
  15. }

當(dāng)定時(shí)器超時(shí)時(shí)，會(huì)觸發(fā) handleElectionTimeout 方法進(jìn)行選舉。

  1. private void handleElectionTimeout() {
  2.     if (this.state != State.FOLLOWER) {
  3.         return;
  4.     }
  5.     // 進(jìn)入候選者狀態(tài)
  6.     becomeCandidate();
  7.     // 發(fā)送投票請(qǐng)求
  8.     sendVoteRequests();
  9. }

這里的邏輯非常清晰了，當(dāng)節(jié)點(diǎn)是 Follower 并且發(fā)生選舉超時(shí)時(shí)，它會(huì)轉(zhuǎn)換為候選者并開(kāi)始發(fā)送投票請(qǐng)求給其他節(jié)點(diǎn)。

3. 日志復(fù)制

在 Raft 中，Leader 負(fù)責(zé)將客戶端的請(qǐng)求日志復(fù)制到 Follower。

  1. public class LeaderState {
  2.     private final NodeImpl node;
  3.     private final LogManager logManager;
  6.     public LeaderState(NodeImpl node) {
  7.         this.node = node;
  8.         this.logManager = node.getLogManager();
  9.     }
  12.     public void replicateLog(final LogEntry logEntry) {
  13.         // 將日志復(fù)制到 Follower 節(jié)點(diǎn)
  14.         for (PeerId peer : node.getReplicatorList()) {
  15.             Replicator replicator = node.getReplicator(peer);
  16.             replicator.sendAppendEntries(logEntry);
  17.         }
  18.     }
  19. }

在這里，Leader 通過(guò) Replicator 將日志復(fù)制到所有 Follower 節(jié)點(diǎn)，sendAppendEntries 方法會(huì)發(fā)送 AppendEntries 請(qǐng)求。

4. 日志一致性

Raft 算法通過(guò)多數(shù)派機(jī)制來(lái)確保日志的一致性，來(lái)看一下源碼：

  1. public class AppendEntriesResponseHandler {
  2.     private final NodeImpl node;
  5.     public void handleResponse(AppendEntriesResponse response) {
  6.         if (response.success) {
  7.             // 更新提交的日志索引
  8.             node.getLogManager().commitIndex(response.index);
  9.         } else {
  10.             // 如果失敗，可能需要重新發(fā)送日志或處理沖突
  11.             node.handleLogReplicationFailure(response);
  12.         }
  13.     }
  14. }

當(dāng)節(jié)點(diǎn)收到 AppendEntriesResponse 時(shí)，如果復(fù)制成功，它會(huì)更新日志的提交索引，確保日志的一致性。

5. 狀態(tài)機(jī)應(yīng)用

一旦日志被提交，Raft 將這些日志應(yīng)用到狀態(tài)機(jī)中，以實(shí)現(xiàn)最終的系統(tǒng)狀態(tài)更新。

  1. public class StateMachineCaller {
  2.     private final StateMachine stateMachine;
  5.     public void onApply(final List<LogEntry> entries) {
  6.         // 將提交的日志應(yīng)用到狀態(tài)機(jī)
  7.         for (LogEntry entry : entries) {
  8.             stateMachine.apply(entry);
  9.         }
  10.     }
  11. }

狀態(tài)機(jī)將處理客戶端請(qǐng)求并更新系統(tǒng)狀態(tài)，這里 apply 方法會(huì)被調(diào)用來(lái)執(zhí)行具體的業(yè)務(wù)邏輯。

我們繼續(xù)深入探討 SOFAJRaft 的其他核心部分，包括**日志管理（Log Management）**、**快照（Snapshot）機(jī)制**和**故障處理**，這些部分在分布式系統(tǒng)中都非常重要，尤其在長(zhǎng)時(shí)間運(yùn)行和高負(fù)載場(chǎng)景下。

6. 日志管理（Log Management）

日志管理是 Raft 協(xié)議中重要的一部分，它保證了每個(gè)節(jié)點(diǎn)在不同時(shí)間點(diǎn)所保存的日志能夠保持一致。SOFAJRaft 使用 LogManager 來(lái)管理日志的存儲(chǔ)和持久化。實(shí)現(xiàn)的代碼是這樣滴：

  1. public class LogManager {
  2.     private final List<LogEntry> logEntries; // 日志條目列表
  3.     private long commitIndex;  // 當(dāng)前提交的日志索引
  4.     private long lastApplied;  // 最后應(yīng)用的日志索引
  7.     public LogManager() {
  8.         this.logEntries = new ArrayList<>();
  9.     }
  12.     public synchronized void appendEntry(LogEntry entry) {
  13.         // 將新日志添加到日志列表
  14.         logEntries.add(entry);
  15.     }
  18.     public synchronized void commitIndex(long newCommitIndex) {
  19.         // 更新提交索引，保證提交的日志能在狀態(tài)機(jī)中被應(yīng)用
  20.         this.commitIndex = newCommitIndex;
  21.     }
  24.     public synchronized List<LogEntry> getUnappliedEntries() {
  25.         // 獲取尚未應(yīng)用到狀態(tài)機(jī)的日志
  26.         return logEntries.subList((int) lastApplied + 1, (int) commitIndex + 1);
  27.     }
  30.     public void applyLogsToStateMachine(StateMachine stateMachine) {
  31.         List<LogEntry> unappliedEntries = getUnappliedEntries();
  32.         for (LogEntry entry : unappliedEntries) {
  33.             stateMachine.apply(entry); // 應(yīng)用日志到狀態(tài)機(jī)
  34.             lastApplied++;
  35.         }
  36.     }
  37. }

在日志管理中，LogManager 負(fù)責(zé)維護(hù) Raft 節(jié)點(diǎn)的所有日志條目，并根據(jù)多數(shù)派的確認(rèn)來(lái)更新提交的日志索引。當(dāng)提交的日志多于 commitIndex 時(shí)，這些日志可以應(yīng)用到狀態(tài)機(jī)中。applyLogsToStateMachine 方法則負(fù)責(zé)將日志條目應(yīng)用到狀態(tài)機(jī)。

7. 快照機(jī)制（Snapshot）

在長(zhǎng)時(shí)間運(yùn)行的集群中，如果僅僅依賴日志復(fù)制，日志可能會(huì)積累得非常龐大，影響性能和磁盤(pán)空間的使用。那要腫么辦呢？因此，Raft 設(shè)計(jì)了快照（Snapshot）機(jī)制來(lái)定期將當(dāng)前狀態(tài)持久化，并丟棄已經(jīng)持久化的日志。

  1. public class SnapshotManager {
  2.     private final StateMachine stateMachine;
  3.     private final LogManager logManager;
  4.     private long lastSnapshotIndex;
  7.     public SnapshotManager(StateMachine stateMachine, LogManager logManager) {
  8.         this.stateMachine = stateMachine;
  9.         this.logManager = logManager;
  10.     }
  13.     public void takeSnapshot() {
  14.         // 生成新的快照
  15.         Snapshot snapshot = stateMachine.saveSnapshot();
  16.         this.lastSnapshotIndex = logManager.getLastAppliedIndex();
  17.         // 持久化快照到磁盤(pán)
  18.         persistSnapshot(snapshot);
  19.         // 清理舊的日志條目
  20.         logManager.truncatePrefix(lastSnapshotIndex);
  21.     }
  24.     private void persistSnapshot(Snapshot snapshot) {
  25.         // 將快照寫(xiě)入磁盤(pán)的實(shí)現(xiàn)邏輯
  26.         // 如將 snapshot 對(duì)象序列化并寫(xiě)入文件系統(tǒng)
  27.     }
  28. }

在 SnapshotManager 中，takeSnapshot 方法會(huì)觸發(fā)狀態(tài)機(jī)生成當(dāng)前的快照，并持久化到磁盤(pán)。當(dāng)快照創(chuàng)建完成后，舊的日志條目可以被截?cái)嘁葬尫糯鎯?chǔ)空間。這極大地減少了日志的冗余，提高了系統(tǒng)的性能。

8. 故障處理與恢復(fù)

SOFAJRaft 具有健全的故障處理機(jī)制，能夠處理節(jié)點(diǎn)的崩潰和網(wǎng)絡(luò)分區(qū)等情況。Raft 協(xié)議通過(guò)日志復(fù)制和 Leader 選舉機(jī)制來(lái)保證系統(tǒng)的容錯(cuò)性。

Follower 的故障恢復(fù)

當(dāng) Follower 恢復(fù)之后，會(huì)向 Leader 請(qǐng)求缺失的日志，Leader 會(huì)通過(guò) InstallSnapshot 或者 AppendEntries 來(lái)將最新的日志發(fā)送給 Follower。

  1. public class FollowerRecovery {
  2.     private final NodeImpl node;
  3.     private final LogManager logManager;
  6.     public FollowerRecovery(NodeImpl node) {
  7.         this.node = node;
  8.         this.logManager = node.getLogManager();
  9.     }
  12.     public void handleInstallSnapshot(InstallSnapshotRequest request) {
  13.         // 收到 Leader 的快照安裝請(qǐng)求
  14.         Snapshot snapshot = request.getSnapshot();
  15.         node.getStateMachine().loadSnapshot(snapshot);
  16.         logManager.reset(snapshot.getLastIndex());
  17.     }
  20.     public void handleAppendEntries(AppendEntriesRequest request) {
  21.         // 收到 Leader 的日志復(fù)制請(qǐng)求
  22.         List<LogEntry> entries = request.getEntries();
  23.         logManager.appendEntries(entries);
  24.     }
  25. }

handleInstallSnapshot 用于處理 Leader 發(fā)送的快照請(qǐng)求，當(dāng)日志缺失過(guò)多時(shí)，Leader 會(huì)將整個(gè)快照發(fā)給 Follower，避免重復(fù)發(fā)送大量的日志。handleAppendEntries 則用于正常情況下的日志復(fù)制和恢復(fù)。

Leader 的故障恢復(fù)

Leader 故障后，集群會(huì)通過(guò)新的 Leader 選舉恢復(fù)正常工作。Leader 選舉過(guò)程在前面的部分已經(jīng)詳細(xì)介紹，當(dāng)一個(gè)新的 Leader 被選出后，它會(huì)嘗試將自己的日志與 Follower 同步。

  1. public class LeaderRecovery {
  2.     private final NodeImpl node;
  3.     private final LogManager logManager;
  6.     public LeaderRecovery(NodeImpl node) {
  7.         this.node = node;
  8.         this.logManager = node.getLogManager();
  9.     }
  12.     public void catchUpFollowers() {
  13.         // 向所有 Follower 發(fā)送最新的日志條目
  14.         for (PeerId peer : node.getReplicatorList()) {
  15.             Replicator replicator = node.getReplicator(peer);
  16.             replicator.sendAppendEntries(logManager.getUncommittedEntries());
  17.         }
  18.     }
  19. }

新的 Leader 會(huì)調(diào)用 catchUpFollowers 來(lái)確保所有的 Follower 都與它保持一致，利用 Raft 的日志復(fù)制機(jī)制恢復(fù)一致性。

9. Multi-Raft-Group 的支持

SOFAJRaft 的一大特色是對(duì) Multi-Raft-Group 的支持，也就是說(shuō)，它能夠管理多個(gè)獨(dú)立的 Raft 集群。這使得它在一些需要分片或者不同業(yè)務(wù)隔離的場(chǎng)景中能夠很好地應(yīng)用。

  1. public class MultiRaftGroupManager {
  2.     private final Map<String, NodeImpl> raftGroups = new ConcurrentHashMap<>();
  5.     public NodeImpl createRaftGroup(String groupId, PeerId serverId, NodeOptions options) {
  6.         NodeImpl node = new NodeImpl(groupId, serverId);
  7.         node.init(options);
  8.         raftGroups.put(groupId, node);
  9.         return node;
  10.     }
  13.     public NodeImpl getRaftGroup(String groupId) {
  14.         return raftGroups.get(groupId);
  15.     }
  16. }

MultiRaftGroupManager 負(fù)責(zé)管理多個(gè) Raft 集群，通過(guò) createRaftGroup 方法可以創(chuàng)建新的 Raft 集群，每個(gè)集群都有自己的 NodeImpl 實(shí)例。這種架構(gòu)設(shè)計(jì)讓系統(tǒng)可以同時(shí)運(yùn)行多個(gè) Raft 實(shí)例，從而大幅提升擴(kuò)展性。

總結(jié)

SOFAJRaft 基于 Raft 算法實(shí)現(xiàn)了一個(gè)高性能、支持 Multi-Raft-Group 的分布式一致性系統(tǒng)。它通過(guò) NodeImpl 負(fù)責(zé) Raft 節(jié)點(diǎn)的管理，通過(guò) Leader 選舉、日志復(fù)制、多數(shù)派機(jī)制等實(shí)現(xiàn)分布式系統(tǒng)中的強(qiáng)一致性。

關(guān)鍵代碼展示了從節(jié)點(diǎn)初始化到日志復(fù)制和一致性維護(hù)的核心流程，這些是 Raft 算法的重要組成部分。

SOFAJRaft 的設(shè)計(jì)通過(guò)日志管理、快照機(jī)制、故障處理以及 Multi-Raft-Group 的支持，提供了一個(gè)健壯且高效的分布式一致性解決方案。通過(guò)對(duì)關(guān)鍵代碼的分析，我們可以看到它在處理日志復(fù)制、一致性維護(hù)和快照生成上的精妙實(shí)現(xiàn)，能夠有效應(yīng)對(duì)高負(fù)載、長(zhǎng)時(shí)間運(yùn)行的分布式系統(tǒng)場(chǎng)景。

好了，整理的學(xué)習(xí)筆記就到這里，分享給大家，希望可以幫助你更加深入的理解 Raft 算法，V 哥在這里求個(gè)關(guān)注和點(diǎn)贊，感謝感謝。

以上內(nèi)容是否對(duì)您有幫助：
在線筆記
App下載
App下載

掃描二維碼

下載編程獅App

公眾號(hào)
微信公眾號(hào)

編程獅公眾號(hào)