该文中的代码均为实验性质，不建议直接在生产中使用

准备工作

• 一台心爱的电脑 # M1 Mac mini

• 安装 Go 语言，参考 这里 # brew install go

• 安装 Git，参考 这里 # brew install git

• 趁手的 IDE # Goland

$ git clone git://g.csail.mit.edu/6.824-golabs-2022 6.824 # 将代码克隆到 6.824 文件夹
$ cd 6.824
$ tree -L 2 .
.
├── Makefile
└── src
    ├── go.mod
    ├── go.sum
    ├── kvraft # Lab 3
    ├── labgob # 封装了 gob 库以检查字段是不是以大写开头
    ├── labrpc # 专门给 Lab 使用的简单的 RPC 库
    ├── main # 可执行文件入口
    ├── models # Lab 3
    ├── mr # Lab 1
    ├── mrapps # Lab 1
    ├── porcupine
    ├── raft # Lab 2 - bingo
    ├── shardctrler # Lab 4
    └── shardkv # Lab 4

12 directories, 3 files

核心状态和规则

状态

规则

• 如果 commitIndex > lastApplied : 增加 lastApplied，将 log[lastApplied] 应用到状态机

• 如果 RPC 请求或者响应包含 term > currentTerm：将 currentTerm 设置为 term，并且将当前节点转为跟随者

• 响应领导者和候选人的 RPC 请求

• 如果选举超时时间过了，期间没有收到领导者的 AppendEntries RPC 也没有在当前任期投出自己的选票 : 当前节点转为候选者，发起新的选举

• 开始选举
• 增加 currentTerm，即当前任期 + 1
• 给自己投一票
• 重置选举超时时间定时器
• 发送 RequestVote 给其他所有节点

• 如果收到大多数节点的选票，则转为领导者

• 如果收到领导者的 AppendEntries RPC，则转为跟随者

• 如果选举超时时间过了，重新发起新的选举

• 在任期内
• 发送空的（entries 是空） AppendEntries RPCs 给所有节点
• 任期内每隔一段时间重复发送心跳包，时间间隔需要远小于选举超时时间间隔

• 收到来自客户端的请求
• 向自己节点的日志里面追加此日志条目
• 日志条目应用到状态机后，响应给客户端（实验中可以异步复制到其他节点，异步提交给状态机，尽快响应给客户端）

• 如果自己最后一个日志条目索引大于等于某个跟随者对应的 nextIndex
• 发送包含 nextIndex 和之后的日志条目的 AppendEntries 给此跟随者
• 如果成功，更新此跟随者对应的 netIndex 和 matchIndex
• 如果因为日志不一致失败，减小 nextIndex 并重试上面的流程

• 如果存在一个 N，N > commitIndex，大多数 matchIndex[i]≥N 且 log[N].term == currentTerm，那么设置 commitIndex = N

定义基本模型

type Entry struct {
	Command interface{} //来自客户端的命令
	Term    int //接收到此命令的任期
}

// A Go object implementing a single Raft peer.
type Raft struct {
	mu        sync.Mutex          // Lock to protect shared access to this peer's state
	peers     []*labrpc.ClientEnd // RPC end points of all peers
	persister *Persister          // Object to hold this peer's persisted state
	me        int                 // this peer's index into peers[]
	dead      int32               // set by Kill()
	state              State
	currentTerm        int
	electionResetEvent time.Time
	electionTimeout    time.Duration
	votedFor           int
	votes              int
	commitIndex        int
	lastApplied        int
	logs               []LogEntry
	nextIndex          []int
	matchIndex         []int
	applyChan          chan ApplyMsg
}

type LogEntry struct {
	Term    int
	Command interface{}
}

type State int

const (
	Follower State = iota
	Leader
	Candidate
)

Part A - 实现领导者选举

• 构建一个多节点的集群，所有节点初始状态均为 Follower，并在随机的 electionTimeout 超时之后开始变为 Candidate 状态并开启新任期的选举流程

• 选举开始后，候选人通过向集群中的其他节点发送 RequestVote RPC 消息来搜集选票，选票超过集群节点大多数则当选此任期领导者

• 当选之后，领导者需定时向跟随者发送心跳 AppendEntries RPC 以保持权威，阻止他们发起新的选举，直到任期结束

RequestVote RPC

• 如果 term < currentTerm，拒绝投票，返回自己的任期和 false

• 如果 votedFor 是空的（我们实现为值是 -1）或者跟 candidateId 相等，并且候选者的日志至少是跟当前节点一样新，则投出选票，返回 true

• Raft 是通过比较最后一个日志条目的索引和任期来判断两个节点谁的日志更新。如果两个节点的最后一个日志条目有不同的任期，任期越大则节点日志越新；如果两个节点最后的日志条目的任期相同，则索引越大，日志越新

AppendEntries RPC

• 如果 term < currentTerm 返回 false

• 如果在 prevLogIndex 处没有可以匹配 prevLogTerm 的日志条目，返回 false

• 如果已经存在的日志条目跟新的条目冲突，即相同的 index，不同的任期，则删除已经存在的条目和所有它后面的条目

• 追加所有不存在的新条目

• 如果 leaderCommit > commitIndex，将 commitIndex 设置为 min(leaderCommit, index of last new entry)

// Make
// the service or tester wants to create a Raft server. the ports
// of all the Raft servers (including this one) are in peers[]. this
// server's port is peers[me]. all the servers' peers[] arrays
// have the same order. persister is a place for this server to
// save its persistent state, and also initially holds the most
// recent saved state, if any. applyCh is a channel on which the
// tester or service expects Raft to send ApplyMsg messages.
// Make() must return quickly, so it should start goroutines
// for any long-running work.
func Make(peers []*labrpc.ClientEnd, me int,
	persister *Persister, applyCh chan ApplyMsg) *Raft {
rf := &Raft{}
	rf.peers = peers
	rf.persister = persister
	rf.me = me

	// Your initialization code here (2A, 2B, 2C).
	rf.state = Follower
	rf.votes = 0
	rf.votedFor = -1
	rf.currentTerm = 0
	rf.electionTimeout = time.Duration(150+rand.Intn(300)) * time.Millisecond
	rf.electionResetEvent = time.Now()
	rf.commitIndex = 0
	rf.lastApplied = 0
	rf.logs = make([]LogEntry, 0)
	rf.applyChan = applyCh

	// initialize from state persisted before a crash
	rf.readPersist(persister.ReadRaftState())

	// start ticker goroutine to start elections
	go rf.ticker()

	return rf
}

Part B - log replication