raft一致性协议

Raft 协议raft协议是基于TCP的选举机制时间 日志 版本核心三要素时间 (随机超时)Follower 都有一个选举超时时间例如 150ms ~ 300ms 的随机值。作用防止多个 Follower 同时变成 Candidate 导致选票分裂死锁。随机化能保证大概率只有一个节点先超时发起选举。版本 (Term/任期)每次选举Candidate 都会将自己的 Term 1。作用类似于“版本号”。如果旧 Leader 恢复后发现自己 Term 落后会立即退位成 Follower如果发现别人的 Term 比自己小会拒绝投票。日志 (Log Completeness)这是最关键的限制。Follower 在投票时会对比自己和 Candidate 的日志新鲜度LastLogIndex 和 LastLogTerm。规则只有当 Candidate 的日志至少和自己一样新时Follower 才会投赞成票。这保证了选出来的 Leader 一定包含所有已提交的日志避免数据丢失。作用leader新注册信息进来后偏移量1配置得半同步模式3/5从节点收到后拿到偏移量对比自己没有先存日志在更新自身数据到一致性剩下得2/5走异Leader: 我同时给4个Follower都发了 谁先回复算谁的 凑够3个(含自己)就算达成一致 剩下的慢慢来2/5异步其实是不管了并不算异步拿到我就算成功只是不关心ack返回后续根据心跳持续跟进日志不一致问题第一道防线只有“日志最全”的人才能当 Leader你可能会担心“如果 Follower A 手里有一条 Leader 没有的日志比如刚写入还没来得及同步结果来了一个新的 Leader B把 A 的这条日志覆盖掉了那这条日志是不是就丢了”Raft 在选举阶段就堵死了这条路。选举限制规则当一个候选人Candidate发起投票请求时它会带上自己最后一条日志的任期号和索引。投票者Follower会进行比对“如果你的日志比我的旧我坚决不投给你”这意味着什么这意味着最终当选的 Leader一定拥有集群中大多数节点里最新、最全的日志。既然新 Leader 拥有最新的日志那么那些被它覆盖掉的 Follower 上的日志一定是过时且未达成共识的脏数据。第二道防线“已提交”与“未提交”的区别我们需要区分两种状态的日志已提交的日志 (Committed)定义已经被复制到超过半数节点的日志。安全性一旦日志被提交它就永远不会丢失。原因因为已经有一半以上的节点有了这条日志根据选举规则必须获得半数票未来的任何一任 Leader 都必须从这“大多数”节点中产生。既然大多数都有这条日志选出来的 Leader 肯定也有。所以Leader 绝不可能覆盖掉已提交的日志。未提交的日志 (Uncommitted)定义只在 Leader 本地存在或者只同步给了少数几个节点的日志。安全性可能会丢失。场景旧 Leader 写入了一条日志还没来得及复制给大多数人自己就挂了。新 Leader 选出它的日志可能不包含这条旧日志。旧 Leader 恢复后变成 Follower发现新 Leader 的日志跟自己对不上于是接受新 Leader 的覆盖。结论这部分数据虽然在旧 Leader 上“写过”但因为没来得及形成共识没达到多数派对于外部世界来说相当于从来没发生过。客户端收到超时或错误通常会重试写入从而保证最终一致性。举个例子假设集群有 5 个节点S1, S2, S3, S4, S5。S1 是 Leader写入日志X。S1 成功把X复制给了 S2。此时 S1, S2 有X其他没有。S1 还没来得及提交Commit就挂了。选举开始S1 想竞选但它发现 S3、S4、S5 的日志都比它新或者至少不比它旧而且 S3/S4/S5 互投选出了S3 为新 Leader。注意S1 因为只有自己和 S2 有日志共2票凑不够 3 票所以它当不上 Leader。S3 成为 Leader开始处理新请求写入了日志Y。S3 把Y复制给 S4, S5。S1 重启回归变成了 Follower。S3 告诉 S1“把你索引 1 的位置清空换成我的日志Y”。结果日志X被覆盖了看起来像是“丢”了。但是对于客户端来说步骤 2 之后并没有收到“写入成功”的响应因为只有凑够多数派 Leader 才会返回成功。客户端会认为写入失败并重试。当客户端重试时会把数据发给新 Leader S3数据最终还是会被保存下来。Raft 牺牲了部分未确认数据的可用性换取了强一致性。只要客户端收到了“成功”响应数据就绝对不会丢。凡是丢掉的日志都是客户端还没收到“成功”响应的。其实Raft 协议采用的是“多数派同步完成”的策略可选择全同步完成、主节点完成、多数派同步完成(推荐这样数据丢失问题会降低同时效率也会增加)日志存储日志存储建议使用TIKV或者其他模式推荐使用TIKVTiKV 的“文件系统”简介raft/目录Raft DB这里面存的是Raft 日志。文件形式主要是.log文件和MANIFEST等元数据文件。作用这就是我们之前聊的 WAL预写日志。为了保证写入速度它是顺序追加写的。kv/目录KV DB这里面存的是真正的业务数据。文件形式一堆以.sst结尾的文件比如000123.sst,000456.sst。作用这就是你说的“持久化文件”。数据最终就固化在这些文件里。TiKV 并没有使用外部的数据库如 MySQL 或 PostgreSQL而是内嵌了一个叫RocksDB的库由 Facebook 开源的 C 库。你可以把RocksDB理解为一个极其强大的“文件管理器”。TiKV 调用 RocksDB 的接口来读写磁盘上的那些文件。RocksDB 是怎么把数据变成文件的LSM-Tree 架构这解释了为什么你觉得它是“文件”而不是传统“数据库”内存阶段 (MemTable)当你写入数据时数据先在内存里跳来跳去跳表结构这时候还没落盘速度极快。刷盘阶段 (Flush - SST)当内存里的数据攒够了比如 128MBRocksDB 会把这块数据一次性打包生成一个新的.sst文件写到磁盘上。注意这里不是修改旧文件而是直接生成新文件。合并阶段 (Compaction)磁盘上的.sst文件越来越多怎么办后台线程会在空闲时把几个小文件读取出来合并成一个大文件然后删除旧文件。一个标准的 SST 文件以 LevelDB/RocksDB 为例在磁盘上的排列顺序大致如下数据块序列 (Data Blocks)这是文件的主体占据了绝大部分空间。里面存的就是你真正的 Key-Value 数据。它们是一个接一个紧密排列的。元数据块 (Meta Blocks) [可选]比如布隆过滤器Bloom Filter用来快速判断“这个数据是否存在”避免无效读取。元索引块 (Meta-Index Block)这是一个特殊的索引专门用来记录上面的“元数据块”在哪里。一级索引块 (Index Block)这就是你问的“索引区”。它记录了前面所有“数据块”的起始位置和对应的最大 Key。页脚 (Footer)——关键点在这里这是文件的最后 48 字节固定大小。因为大小固定程序只需要从文件末尾往前倒数 48 个字节就能立刻读到它读取顺序第一步读 Footer最后 48 字节程序直接读取文件末尾的这 48 个字节。Footer 里只存了两个核心信息一级索引块的位置偏移量 长度。元索引块的位置偏移量 长度。为什么要这么设计如果把索引位置放在文件头每次写入新数据导致索引变大时就需要移动整个文件的数据效率太低。放在文件尾写完数据追加写入即可无需移动旧数据。第二步读 Index Block索引区根据 Footer 提供的地址程序跳转到文件的对应位置把Index Block加载到内存中。此时内存里就有了一张“地图”知道了数据区被切分成了哪些块以及每个块的范围。第三步定位 Data Block数据区当你要查数据时先在内存的 Index Block 里比对找到对应的数据块偏移量然后直接去读取那个具体的Data Block。其实就是磁盘光标移动在 Java 的标准 IO 体系里java.io.RandomAccessFile确实是唯一一个直接提供seek(long pos)方法让你能自由地将读写指针“瞬移”到文件任意位置的类。举个例子假设这 10 条数据的 Key 分别是Key_001到Key_010。由于数据量可能比较大或者为了演示结构我们假设前 4 条数据(Key_001-Key_004) 填满了第 1 个数据块 (Block A)。中间 4 条数据(Key_005-Key_008) 填满了第 2 个数据块 (Block B)。最后 2 条数据(Key_009-Key_010) 填入了第 3 个数据块 (Block C)。RocksDB 不会写一条记一条索引而是像“写书”一样写正文Data Blocks先把Key_001到Key_010依次写入磁盘。此时磁盘上已经有了这三块数据。生成目录Index Block等数据都写完了RocksDB 回头看看刚才写了啥在内存里生成一张表索引块“如果你找Key_001到Key_004去偏移量0的位置找。”“如果你找Key_005到Key_008去偏移量4096的位置找。”“如果你找Key_009到Key_010去偏移量8192的位置找。”把这个索引块追加写到数据后面比如在偏移量 12288 处。写封底Footer最后写入那个固定的48 字节 Footer。Footer是什么当你打开文件时你只需要问操作系统“这个文件一共多大”假设文件总共 13000 字节。你直接计算13000 - 48 12952。于是你知道Footer 就在 12952 这个位置。读取 Footer 后你会发现里面只存了两个关键数字简化版索引块在哪里比如偏移量 12288索引块有多大比如600 字节当你有新的数据要写入时TiKV (RocksDB) 会怎么做它不会去动旧的那个 SST 文件。它会创建一个全新的、编号更大的 SST 文件比如000005.sst。把新数据在这个新文件里重新排版、写数据块、写索引、写 Footer。结果就是你的磁盘上会同时存在000004.sst旧的、000005.sst新的、000006.sst更新的……它们互不干扰各自独立。生成SST文件规则SST 文件是一次性打包生成的不存在“写到一半断电导致索引和数据对不上”的情况。它的生成过程是在内存里完成的流程非常严谨第一步在内存里攒数据程序先在内存里开辟一块区域叫MemTable或BlockBuilder把数据一条条放进去。这时候数据块的内容确定了对应的索引位置也在内存里算好了。第二步一次性刷盘等这一批数据都准备好了程序会调用一次系统写入命令把数据块 索引块 Footer作为一个整体一口气写到磁盘上的一个新文件里。第三步原子性保证只有当整个文件完整写入并成功关闭后这个文件才会被标记为“有效”。如果写到一半断电了怎么办操作系统会发现这个文件没写完大小不对或者没有合法的 Footer下次启动时RocksDB 会直接把这个“残缺”的文件当成垃圾删掉。所以只要这个文件能被打开就说明它的数据块、索引块、Footer 绝对是完美匹配的既然有那么多独立的 SST 文件001.sst,002.sst...我要查一个 Key怎么知道去哪个文件找这就轮到MANIFEST 文件也就是“元数据文件”出场了。SST 文件负责存具体的业务数据书的内容。MANIFEST 文件负责记录“现在磁盘上有哪些有效的 SST 文件每个文件的 Key 范围是多少”。如果要读取全部数据TiKV 不会傻乎乎地去遍历磁盘上的每一个文件。它有一个版本管理器基于 MANIFEST 文件。内存里维护着一张最新的文件清单。这张清单明确记录了当前有哪些 SST 文件是有效的比如001.sst,005.sst,008.sst。那些已经被合并、删除的旧文件比如002.sst,003.sst清单里直接没有它们扫描时会自动忽略。所以扫描过程是查清单 - 拿到有效文件列表 - 依次打开顺序读取。如果你只是普通的查询比如GET key_100或者是范围查询SCAN key_100 TO key_200完全不需要扫描所有文件点查询通过索引直接定位到某一个文件的某一个块读完就走。范围查询计算出这个范围的数据可能落在哪几个文件里只读这几个文件。Metadata文件作用每个文件都有“身份证”File Metadata文件名最小 Key最大 Key状态001.sstKey_001Key_050有效002.sstKey_051Key_100有效003.sstKey_101Key_150有效当你想找Key_100时TiKV 根本不会去读文件内容它只需要在内存里对比这些范围第一步看001.sst的范围 (001-050)。Key_100比最大值050还大肯定不在里面。跳过第二步看002.sst的范围 (051-100)。Key_100刚好在这个范围内等于最大值。命中第三步既然锁定了002.sst就不需要再看003.sst了。这个过程是在内存里进行的速度极快纳秒级哪怕你有几万个文件通过二分查找也能瞬间定位MANIFEST 文件长什么样如果你去 TiKV 的数据目录比如kv/db下看你会看到类似这样的文件MANIFEST-000001CURRENT1.MANIFEST-xxxxxx这就是真正的元数据文件。它的文件名后面带一串数字版本号。存什么它记录了从数据库启动到现在所有的历史变更操作。比如“添加了文件 A”、“删除了文件 B”、“合并了文件 C 和 D 生成文件 E”。特点它是追加写的。每次有新的 SST 文件生成或者旧的被清理RocksDB 都会往这个文件里追加一条记录。2.CURRENT这是一个非常小的文本文件。存什么里面只有一行字写着当前最新的MANIFEST文件名。作用因为 MANIFEST 文件会不断变大RocksDB 有时会生成新的 MANIFEST 文件比如做 Compaction 整理时。CURRENT文件就是告诉程序“别找错了最新的账本是这一个”工作原理当你重启 TiKV 时流程是这样的读CURRENT先看看最新的 MANIFEST 是哪个。读MANIFEST把这个巨大的“账本”从头读到尾。读到第一行“创建了001.sst(范围 1-50)” - 内存里记下001.sst。读到第十行“创建了002.sst(范围 51-100)” - 内存里记下002.sst。读到第五十行“删除了001.sst” - 内存里把001.sst划掉。构建内存地图读完最后一行内存里就得到了一张最新、最全的 SST 文件清单VersionSet。文件功能总结1. “总账本”MANIFEST 文件文件名示例MANIFEST-000001作用这就是我们刚才聊的 Metadata。它记录了所有的 SST 文件清单以及它们的 Key 范围。特点它是追加写的。数据库每次启动都要读这个文件来重建内存里的地图。2. “索引卡片”CURRENT 文件文件名示例CURRENT作用这是一个极小的文本文件。因为 MANIFEST 文件可能会变多比如旧的被归档了这个文件里只写了一行字告诉程序“当前最新的 MANIFEST 是哪一个”。地位它是整个数据库的入口点。3. “书架上的书”SST 文件文件名示例000005.sst,000006.sst...作用这才是真正存你业务数据的地方。数量这是目录里数量最多的文件。随着数据写入这个列表会越来越长。4. “临时记事本”WAL 日志 (Write Ahead Log)文件名示例000010.log作用我们在最开始提到的 WAL。当数据还在内存表MemTable里没刷盘生成 SST 时为了防止断电丢失先写在这里。一旦对应的数据生成了 SST 文件这个日志就可以被删除了。5. “门锁”LOCK 文件文件名示例LOCK作用这是一个空文件但非常重要。当你启动一个 TiKV 实例时它会尝试给这个文件加锁。如果加锁成功说明你是唯一的主人。如果加锁失败比如你已经开了一个进程又手滑开了一个程序会直接报错退出。这是为了防止两个进程同时修改同一个数据库导致数据打架。文件类型角色比喻数量核心功能SST书籍很多 (成百上千)存真正的数据不可变。MANIFEST总目录通常 1 个 (偶尔多个)记录哪些 SST 是有效的。CURRENT目录指引1 个指向最新的 MANIFEST。LOG草稿纸几个保护还没变成 SST 的新数据。LOCK门卫1 个防止重复启动。简单代码实现代码import java.util.*; import java.util.concurrent.*; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.locks.ReentrantLock; /** * 极简 Raft 协议 Java 实现 * 核心包含Leader 选举、日志复制、心跳维持 */ public class SimpleRaftNode { // --- 常量定义 --- private static final int ELECTION_TIMEOUT_MIN 150; // 最小选举超时(ms) private static final int HEARTBEAT_INTERVAL 50; // 心跳间隔(ms) // --- 持久化状态 (对应 TiKV 的 Raft DB) --- private int currentTerm 0; // 当前任期 private String votedFor null; // 本任期投给了谁 (null 表示没投) private ListString log new ArrayList(); // 日志条目 (简化为字符串列表) // --- 易失性状态 --- private int commitIndex -1; // 已提交的最大日志索引 private int lastApplied -1; // 已应用到状态机的最大日志索引 // --- 领导人的易失性状态 --- private MapInteger, Integer nextIndex new ConcurrentHashMap(); // 每个 Follower 下一条日志索引 private MapInteger, Integer matchIndex new ConcurrentHashMap(); // 每个 Follower 已匹配的最大日志索引 // --- 运行时状态 --- public enum State { FOLLOWER, CANDIDATE, LEADER } private State state State.FOLLOWER; private final String nodeId; // 当前节点 ID private final ListSimpleRaftNode clusterNodes; // 集群中其他节点的引用 (模拟 RPC) // 定时器与锁 private ScheduledExecutorService executor Executors.newSingleThreadScheduledExecutor(); private ReentrantLock lock new ReentrantLock(); private ScheduledFuture? electionTimer; // --- 构造函数 --- public SimpleRaftNode(String id, ListSimpleRaftNode peers) { this.nodeId id; this.clusterNodes peers; resetElectionTimer(); } // --- 核心入口重置选举计时器 --- private void resetElectionTimer() { if (electionTimer ! null) electionTimer.cancel(false); // 随机超时时间防止选票分裂 int timeout ELECTION_TIMEOUT_MIN new Random().nextInt(150); electionTimer executor.schedule(() - { startElection(); }, timeout, TimeUnit.MILLISECONDS); } // --- 1. 选举机制 (Candidate 状态) --- private void startElection() { lock.lock(); try { System.out.println(nodeId - 触发选举...); // 1. 转为 Candidate state State.CANDIDATE; currentTerm; // 任期 1 votedFor nodeId; // 投给自己 // 2. 获取自己最新的日志信息 (用于判断日志是否足够新) int lastLogIndex log.size() - 1; int lastLogTerm (lastLogIndex 0) ? currentTerm : 0; // 简化处理 // 3. 并行向所有其他节点发送 RequestVote RPC AtomicInteger votes new AtomicInteger(1); // 自己的一票 CountDownLatch latch new CountDownLatch(clusterNodes.size()); for (SimpleRaftNode peer : clusterNodes) { executor.submit(() - { boolean granted peer.requestVote(currentTerm, nodeId, lastLogIndex, lastLogTerm); if (granted) votes.incrementAndGet(); latch.countDown(); }); } // 等待所有投票回来 (实际生产中会有超时控制) latch.await(100, TimeUnit.MILLISECONDS); // 4. 统计票数 if (votes.get() (clusterNodes.size() 1) / 2) { becomeLeader(); } else { // 选举失败退回 Follower 等待下次超时 state State.FOLLOWER; resetElectionTimer(); } } catch (Exception e) { e.printStackTrace(); } finally { lock.unlock(); } } /** * 接收者处理投票请求 (Voter 逻辑) * param term 候选人的任期 * param candidateId 候选人 ID * param lastLogIndex 候选人最后一条日志索引 * param lastLogTerm 候选人最后一条日志任期 */ public boolean requestVote(int term, String candidateId, int lastLogIndex, int lastLogTerm) { lock.lock(); try { System.out.println(nodeId - 收到投票请求: term term); // 规则 1: 如果对方的 Term 小于我的拒绝 (说明对方过时了) if (term currentTerm) return false; // 规则 2: 更新自己的 Term (如果发现更大的 Term立刻变回 Follower) if (term currentTerm) { currentTerm term; state State.FOLLOWER; votedFor null; } // 规则 3: 检查日志完整性 (关键) // 只有当候选人的日志 我自己的日志时我才投票 int myLastLogIndex log.size() - 1; int myLastLogTerm (myLastLogIndex 0) ? currentTerm : 0; boolean isLogUpToDate (lastLogTerm myLastLogTerm) || (lastLogTerm myLastLogTerm lastLogIndex myLastLogIndex); // 规则 4: 这一任期内还没投过人且对方日志够新 if ((votedFor null || votedFor.equals(candidateId)) isLogUpToDate) { votedFor candidateId; resetElectionTimer(); // 重置超时因为收到了合法的通信 System.out.println(nodeId - 投票给: candidateId); return true; } return false; } finally { lock.unlock(); } } // --- 2. 成为 Leader 后的逻辑 --- private void becomeLeader() { state State.LEADER; System.out.println(nodeId *** 当选为 Leader (Term: currentTerm ) ***); // 初始化 Leader 特有的变量 for (SimpleRaftNode peer : clusterNodes) { nextIndex.put(peer.nodeId, log.size()); matchIndex.put(peer.nodeId, -1); } // 启动心跳循环 sendHeartbeatsLoop(); } private void sendHeartbeatsLoop() { // 只要我是 Leader就不断发心跳 executor.scheduleAtFixedRate(() - { lock.lock(); try { if (state State.LEADER) { broadcastAppendEntries(true); // true 代表纯心跳 } } finally { lock.unlock(); } }, HEARTBEAT_INTERVAL, HEARTBEAT_INTERVAL, TimeUnit.MILLISECONDS); } // --- 3. 日志复制 (AppendEntries RPC) --- // 客户端调用接口 public void appendLog(String command) { lock.lock(); try { if (state ! State.LEADER) { System.out.println(错误: 我不是 Leader无法处理写请求); return; } // 1. 追加到本地日志 log.add(command); System.out.println(nodeId - 接收客户端指令: command); // 2. 广播给 Follower broadcastAppendEntries(false); } finally { lock.unlock(); } } private void broadcastAppendEntries(boolean isHeartbeat) { for (SimpleRaftNode peer : clusterNodes) { executor.submit(() - { int nextIdx nextIndex.get(peer.nodeId); // 计算 prevLogIndex 和 prevLogTerm (一致性检查用) int prevLogIndex nextIdx - 1; int prevLogTerm (prevLogIndex 0) ? currentTerm : 0; // 准备要发送的日志条目 (如果是心跳则为空) ListString entries isHeartbeat ? new ArrayList() : log.subList(nextIdx, log.size()); boolean success peer.appendEntries(currentTerm, nodeId, prevLogIndex, prevLogTerm, entries, commitIndex); if (success) { // 更新 nextIndex 和 matchIndex nextIndex.put(peer.nodeId, nextIdx entries.size()); matchIndex.put(peer.nodeId, nextIdx entries.size() - 1); // 尝试推进 commitIndex (简单实现只要有大多数匹配即可) updateCommitIndex(); } }); } } /** * 接收者处理 AppendEntries 请求 */ public boolean appendEntries(int term, String leaderId, int prevLogIndex, int prevLogTerm, ListString entries, int leaderCommit) { lock.lock(); try { // 1. 校验 Term if (term currentTerm) return false; if (term currentTerm) { currentTerm term; state State.FOLLOWER; votedFor null; } // 2. 收到合法 Leader 的心跳/日志重置选举计时器 resetElectionTimer(); state State.FOLLOWER; // 确保自己是 Follower // 3. 一致性检查如果 prevLogIndex 处的日志 Term 不匹配拒绝 if (prevLogIndex 0) { if (prevLogIndex log.size()) return false; // 日志缺失 // 简化版这里应该检查 log.get(prevLogIndex).term prevLogTerm // 在我们的简化模型中假设 term 就是 currentTerm所以略过复杂判断 } // 4. 追加新日志 if (!entries.isEmpty()) { // 冲突检测与覆盖逻辑 (简化版直接追加生产环境需处理冲突覆盖) log.addAll(entries); } // 5. 提交日志 if (leaderCommit commitIndex) { commitIndex Math.min(leaderCommit, log.size() - 1); applyLogToStateMachine(); } return true; } finally { lock.unlock(); } } private void updateCommitIndex() { // 统计 matchIndex找到最大的 N使得 N 及之前的日志被多数派复制 // 这里仅做演示不做严格算法实现 // ... } private void applyLogToStateMachine() { // 将已提交的日志应用到 KV 数据库 (TiKV 的 KV DB) while (lastApplied commitIndex) { lastApplied; String command log.get(lastApplied); System.out.println(nodeId 应用日志到状态机: command); // 这里调用 RocksDB put(key, value) } } // --- 测试主函数 --- public static void main(String[] args) throws InterruptedException { // 创建 3 个节点 SimpleRaftNode n1 new SimpleRaftNode(Node-1, new ArrayList()); SimpleRaftNode n2 new SimpleRaftNode(Node-2, new ArrayList()); SimpleRaftNode n3 new SimpleRaftNode(Node-3, new ArrayList()); // 互相注入引用 (模拟集群发现) n1.clusterNodes.add(n2); n1.clusterNodes.add(n3); n2.clusterNodes.add(n1); n2.clusterNodes.add(n3); n3.clusterNodes.add(n1); n3.clusterNodes.add(n2); // 等待选举发生 Thread.sleep(500); // 假设 n1 成为了 Leader写入数据 n1.appendLog(SET user_1 100); n1.appendLog(SET user_2 200); Thread.sleep(1000); } }