目录前言一、引子二、3A实现2.1 结构体设计每个节点应有的状态定义一些枚举常量raft结构体2.2 RPC 结构RequestVote / AppendEntries2.3 raft节点初始化2.4 定时检测的go协程ticker2.5 开始选举 startElection广播心跳2.6 RequestVote 处理逻辑2.7 AppendEntries 心跳逻辑可以先做简单的实现到B部分再做完整实现2.8节点是否被关闭和关闭节点2.9 获取节点状态是否为leader三、测试和遇到的问题四、收获前言花两天把lab3A写了一下自己构建思路后再来看本篇会更有收获。gitee地址也放在末尾。一、引子2023之后raft被放到lab3lab2是实现一个简单的kv服务器在lab3中主要分为ABCD四个部分分别是raft的领导选举状态机的日志复制持久化和快照。根据我的观察只要把论文看个三四遍把细节都看一下应该没问题。具体难度排行DBAC当然只看论文估计也很难理解需要额外在b站上找理解视频或者辅助学习文章。二、3A实现2.1 结构体设计在开始前我们先看一下这个测试函数看看他们的调用关系以及需要我们实现什么东西把自己的思路理清一下TestInitialElection3A验证在一个可靠网络、3 个节点中能否在一段时间内选出一个 Leader。验证 term 至少从 1 开始并且在没有任何网络分区的情况下Leader 和 term 都是稳定的不会乱跳。TestReElection3A断开当前 Leader 后剩下的节点能否重新选出新的 Leader。旧 Leader 回来之后能否变成 Follower 而不是抢占新 Leader。当没有多数派断开 2 个节点只剩 1 个时集群不会错误地产生 Leader多数派恢复后又能重新选出 Leader。TestManyElections3A在 7 节点、频繁网络断连/重连的情况下是否始终“最多存在一个 Leader”且几轮下来不会出现长时间无 Leader 的情况。这是最容易暴露随机超时没写好、死锁、锁没释放、Candidate 不退位等问题的测试。func TestInitialElection3A(t *testing.T) { servers : 3 ts : makeTest(t, servers, true, false) defer ts.cleanup() tester.AnnotateTest(TestInitialElection3A, servers) ts.Begin(Test (3A): initial election) // is a leader elected? ts.checkOneLeader() // sleep a bit to avoid racing with followers learning of the // election, then check that all peers agree on the term. time.Sleep(50 * time.Millisecond) term1 : ts.checkTerms() if term1 1 { ts.t.Fatalf(term is %v, but should be at least 1, term1) } // does the leaderterm stay the same if there is no network failure? time.Sleep(2 * RaftElectionTimeout) term2 : ts.checkTerms() if term1 ! term2 { fmt.Printf(warning: term changed even though there were no failures) } // there should still be a leader. ts.checkOneLeader() } func TestReElection3A(t *testing.T) { servers : 3 ts : makeTest(t, servers, true, false) defer ts.cleanup() tester.AnnotateTest(TestReElection3A, servers) ts.Begin(Test (3A): election after network failure) leader1 : ts.checkOneLeader() // if the leader disconnects, a new one should be elected. ts.g.DisconnectAll(leader1) tester.AnnotateConnection(ts.g.GetConnected()) ts.checkOneLeader() // if the old leader rejoins, that shouldnt // disturb the new leader. and the old leader // should switch to follower. ts.g.ConnectOne(leader1) tester.AnnotateConnection(ts.g.GetConnected()) leader2 : ts.checkOneLeader() // if theres no quorum, no new leader should // be elected. ts.g.DisconnectAll(leader2) ts.g.DisconnectAll((leader2 1) % servers) tester.AnnotateConnection(ts.g.GetConnected()) time.Sleep(2 * RaftElectionTimeout) // check that the one connected server // does not think it is the leader. ts.checkNoLeader() // if a quorum arises, it should elect a leader. ts.g.ConnectOne((leader2 1) % servers) tester.AnnotateConnection(ts.g.GetConnected()) ts.checkOneLeader() // re-join of last node shouldnt prevent leader from existing. ts.g.ConnectOne(leader2) tester.AnnotateConnection(ts.g.GetConnected()) ts.checkOneLeader() } func TestManyElections3A(t *testing.T) { servers : 7 ts : makeTest(t, servers, true, false) defer ts.cleanup() tester.AnnotateTest(TestManyElection3A, servers) ts.Begin(Test (3A): multiple elections) ts.checkOneLeader() iters : 10 for ii : 1; ii iters; ii { // disconnect three nodes i1 : rand.Int() % servers i2 : rand.Int() % servers i3 : rand.Int() % servers ts.g.DisconnectAll(i1) ts.g.DisconnectAll(i2) ts.g.DisconnectAll(i3) tester.AnnotateConnection(ts.g.GetConnected()) // either the current leader should still be alive, // or the remaining four should elect a new one. ts.checkOneLeader() ts.g.ConnectOne(i1) ts.g.ConnectOne(i2) ts.g.ConnectOne(i3) tester.AnnotateConnection(ts.g.GetConnected()) } ts.checkOneLeader() }对于大致的结构体其实论文中的表格以及给的很清楚了我们只需要补充部分细节。每个节点应有的状态由此我们可以在raft中定义如上几个变量(对于表格中的字段的个人理解以注释方式写在代码中了。定义一些枚举常量// HeartBeatTimeout 定义一个全局心跳超时时间 var HeartBeatTimeout 50 * time.Millisecond // 固定超时时间枚举 const ( HeartbeatInterval 50 * time.Millisecond // 心跳间隔固定 ElectionTimeoutMin 150 * time.Millisecond // 选举超时下限 ElectionTimeoutMax 300 * time.Millisecond // 选举超时上限 RPCTimeout 100 * time.Millisecond // RPC 超时固定 ) // 枚举节点状态 type PeerState int const ( Follower PeerState iota //追随者 Candidate //候选者 Leader //领导者 ) // 日志条目 type LogEntry struct { Term int //任期 Command interface{} //命令 }raft结构体// A Go object implementing a single Raft peer. type Raft struct { mu sync.Mutex // Lock to protect shared access to this peers state 互斥锁保护共享访问 peers []*labrpc.ClientEnd // RPC end points of all peers 所有节点的RPC端点 persister *tester.Persister // Object to hold this peers persisted state 持久化状态 me int // this peers index into peers[] 当前节点在peers中的索引 // Your data here (3A, 3B, 3C). // Look at the papers Figure 2 for a description of what // state a Raft server must maintain. state PeerState //当前节点状态 3A term int //当前任期 3A votedFor int //当前任期投票给谁 3A votenums int //当前节点获取到的票数 3A logs []LogEntry //日志 3A 下标0的内容是占位的真正的第一条命令在下标1 //所有节点共享的不稳定状态 commitIndex int //已提交的日志索引 lastApplied int //已应用的日志索引 //leader的不稳定状态 leader专用 nextIndex []int //下一个要发送的日志索引 matchIndex []int //已匹配的日志索引 lastHeartBeatTime time.Time //最后一次心跳时间 3A electionTimeout time.Duration //当前选举超时时间 3A applyChan chan raftapi.ApplyMsg // 用来写入应用消息的通道 dead int32 //节点是否死亡 applyCond *sync.Cond // 用于唤醒 applier }logs[0]占位一条{Term:0, Command:nil}是为了后面索引从 1 开始和论文里 Figure 2 对齐真实命令从 index1 起。commitIndex / lastApplied / nextIndex / matchIndex在 3A 中其实用得不多但是提早放进去是为了 3B 日志复制和 3C 不用再大改结构体。2.2 RPC 结构RequestVote / AppendEntries从严格意义上说这个可能算是2b的内容因为是日志增量同步但是对于leader选举后的心跳建立来讲这又属于2a固在此一起定义了。// 追加条目RPC请求参数 type AppendEntriesArgs struct { // Your data here (3B). Term int //leader的任期 LeaderId int //leader的ID PrevLogIndex int //紧接着新条目之前的最后一个条目的索引 PrevLogTerm int //紧接着新条目之前的最后一个条目的任期 Entries []LogEntry //需要追加的新条目 LeaderCommit int //leader的CommitIndex } // 追加条目RPC回复参数 type AppendEntriesReply struct { // Your data here (3B). Term int //用于leader更新自己当前的任期 Success bool //如果 follower 包含匹配的 prevLogIndex 和 prevLogTerm 条目则为 true }3A 阶段实际只用到Term和LeaderId做心跳PrevLogIndex/PrevLogTerm/Entries/LeaderCommit是为 3B 的日志复制预留的。RequestVote RPC的定义type RequestVoteArgs struct { //投票请求参数 // Your data here (3A, 3B). Term int //候选人的任期 CandidateId int //候选人的ID LastLogIndex int //候选人的最后一个日志条目的索引 LastLogTerm int //候选人的最后一个日志条目的任期 } // example RequestVote RPC reply structure. // field names must start with capital letters! type RequestVoteReply struct { //投票回复参数 // Your data here (3A). Term int //leader的当前任期给候选人自行更新 VoteGranted bool //是否获得投票 }2.3 raft节点初始化unc Make(peers []*labrpc.ClientEnd, me int, persister *tester.Persister, applyCh chan raftapi.ApplyMsg) raftapi.Raft { rf : Raft{} rf.peers peers rf.persister persister rf.me me rf.applyChan applyCh //初始化状态和日志 rf.state Follower //初始化时为追随者 rf.term 0 rf.votedFor -1 rf.votenums 0 rf.logs []LogEntry{{Term: 0, Command: nil}} //对0下标占位 rf.commitIndex 0 rf.lastApplied 0 rf.nextIndex make([]int, len(peers)) rf.matchIndex make([]int, len(peers)) rf.lastHeartBeatTime time.Now() rf.electionTimeout rf.randElectionTimeout() // Your initialization code here (3A, 3B, 3C). // initialize from state persisted before a crash rf.readPersist(persister.ReadRaftState()) // start ticker goroutine to start elections go rf.ticker() go rf.applier() //apply 日志到状态机 3A 可以空实现或简单阻塞3B 再写 rand.Seed(time.Now().UnixNano() int64(rf.me)) //初始化随机种子 return rf }随机超时时间生成函数// 随机选举超时时间生成 150-300ms论文推荐 func (rf *Raft) randElectionTimeout() time.Duration { return time.Duration(150rand.Intn(150)) * time.Millisecond }2.4 定时检测的go协程ticker如果是当前节点是leader就广播心跳如果不是并且距离leader的上一次心跳时间已经超过了超时时间就发起选举// 选举超时ticker func (rf *Raft) ticker() { for !rf.Killed() { // Your code here (3A) // Check if a leader election should be started. time.Sleep(30 * time.Millisecond) //30ms检查一次 rf.mu.Lock() state : rf.state if state Leader { // Leader 固定频率发心跳 // 这里可以加一个 heartbeatTimer或者利用 Sleep 的频率 rf.mu.Unlock() rf.broadcastHeartbeat() } else { // Follower/Candidate 逻辑检查是否选举超时 elapsed : time.Since(rf.lastHeartBeatTime) timeout : rf.electionTimeout rf.mu.Unlock() if elapsed timeout { DPrintf([%d] ticker: 选举超时 触发 startElection state%d, rf.me, state) rf.startElection() } } } }2.5 开始选举 startElection先看看当前状态是不是leader然后把这个时候的状态都记录下来然后向其他节点发送请求投票RPC然后查看回复如果有节点的任期比自己的大马上转为follower其他状态也重置当大多数节点同意并且这个节点状态是候选人那么这个节点当选leader然后初始化leader独有的变量接着向其他节点放送心跳宣示主权// 开始选举 func (rf *Raft) startElection() { rf.mu.Lock() if rf.state Leader { //如果是leader返回 rf.mu.Unlock() return } rf.state Candidate //切换到候选人 rf.term //任期 rf.votedFor rf.me //投票给自己 rf.votenums 1 DPrintf([%d] startElection: 转为 Candidate term%d, rf.me, rf.term) rf.lastHeartBeatTime time.Now() //更新最后的心跳时间 rf.electionTimeout rf.randElectionTimeout() // 每次竞选都要重置随机时间 //记录当前 term 和自己的 lastLogIndex/Term // 拷出来放在局部变量防止 RPC 回来时 term 已变 curTerm : rf.term lastlogindex : len(rf.logs) - 1 lastlogterm : rf.logs[lastlogindex].Term rf.mu.Unlock() for i : 0; i len(rf.peers); i { if i rf.me { continue } go func(server int) { rf.mu.Lock() args : RequestVoteArgs{ Term: curTerm, CandidateId: rf.me, LastLogIndex: lastlogindex, LastLogTerm: lastlogterm, } rf.mu.Unlock() reply : RequestVoteReply{} res : rf.sendRequestVote(server, args, reply) if res { //请求投票返回成功 rf.mu.Lock() //如果收到任期比自己大的节点的回复 if reply.Term args.Term { if reply.Term rf.term { rf.term reply.Term } rf.state Follower rf.votedFor -1 rf.votenums 0 DPrintf([%d] startElection: 收到更大 term 退回 Follower from%d replyTerm%d, rf.me, server, reply.Term) rf.mu.Unlock() return } //判断自己是否还是竞选者且任期不冲突 if rf.state ! Candidate || args.Term rf.term { rf.mu.Unlock() return } //获得投票 if reply.VoteGranted { rf.votenums if (rf.votenums (len(rf.peers)/2 1)) (rf.state Candidate) { //条件满足变为leader rf.state Leader rf.votedFor -1 rf.votenums 0 DPrintf([%d] startElection: 当选 Leader term%d votes%d, rf.me, rf.term, rf.votenums1) // 初始化 nextIndex 和 matchIndex rf.nextIndex make([]int, len(rf.peers)) rf.matchIndex make([]int, len(rf.peers)) for i : range rf.nextIndex { rf.nextIndex[i] len(rf.logs) } rf.mu.Unlock() //先解锁再发送心跳 // 发送心跳 rf.broadcastHeartbeat() } else { rf.mu.Unlock() return } } else { rf.mu.Unlock() return } } }(i) } }在每次竞选时都要更新超时时间你可以想象一下如果超时时间都是一样的刚好全部节点同时发起开始选举那么每个人都先投自己一票每个人都得不到大多数票就没有leader然后持续到选举结束又重新选举但是还是这个情况然后选举时对外的服务接口是关闭的那带来的后果是非常严重的。广播心跳func (rf *Raft) broadcastHeartbeat() { rf.mu.Lock() defer rf.mu.Unlock() if rf.state ! Leader { return } curTerm : rf.term // 获取 Leader 自己的最后一条日志信息用于填充 PrevLogIndex/Term lastLogIndex : len(rf.logs) - 1 lastLogTerm : rf.logs[lastLogIndex].Term //遍历所有节点发送心跳 for i : range rf.peers { if i rf.me { continue } args : AppendEntriesArgs{ Term: curTerm, LeaderId: rf.me, PrevLogIndex: lastLogIndex, // 心跳用 leader 最后一条日志索引 PrevLogTerm: lastLogTerm, // 对应任期 Entries: []LogEntry{}, // 空切片表示心跳 LeaderCommit: rf.commitIndex, } reply : AppendEntriesReply{} // 发送 RPC go rf.sendAppendEntries(i, args, reply) } }2.6 RequestVote 处理逻辑------每个节点只能投一票只投票给任期不小于自己并且日志至少和自己一样新------// example RequestVote RPC handler. // 请求投票接口 func (rf *Raft) RequestVote(args *RequestVoteArgs, reply *RequestVoteReply) { // Your code here (3A, 3B). rf.mu.Lock() defer rf.mu.Unlock() DPrintf([%d] RequestVote from candidate%d argsTerm%d myTerm%d, rf.me, args.CandidateId, args.Term, rf.term) if args.Term rf.term { DPrintf([%d] RequestVote reject: argsTerm %d myTerm %d, rf.me, args.Term, rf.term) reply.Term rf.term reply.VoteGranted false return } if args.Term rf.term { //在任期相等时不应该更新任期并重置投票 //在同一任期内如果有多个 Candidate // 它们会因为收到对方的投票请求而互相降级 // 导致谁都选不上甚至出现逻辑死循环 if args.Term rf.term { rf.term args.Term rf.votedFor -1 //重置投票 } rf.state Follower } //检查候选人日志是否和我的一样新 isuptodate : rf.isLogUpToDate(args) //没有投票或者已经投票给这个候选人并且候选人日志是否和我的一样新时 //投票给他重置选举计时器 if (rf.votedFor -1 || rf.votedFor args.CandidateId) isuptodate { rf.votedFor args.CandidateId reply.VoteGranted true rf.lastHeartBeatTime time.Now() DPrintf([%d] RequestVote grant to %d term%d, rf.me, args.CandidateId, rf.term) } else { reply.VoteGranted false DPrintf([%d] RequestVote reject: votedFor%d or !isUpToDate, rf.me, rf.votedFor) } reply.Term rf.term } // 判断候选人日志是否和我的一样新 func (rf *Raft) isLogUpToDate(args *RequestVoteArgs) bool { myLastIndex : len(rf.logs) - 1 myLastTerm : rf.logs[myLastIndex].Term if args.LastLogTerm myLastTerm { return true } if args.LastLogTerm myLastTerm { return false } // 任期相同比较索引 return args.LastLogIndex myLastIndex }2.7 AppendEntries 心跳逻辑可以先做简单的实现到B部分再做完整实现“这一步先只处理 term 检查 心跳刷新lastHeartBeatTime Candidate 退位3B 再在这里加上 PrevLogIndex/PrevLogTerm 检查和日志截断追加。”// 追加条目rpc也用来发送心跳 func (rf *Raft) AppendEntries(args *AppendEntriesArgs, reply *AppendEntriesReply) { rf.mu.Lock() defer rf.mu.Unlock() DPrintf([%d] AppendEntries from L%d term%d, my term%d state%d, rf.me, args.LeaderId, args.Term, rf.term, rf.state) // 发现任期比自己小 if args.Term rf.term { DPrintf([%d] AppendEntries: 拒绝 term 更小 leader%d argsTerm%d myTerm%d, rf.me, args.LeaderId, args.Term, rf.term) reply.Success false reply.Term rf.term return } rf.lastHeartBeatTime time.Now() if args.Term rf.term || rf.state Candidate { rf.term args.Term rf.state Follower DPrintf([%d] step down to Follower due to AE term%d, rf.me, rf.term) } reply.Success true reply.Term rf.term } ​ func (rf *Raft) sendAppendEntries(server int, args *AppendEntriesArgs, reply *AppendEntriesReply) bool { ok : rf.peers[server].Call(Raft.AppendEntries, args, reply) if ok { rf.mu.Lock() defer rf.mu.Unlock() // 处理回复任期更新、调整 nextIndex 等 if reply.Term rf.term { rf.term reply.Term rf.state Follower rf.votedFor -1 return ok } // 2. 状态检查 if rf.state ! Leader || rf.term ! args.Term { return ok } // 3B 以后在这里处理日志同步的 reply.Success 为 false 的情况 } return ok } ​2.8节点是否被关闭和关闭节点func (rf *Raft) Kill() { atomic.StoreInt32(rf.dead, 1) } func (rf *Raft) Killed() bool { z : atomic.LoadInt32(rf.dead) return z 1 //1表示死亡0表示存活 }2.9 获取节点状态是否为leader// return currentTerm and whether this server // believes it is the leader. func (rf *Raft) GetState() (int, bool) { //获取当前任期和是否是领导者 var term int var isleader bool // Your code here (3A). rf.mu.Lock() term rf.term isleader (rf.state Leader) rf.mu.Unlock() return term, isleader }三、测试和遇到的问题在src目录下执行cd raft1 go test -run 3A -vstartElection 中「收到票但未过半」未解锁当restrue且VoteGrantedtrue但票数未过半时会进入if reply.VoteGranted却不进入「过半」分支也未进入else { Unlock; return }导致持锁退出、锁泄漏。多轮选举后所有要拿rf.mu的 goroutine 阻塞 → TestManyElections3A 卡住。修复在「获得投票」分支里无论是否过半都要在 return 前保证执行一次rf.mu.Unlock()例如未过半时也 Unlock 再 return。成为 Leader 后调用 broadcastHeartbeat 死锁在持锁状态下调用rf.broadcastHeartbeat()而broadcastHeartbeat()内部会再次Lock()同 goroutine 二次加锁导致死锁。修复在「当选 Leader」分支里初始化完 nextIndex/matchIndex 后先rf.mu.Unlock()再调用rf.broadcastHeartbeat()然后 return。startElection 入口读 state 未加锁开头if rf.state Leader { return }存在竞态。修复函数入口先rf.mu.Lock()再判断rf.state Leader若是则 Unlock 后 return。AppendEntries 收到合法心跳未退位本节点为 Candidate 时若收到合法 Leader 的 AppendEntries同 term 或更高应转为 Follower 并重置选举计时。修复在接受该 RPC 时设rf.state Follower若args.Term rf.term则更新rf.term、rf.votedFor -1并更新lastHeartBeatTime。isLogUpToDate 比较错误若 term 相同再比LastLogIndex myLastIndex。原逻辑用args.LastLogIndex myLastTerm索引与任期混比。论文要求先比最后一条日志的term再比index。修复先比LastLogTerm与myLastTerm这是最终测试通过的输出lcziv-yef3xahqtc5i3z5jzmr5:~/mit6.5840/6.5840/src$ make RUN-run 3A raft1go build -race -o main/raft1d main/raft1d.gocd raft1 go test -v -race -run 3A RUN TestInitialElection3A Test (3A): initial election (reliable network)... ... Passed -- time 3.0s #peers 3 #RPCs 192 #Ops 0--- PASS: TestInitialElection3A (3.47s) RUN TestReElection3A Test (3A): election after network failure (reliable network)... ... Passed -- time 4.6s #peers 3 #RPCs 390 #Ops 0--- PASS: TestReElection3A (5.12s) RUN TestManyElections3A Test (3A): multiple elections (reliable network)... ... Passed -- time 5.6s #peers 7 #RPCs 1680 #Ops 0--- PASS: TestManyElections3A (6.59s)PASS ok 6.5840/raft1 16.203s四、收获在把论文看了几遍、理解清楚选举流程之后3A 本身其实不算很难。真正折磨人的是那些一开始测不出来的小 bug比如锁没释放导致TestManyElections3A卡死超时时间没随机导致选举风暴Candidate 收到心跳不退位导致一直没有稳定的 Leader。最后真正让我过关的是在关键路径上加足够多的调试打印RequestVote/AppendEntries/startElection/ticker把整个选举过程“看懂”。建议每个人按自己的理解再实现一遍而不是完全抄参考实现这样后面做 3B/3C 的时候会轻松很多。最后希望有错误的地方多指正指正~完整实现mit6.5840: 用来记录mit6.5840原6.824的实现历程觉得有收获的可以帮忙点个star~