0. 概述
在上一篇文章中介绍了 raft 读请求源码走读，本文继续结合 etcd-raft 源码介绍 raft 写请求流程。
需要说明的是，本文使用的是单节点集群环境，后续会补充多节点集群环境加以介绍。
1. 写请求流程
1.1 客户端写入请求
客户端通过 PUT 请求写入键值对请求：
curl -L http://127.0.0.1:12380/my-key -XPUT -d hello
该请求被 httpKVAPI 处理：
func (h *httpKVAPI) ServeHTTP(w http.ResponseWriter, r *http.Request) { key := r.RequestURI defer r.Body.Close() switch r.Method { case http.MethodPut: // 匹配 PUT 方法，处理写请求v, err := io.ReadAll(r.Body) if err != nil { log.Printf("Failed to read on PUT (%v)\n", err) http.Error(w, "Failed on PUT", http.StatusBadRequest) return } h.store.Propose(key, string(v)) // 调用 h.store.Propose 提交该请求w.WriteHeader(http.StatusNoContent)}...
}func (s *kvstore) Propose(k string, v string) { var buf strings.Builder if err := gob.NewEncoder(&buf).Encode(kv{k, v}); err != nil { log.Fatal(err) } s.proposeC <- buf.String()
}
httpKVAPI.ServeHTTP 的重点在于 httpKVAPI.store.Propose 方法。该方法的作用是将写请求键值对编码后写入 kvstore.proposeC 通道。
那么谁在消费 proposeC 呢？
1.2 应用层消费 proposeC
应用层 raftNode 消费客户端写入 proposeC 的写请求。
首先查看 raftNode 的启动流程及整体结构。
1.2.1 raftNode 启动流程
在 main.go 中 newRaftNode 会创建应用层 RaftNode 节点：
func main() {...proposeC := make(chan string) defer close(proposeC) confChangeC := make(chan raftpb.ConfChange) defer close(confChangeC)var kvs *kvstore getSnapshot := func() ([]byte, error) { return kvs.getSnapshot() } // 创建 RaftNode 节点，并且将客户端写入的 proposeC 作为参数传递给 newRaftNodecommitC, errorC, snapshotterReady := newRaftNode(*id, strings.Split(*cluster, ","), *join, getSnapshot, proposeC, confChangeC)...
}func newRaftNode(id int, peers []string, join bool, getSnapshot func() ([]byte, error), proposeC <-chan string, confChangeC <-chan raftpb.ConfChange,
) (<-chan *commit, <-chan error, <-chan *snap.Snapshotter) { commitC := make(chan *commit) errorC := make(chan error) rc := &raftNode{ proposeC: proposeC, // proposeC 传入 raftNode... } go rc.startRaft() return commitC, errorC, rc.snapshotterReady
}
创建 raftNode 后，调用 raftNode.startRaft 方法启动一个协程异步运行 startRaft 方法：
func (rc *raftNode) startRaft() { ...go rc.serveRaft() // 启动协程监听 raft apigo rc.serveChannels() // 启动协程监听 raftNode 的通道
}
重点介绍 raft.serveChannels 方法：
func (rc *raftNode) serveChannels() { ... // send proposals over raft go func() { confChangeCount := uint64(0) for rc.proposeC != nil && rc.confChangeC != nil { select { case prop, ok := <-rc.proposeC: // 在这里消费 proposeC 的写请求键值对if !ok { rc.proposeC = nil } else { // blocks until accepted by raft state machine rc.node.Propose(context.TODO(), []byte(prop)) } } } // client closed channel; shutdown raft if not already close(rc.stopc) }() // for-select 事件循环 for { select { case <-ticker.C: rc.node.Tick() // store raft entries to wal, then publish over commit channel case rd := <-rc.node.Ready(): ...} }
}
在 serveChannels 中应用层 raftNode 从 proposeC 中消费写请求。
如果 proposeC 关闭将 proposeC 设为 nil 不允许写入。
如果正常读取写请求，则进入 raftNode.node.Propose 处理。
raftNode 组合 node 对象，node 是算法层的入口，通过提供的一系列方法暴露给应用层调用。Propose 即为暴露的其中一个方法。
1.2.2 算法层处理写请求
1.2.2.1 算法层初始化流程
为了介绍逻辑的完整性，这里有必要在进入 raftNode.startRaft 方法查看算法层 node 是如何创建的：
func (rc *raftNode) startRaft() {...c := &raft.Config{ ID: uint64(rc.id), ElectionTick: 10, HeartbeatTick: 1, Storage: rc.raftStorage, MaxSizePerMsg: 1024 * 1024, MaxInflightMsgs: 256, MaxUncommittedEntriesSize: 1 << 30, } if oldwal || rc.join { rc.node = raft.RestartNode(c) } else { // 首次创建 raftNode.noderc.node = raft.StartNode(c, rpeers) }...
}func StartNode(c *Config, peers []Peer) Node { n := setupNode(c, peers) go n.run() return n
}
在 raft.StartNode(或 raft.RestartNode) 创建 raftNode.node 对象。实际是异步启了一个协程运行 node.run 方法。
1.2.2.2 node.Propose 处理写请求提交
上一节创建 node后，应用层调用 raftNode.node.Propose 方法提交写请求到算法层：
func (n *node) Propose(ctx context.Context, data []byte) error { // 将写请求数据封装为 Entry，将 Entry 封装为 Messagereturn n.stepWait(ctx, pb.Message{Type: pb.MsgProp, Entries: []pb.Entry{{Data: data}}})
}func (n *node) stepWait(ctx context.Context, m pb.Message) error { // 等待提交写请求处理完成return n.stepWithWaitOption(ctx, m, true)
}func (n *node) stepWithWaitOption(ctx context.Context, m pb.Message, wait bool) error { ...ch := n.propc pm := msgWithResult{m: m} // 将 Message 封装为 msgWithResultif wait { pm.result = make(chan error, 1) } select { case ch <- pm: // 将 msgWithResult 写入 node.propc 通道if !wait { return nil } case <-ctx.Done(): return ctx.Err() case <-n.done: return ErrStopped } select { case err := <-pm.result: // 等待 msgWithResult.result 返回结果if err != nil { return err } ...} return nil
}
算法层 node.Propose 将数据封装为 msgWithResult 发给 node.propc 通道，并且监听 msgWithResult.result 通道等待返回结果。
那么 msgWithResult 的数据是在哪里消费的呢？
1.2.2.3 消费 msgWithResult
在 1.2.2.1 算法层初始化流程 介绍了创建 node 对象后会异步启动 node.run 协程，这里继续看协程内部在做什么：
func (n *node) run() { var propc chan msgWithResult var readyc chan Ready var advancec chan struct{} var rd Ready r := n.rn.raft lead := None for { ... if lead != r.lead { if r.hasLeader() { if lead == None { r.logger.Infof("raft.node: %x elected leader %x at term %d", r.id, r.lead, r.Term) } else { r.logger.Infof("raft.node: %x changed leader from %x to %x at term %d", r.id, lead, r.lead, r.Term) } // 如果集群有 leader 则将 node.propc 赋给 propc// 意味着只在有 leader 的情况下处理写请求提交propc = n.propc } else { r.logger.Infof("raft.node: %x lost leader %x at term %d", r.id, lead, r.Term) // 如果集群无 leader 将 propc 设为 nil// 意味着无 leader 时不会处理写请求提交propc = nil } lead = r.lead } select { case pm := <-propc: // 从 node.propc 中读取 msgWithResult 对象m := pm.m // 获取消息m.From = r.id // 消息是在同一个 raft 节点处理的，将 m.From 设为当前 raft 节点的 iderr := r.Step(m) // 调用 raft.Step 处理消息if pm.result != nil { pm.result <- err close(pm.result) } ... } }
}
node.run 消费 node.propc，并且调用 raft.Step 处理消费的消息：
func (r *raft) Step(m pb.Message) error { switch { case m.Term == 0: // 消息的 Term 为 0，表示该消息为本地消息// local message case m.Term > r.Term: ...case m.Term < r.Term: ...} switch m.Type { case pb.MsgHup: ...case pb.MsgStorageAppendResp: ...case pb.MsgStorageApplyResp: ... case pb.MsgVote, pb.MsgPreVote: ... default: err := r.step(r, m) // 这里的消息类型为 MsgProp 会进到默认分支if err != nil { return err } } return nil
}
消息的类型为 MsgProp 会进入到 raft.step 方法处理。raft.step 是一个指针函数指向不同角色的 step 处理函数。
根据不同角色有如下几种处理函数。
candidate
候选者 candidate 处理 MsgProp 消息：
func stepCandidate(r *raft, m pb.Message) error { switch m.Type { case pb.MsgProp: r.logger.Infof("%x no leader at term %d; dropping proposal", r.id, r.Term) return ErrProposalDropped...
}
候选者并不能直接处理 MsgProp 写请求的提交消息，如果收到将直接丢弃。
follower
func stepFollower(r *raft, m pb.Message) error { switch m.Type { case pb.MsgProp: if r.lead == None { r.logger.Infof("%x no leader at term %d; dropping proposal", r.id, r.Term) return ErrProposalDropped } else if r.disableProposalForwarding { r.logger.Infof("%x not forwarding to leader %x at term %d; dropping proposal", r.id, r.lead, r.Term) return ErrProposalDropped } m.To = r.lead r.send(m) case pb.MsgApp: ... case pb.MsgHeartbeat: ... case pb.MsgSnap: ... case pb.MsgTransferLeader: ... case pb.MsgTimeoutNow: ...} return nil
}
跟随者 follower 收到 MsgProp 写请求提交消息，会判断当前集群有无 leader。如果有，则调用 raft.send 转发该消息给 leader 处理。如果没有，则丢弃该消息。
leader
func stepLeader(r *raft, m pb.Message) error { switch m.Type { case pb.MsgBeat: .... case pb.MsgCheckQuorum: ... case pb.MsgProp: if len(m.Entries) == 0 { r.logger.Panicf("%x stepped empty MsgProp", r.id) } if r.trk.Progress[r.id] == nil { return ErrProposalDropped } if r.leadTransferee != None { r.logger.Debugf("%x [term %d] transfer leadership to %x is in progress; dropping proposal", r.id, r.Term, r.leadTransferee) return ErrProposalDropped } for i := range m.Entries { e := &m.Entries[i] var cc pb.ConfChangeI // Entry 类型为 EntryNormal，不匹配if e.Type == pb.EntryConfChange { ... } else if e.Type == pb.EntryConfChangeV2 { ...} if cc != nil { ... } } // 调用 raft.appendEntry 处理 msgWithResult 的Entryif !r.appendEntry(m.Entries...) { return ErrProposalDropped } // 广播给 follower 节点处理r.bcastAppend() return nil
stepLeader 匹配 MsgProp 消息，并且根据消息中的 Entry.Type 执行不同的处理逻辑。对于 EntryNormal 的 Entry 将调用 raft.appendEntry 处理。
func (r *raft) appendEntry(es ...pb.Entry) (accepted bool) { // 调用 raft.raftLog.lastIndex 获取最近的 indexli := r.raftLog.lastIndex() // 更新 Entry 的 Term 和 Indexfor i := range es { es[i].Term = r.Term es[i].Index = li + 1 + uint64(i) }// 将 Entry 添加到 raft.raftLog// 这里两阶段提交的第一阶段添加的是 raft.raftLog.unstableli = r.raftLog.append(es...) // 获取最新的 last index 并组成 MsgAppResp 的消息发送r.send(pb.Message{To: r.id, Type: pb.MsgAppResp, Index: li}) return true
}func (l *raftLog) lastIndex() uint64 { // 先从 unstable 获取最近的 indexif i, ok := l.unstable.maybeLastIndex(); ok { return i } // 如果 unstable 获取不到最近的 index 就从 raftLog.storage 获取最近的 indexi, err := l.storage.LastIndex() if err != nil { panic(err) // TODO(bdarnell) } return i
}func (l *raftLog) append(ents ...pb.Entry) uint64 { // 如果 Entry 无数据，则返回if len(ents) == 0 { return l.lastIndex() } // 如果最新的 Entry index 小于 raftLog.committed 触发 panic// 正常情况应该是 raftLog.committed 小于插入的 Entry indexif after := ents[0].Index - 1; after < l.committed { l.logger.Panicf("after(%d) is out of range [committed(%d)]", after, l.committed) } // 将 Entry 插入到 raftLog.unstablel.unstable.truncateAndAppend(ents) return l.lastIndex()
}
在 raft.appendEntry 将 Entry 存入 raftLog.unstable 后调用 raft.send 发送类型为 MsgAppResp 的消息：
func (r *raft) send(m pb.Message) {if m.From == None { m.From = r.id }if m.Type == pb.MsgVote || m.Type == pb.MsgVoteResp || m.Type == pb.MsgPreVote || m.Type == pb.MsgPreVoteResp { ...}// m.Type 为 MsgAppResp 匹配当前分支if m.Type == pb.MsgAppResp || m.Type == pb.MsgVoteResp || m.Type == pb.MsgPreVoteResp {r.msgsAfterAppend = append(r.msgsAfterAppend, m) ...
}
raft.send 方法处理 MsgAppResp 类型的消息是往 raft.msgsAfterAppend 数据中存入该消息。
存入 raft.msgsAfterAppend 后 node.run 将执行结果传递到 msgWithResult.result 通道内，该结果会被 node.Propose 消费：
func (n *node) run() {...select { case pm := <-propc: m := pm.m m.From = r.id err := r.Step(m) if pm.result != nil { pm.result <- err close(pm.result) }...}...
}func (n *node) stepWithWaitOption(ctx context.Context, m pb.Message, wait bool) error { ... select { case err := <-pm.result: // 消费 pm.result 并返回if err != nil { return err } case <-ctx.Done(): return ctx.Err() case <-n.done: return ErrStopped } return nil
}
1.2.2.4 ready 消息
通过上一节的描述 node.Propose 的交互就完成了。node.Propose 作为算法层的入口将 Propose 请求封装成消息发送给 node.run 处理，node.run 将消息（Entry）添加到 raftLog.unstable 中暂存，然后将请求数据封装为 MsgAppResp 类型的消息添加到 raft.msgsAfterAppend 数组。
本小节继续看这里为什么要添加到 raft.msgsAfterAppend 数组。
进入 node.run ：
// node.run 是一个不退出循环
func (n *node) run() { var propc chan msgWithResult var readyc chan Ready var advancec chan struct{} var rd Ready r := n.rn.raft lead := None for { // 进入 node.rn.HasReadyif advancec == nil && n.rn.HasReady() { rd = n.rn.readyWithoutAccept() readyc = n.readyc }...}...
}func (rn *RawNode) HasReady() bool { ...// 通过 raft.msgAfterAppend 判断是否 readyif len(r.msgs) > 0 || len(r.msgsAfterAppend) > 0 { return true } ... return false
}
可以看到，RawNode.HasReady 通过 raft.msgAfterAppend 判断是否 ready。并且，这里的 advancec 通道并未创建，为 nil。此时，程序会走到 node.rn.readyWithoutAccept：
func (rn *RawNode) readyWithoutAccept() Ready { r := rn.raft rd := Ready{ // 获取 raft.raftLog.unstable 中的 EntryEntries: r.raftLog.nextUnstableEnts(), CommittedEntries: r.raftLog.nextCommittedEnts(rn.applyUnstableEntries()), Messages: r.msgs, } ... return rd
}
node.rn.readyWithoutAccept 主要是构造 Ready 对象，接着在 node.run 中将该对象发到 readyc 通道：
func (n *node) run() { ...for {select {case readyc <- rd: // 将 ready 发送到 readyc 通道n.rn.acceptReady(rd) if !n.rn.asyncStorageWrites { advancec = n.advancec // 这里 node.rn.asyncStorageWrites 为 false，会走到这里} else { rd = Ready{} } readyc = nil // 将 readyc 通道关闭，只读不写}
}
在发送 ready 到 readyc 后 node.run 会继续往下处理。此时，readyc 通道另一端也在消费 ready。
我们先往下走，看发完 ready 后 node.run 做了什么？
进入 n.rn.acceptReady:
func (rn *RawNode) acceptReady(rd Ready) {...// 这里异步写为 falseif !rn.asyncStorageWrites {... // needStorageAppendRespMsg 判断 raftLog.unstable 是否有 Entry// 如果有调用 newStorageAppendRespMsg 组消息if needStorageAppendRespMsg(rn.raft, rd) { m := newStorageAppendRespMsg(rn.raft, rd) rn.stepsOnAdvance = append(rn.stepsOnAdvance, m) } // needStorageApplyRespMsg 判断 ready.CommittedEntries 是否有 Entry// 这里在这个阶段无 CommittedEntries，先跳过if needStorageApplyRespMsg(rd) { m := newStorageApplyRespMsg(rn.raft, rd.CommittedEntries) rn.stepsOnAdvance = append(rn.stepsOnAdvance, m) } }...
}// newStorageAppendRespMsg 组类型为 MsgStorageAppendResp 的消息
func newStorageAppendRespMsg(r *raft, rd Ready) pb.Message { m := pb.Message{ Type: pb.MsgStorageAppendResp, To: r.id, From: LocalAppendThread, Term: r.Term, }// 判断 raftLog.unstable 是否有 Entryif r.raftLog.hasNextOrInProgressUnstableEnts() { last := r.raftLog.lastEntryID() m.Index = last.index m.LogTerm = last.term } ... return m
}
基本上发完 ready 后，node.run 会组消息类型为 MsgStorageAppendResp 的消息，并添加到 rn.stepsOnAdvance 数组。
这里留了两个问题：
- 前面发完 ready 后，哪个组件消费 ready？
MsgStorageAppendResp消息发往数组后，在哪里处理的呢？
首先看第一个问题，发完 ready 后哪个组件消费 ready？
1.2.3 应用层处理 ready
应用层的 raftNode.serveChannels 会监听 readyc 通道，消费 ready：
func (rc *raftNode) serveChannels() {...// event loop on raft state machine updates for { select { // store raft entries to wal, then publish over commit channel case rd := <-rc.node.Ready(): // 首先 raftNode 调用预写日志 WAL 将 rd 的状态和数据写入到磁盘日志// 这里写入不成功咋办？好像没有写入不成功的处理？rc.wal.Save(rd.HardState, rd.Entries) // 将 ready.Entries 添加到 raftNode.raftStoragerc.raftStorage.Append(rd.Entries) // 调用 raftNode.node.Advance 方法往 advancec 通道发消息rc.node.Advance() } }
}
raftNode 处理 ready 后，此时的状态是预写日志存储 Entries，raftStorage 添加 Entries。最后往 advancec 通道写数据通知算法层。
可以想到算法层的 node.run 会监听 advancec 通道并处理，我们直接进入此处逻辑：
func (n *node) run() {...for {select {case <-advancec: // 重点看 node.rn.Advancen.rn.Advance(rd) rd = Ready{} advancec = nil}}
}func (rn *RawNode) Advance(_ Ready) { // 在这里对数组做处理，从前面可知此时数组存的是 MsgStorageAppendResp 类型的消息for i, m := range rn.stepsOnAdvance { // 调用 raft.Step 状态机处理消息_ = rn.raft.Step(m) rn.stepsOnAdvance[i] = pb.Message{} } rn.stepsOnAdvance = rn.stepsOnAdvance[:0]
}func (r *raft) Step(m pb.Message) error {...switch m.Type {case pb.MsgStorageAppendResp: if m.Index != 0 { // 进入 raft.raftLog.stableTo 将 raftLog.unstable 的 Entry 清掉// 该 Entry 已经存储到 WAL 和 raftStorage 中了r.raftLog.stableTo(entryID{term: m.LogTerm, index: m.Index}) } ...}return nil
}
raftNode 处理完 ready 消息后，会往 advancec 发消息，通知算法层 ready 消息处理完毕。算法层根据消息类型做不同处理，这里将 raftLog.unstable 中的 Entry 清理掉（这个 Entry 已经存在 raftStorage 和 WAL 中了）。
到这里还没有结束。接下来继续进入 node.run 循环，看算法层和应用层是如何进行第二轮交互的。
1.2.4 算法层和应用层交互
func (n *node) run() {...for { // 在第一轮算法层-应用层交互后，算法层将 advancec 设为 nil// 进入 node.rn.HasReady 判断是否可以组 readyif advancec == nil && n.rn.HasReady() {// 和前面第一轮交互类似，这里是组 ready 消息，后续将 ready 发往 readyc 通道rd = n.rn.readyWithoutAccept() readyc = n.readyc }
}func (rn *RawNode) HasReady() bool {// 这个条件分支将被选中此时的 hardState 和 rn.prevHardState 不一致// hardState 的 Commit 大于 rn.prevHardState，意味着需要 Commit Entryif hardSt := r.hardState(); !IsEmptyHardState(hardSt) && !isHardStateEqual(hardSt, rn.prevHardSt) { return true }
}func (rn *RawNode) readyWithoutAccept() Ready { r := rn.raft // 不同于第一轮交互，这里的 CommittedEntries 是要 Commit 给客户端的存储的 Entry rd := Ready{ Entries: r.raftLog.nextUnstableEnts(), CommittedEntries: r.raftLog.nextCommittedEnts(rn.applyUnstableEntries()), Messages: r.msgs, }...
}
类似于第一轮算法层-应用层交互，这里算法层组 ready 发往应用层。不同的是这里的 Entry 是 CommittedEntry。
算法层发完 ready 后会继续进入 node.run 循环：
func (n *node) run() {for {select {case readyc <- rd:n.rn.acceptReady(rd) if !n.rn.asyncStorageWrites { advancec = n.advancec } else { rd = Ready{} } readyc = nil}}
}func (rn *RawNode) acceptReady(rd Ready) {if !rn.asyncStorageWrites {// 类似于第一轮算法层-应用层交互，这里组消息类型为 MsgStorageApplyResp 的消息加到 rn.stepsOnAdvance，后续算法层收到 advance 信号后会处理if needStorageApplyRespMsg(rd) { m := newStorageApplyRespMsg(rn.raft, rd.CommittedEntries) rn.stepsOnAdvance = append(rn.stepsOnAdvance, m) }}...
}
继续看应用层是如何处理 ready 的：
func (rc *raftNode) serveChannels() {for { select {case rd := <-rc.node.Ready():// 重点在 raftNode.publishEntriesapplyDoneC, ok := rc.publishEntries(rc.entriesToApply(rd.CommittedEntries)) if !ok { rc.stop() return } rc.node.Advance()}
}func (rc *raftNode) publishEntries(ents []raftpb.Entry) (<-chan struct{}, bool) {data := make([]string, 0, len(ents)) for i := range ents { switch ents[i].Type { case raftpb.EntryNormal: // 读取 Commit Entry 中的数据存入 data if len(ents[i].Data) == 0 { // ignore empty messages break } s := string(ents[i].Data) data = append(data, s) }}var applyDoneC chan struct{} if len(data) > 0 { applyDoneC = make(chan struct{}, 1) select { // 将数据和 applyDoneC 通道发给 raftNode.commitCcase rc.commitC <- &commit{data, applyDoneC}: case <-rc.stopc: return nil, false } } // commit 后更新 raftNode.appliedIndex rc.appliedIndex = ents[len(ents)-1].Index return applyDoneC, true
}
应用层收到 ready 后，组 commit 发往 raftNode.commitC 通道。通道的另一头是客户端在消费：
func (s *kvstore) readCommits(commitC <-chan *commit, errorC <-chan error) { for commit := range commitC { if commit == nil { ... continue } for _, data := range commit.data { var dataKv kv dec := gob.NewDecoder(bytes.NewBufferString(data)) if err := dec.Decode(&dataKv); err != nil { log.Fatalf("raftexample: could not decode message (%v)", err) } // 将数据（键值对）存储到 kvstore.kvStore 中s.mu.Lock() s.kvStore[dataKv.Key] = dataKv.Val s.mu.Unlock() } // 写入成功后，关闭 commit.applyDoneC 通知应用层close(commit.applyDoneC) } if err, ok := <-errorC; ok { log.Fatal(err) }
}
客户端会监听 raftNode.commitC 通道，并将通道内的数据编码存储到 kvStore 中。
至此，写入的流程基本结束了。后续就是应用层和算法层的收尾工作。应用层发往 commit 后，发信号到 advancec，通知算法层 commit 已完成，算法层在做相应处理，这里和第一轮交互有点重复，就不赘述了。
2. 小结
可以看到 raft 的写流程是比较复杂的，涉及客户端/应用层/算法层和多个通道的交互，这里只介绍了个大概，部分比较重要的内容如多节点通信等并未介绍，后续会着重开始走读这方面的内容以及学习 raft 的论文加深理解。
这里给出交互流程图如下：
3. 参考资料
- raft 工程化案例之 etcd 源码实现
概述)
