如何使用raft算法开发强一致kv存储系统

本文内容 ##

本文结合etcd源码来进行介绍，etcd/contrib/raftexample提供了一个基于etcd/raft实现的kv存储系统。从该示例出发，我们来看一看如何基于raft算法开发一个强一致的kv存储系统。

看完本文的源码分析后，上手一个raft强一致系统开发就不是什么难事了。

ps：假定读者已经阅读并理解了raft论文，这里有我的批注版的In Search of an Understandable Consensus Algorithm.pdf，读起来可能会好理解点。

etcd/raft ##

etcd服务端程序入口：see 源码

• 启动过程中区分当前节点类型：根据data-dir目录下的目录名member/proxy/empty来区分，然后启动etcd实例或者proxy；
• 启动etcd服务节点：startEtcd这个函数，逻辑主要包括启动供集群节点间通信的rafthttp服务，以及供客户端请求的服务；
• 启动etcd proxy：startProxy这个函数，逻辑主要是启动etcd代理；

etcd哪些部分值得学习：

• etcd proxy从项目功能上来说虽然很重要，但是从学习角度来说没那么有价值，不看这个；
• etcd server从项目功能上来说是核心，但是我们也没有必要学习所有的请求处理逻辑，重点是关注读写操作时如何基于raft实现强一致；
• raft：这部分是raft算法的核心实现，从理解raft论文到算法工程化需要额外做出巨大的优化，这些知识点往往是通用的，重点掌握；

raft部分：

• pb协议：
  • raft peers的通信协议，see 源码；
  • raft算法中提到核心的几个rpc就是Vote、AppendEntries，但是工程中需要考虑更多，详见上述pb中的enum MessageType；
  • 上述pb中的message Message类型定义了rpc通信过程中的请求/响应，不同rpc通过MessageType type字段区分；
• 状态机：
  • raft实现数据一致性是通过replicated log（复制日志）实现的，这里的replicated log有时也称为WAL（write ahead log）；
  • raft算法中，每个节点raftnode可能处于以下状态中的一种：follower、candidate、precandidate、leader；
  • raft算法中，每个节点的状态可以通过一个状态机来建模；

了解了这些基础知识之后，我们结合etcd/contrib/raftexample来解释下raft如何选举，以及leader遇到写操作如何保证数据强一致。

etcd/raft如何进行leader选举 ##

newRaftNode newRaftNode，see 源码，这个函数包括创建一个var rc raftNode，然后rc.startRaft()，这个函数包含非常重要的几个部分：

• startNode，see 源码，这个主要是建立好raftnode启动时的一些初始状态转换，有一个for事件循环处理，如改变raftnode的状态：tick函数、step函数，以及一些message的处理等等；
• serveRaft：
• serveChannels：see 源码

startNode:

• 如何查看这部分源码呢，首先从启动一个raftnode开始吧：see 源码；
• StartNode函数启动一个raftnode，节点刚启动的时候state都是follower：see 源码；
• StartNode→Bootstrap(peers)通过配置告诉当前raftnode有多少个raftpeers，然后这些raftpeers加入与当前节点所在的集群属于变更配置，也要记录到raftlog中；
• raftnode真正跑起来是在这里：see 源码，这里有个大的for循环，node的主要逻辑都在这里了；
  • tickElection：for/switch-case n.tickC，选举逻辑，此时如果当前raftnode为follower或者candidate吧，此时的tick函数为tickElection，如果选举超时时间过了并且没收到leader的heartbeat来重置选举超时时间，此时会将MsgHup消息类型传入step函数中，将当前follower变为candidate发起选举：see 源码。这里的选举在raft论文中是直接就是选举动作，但是工程上做了优化，引入了一个可选的两阶段选举prevote。虽然可以tick是触发了tickElection，但是这个后续执行中会检查当前节点是否有资格成为leader，不一定有资格（比如自身的WAL不满足条件）。
  • 假如有资格发起选举，则会调用becomeCandidate，会将当前raftnode的term+1，并且step函数变为stepCandidate。然后会调用r.poll来判断是否胜选：see 源码，其实这里是判断的自己给自己投票的话能否胜选，对于single raftnode的集群有用，假如是多节点集群那么这里无法胜选，继续看。ps：如果胜选就becomeLeader成为leader了。如果不是单节点，就要通过r.send发送投票给各个peers：see 源码。这里的r.send并不是真的网络发送，而是记录到r.msgs里面等下处理这里的r.msgs。注意这里r.send的时候已经编程了MsgVote类型了，表示投票请求，后续也应该收到MsgVoteResp。
  • r.msgs什么时候处理呢？还是前面我们提到的这个大循环体，每轮循环都会检查r.msgs中有没有message要处理：see 源码，这里的函数n.rn.HasReady()方法检查到len(r.msgs)>0，则认为有消息要处理，这个消息最终会被包装到一个Ready{}事件中，这个事件会被丢到n.readyc这个chan中，什么时候处理在下面serveChannels中介绍。
  • becomeLeader：see 源码，step函数变为stepLeader，tick函数则变成tickHeartbeat，意味着当前为leader需要给followers定时发送heartbeat来重置它们的选举超时时间，那么heartbeat是什么形式的呢？其实就是通过appendEntry，只不过entry为空，用这种空的entry来表示心跳。leader就要担负起write请求的重任了。
  • 但是如果没胜选的话，raftnode的状态就是candidate，step函数未stepCandidate，下面会继续用到。

serveChannels： 前面关于r.msgs的消息没跟踪到在哪里处理的，我们看下是不是在serveChannels里面？ serveChannels: see 源码

• 这个函数里面也有一个for事件循环，当它发现rc.node.Ready()有var rd Ready{}事件可处理时，如果rd上有非空的snapshot，就写入storage，然后将rd.Entries也记录到rd.HardState，然后将rd.Entries也写入storage，最后将rd.Messages发送到peers。我们感觉voteMsg是在这个时候发送给peers的，到底是不是呢：see 源码。是的，这里的rd事件就是从raftnode.Ready()从发从其raftnode.readyc这个channel中取出来的。取出来后通过transport发送出去，这样voteMsg就发送出去了，那么投票的响应又是什么时候收到、什么时候处理的呢？
• startRaft/AddPeer的时候会调用startPeer，内部会开始循环收包，收peer发来的raftmessage放入一个recvc chan中，startPeer中专门开启了一个goroutine来检查recvc中有没有peer发送来的消息，比如peer发送给我们的voteMsg的响应包：see 源码。这里通过raft.Process(ctx,m)对raftmessage进行处理。如何处理是在这个示例代码中定义的：see 源码，即调用step函数进行处理。我们再往回看下，发送这个消息前已经把节点的step函数修改为了stepCandidate，那我们再看下这个函数里面干了啥，猜测应该有判断是否收到多数投票确认的逻辑；
• stepCandidate：see 源码，我们不考虑可选的prevote阶段，很显然这个消息MsgVote的响应类型应该是MsgVoteResp，如果是的话，就继续r.poll检查下是否胜选吧，如果胜选了，则自己becomeLeader，然后广播appendEntries，这里append是干啥，是为了通治其他peers更新commit index吧。

这样leader选举就完成了！！！

etcd/raft leader执行put操作如何保证强一致 ##

用etcd/raft实现强一致的系统示例：https://github.com/etcd-io/etcd/tree/main/contrib/raftexample。我们不妨从这个项目入手来看下到底是怎么工作的。上述项目是一个暴露http接口的kv强一致存储系统。

接下来重点看一个leader负责执行命令put <key> <value>时的执行逻辑，是怎样的，领略下这个过程中raft扮演的角色。 put命令是通过http put method实现的，see 源码。

这里的处理逻辑也很简单，它直接调用了h.store.Propose(key, string(v))，h.store是一个kvstore，这里的Propose是干嘛呢？这里就可以跟raft算法中的MsgProp关联起来了，还记得吗？MsgProp这种消息类型是用来appendEntries的。这里的思想就是WAL（write ahead log）的思想，先把动作记录到日志中，后面在通过日志来更新状态机。状态机的状态都包含什么呢，我们前面已经知道有各种状态的流转，那么这个日志中记录的数据存储在哪呢？

就是这里的kvstore啊！一个raftnode启动后要把快照、日志中记录的事件还原到一个特定的存储中，这个示例中就是一个内存中的kv数据结构。h.store.Propose(key,string(v))首先异步地调用kvstore.Propose(key,val)将数据写入到proposeC这个chan中：see 源码，然后再异步地从中取出来：see 源码，通过rc.node.Propose(ctx, prop)，转入raft.node.Propose实现：see 源码，这里的n.stepWait方法将MsgProp消息类型以及要写入的日志数据传给stepWait，这里面将消息写入到raftnode.propc就完事返回了。

startPeer从这个propc chan中取出消息m，然后r.Process(ctx, m)去处理，r.Process方法是在示例代码中自定义的，see 源码，通过r.Process进入step函数又来到r.Step(m)，此时raftnode.Step函数是什么呢？赶紧看看发送MsgProp消息时又没有更改raftnode.Step，没有，那这个Step应该是stepLeader…没错，沿着stepLeader一路看下来：see 源码，这里果然是leader让peers appendEntries的动作，干了什么呢？

首先当前raftnode.appendEntries，把MsgProp消息里的日志项（可能有多条）先追加到自己的log entries里面，然后bcastAppend发送给所有的peers让它们去append entries，它们追加成功后肯定回回包MsgPropResp消息类型的消息。我们看看这个消息是在哪里处理的？感觉应该也是在startPeer函数中的收包逻辑里面。那应该也是从recvc chan中取出回包处理。

哈哈，看半天竟然没搜索到MsgPropResp消息类型，前面读源码时有个细节漏掉了。sendAppend的时候实际上会把消息类型改成MsgApp（MessageAppend）去追加日志，followers处理完成后响应一个MsgAppResp消息类型。对于leader raftnode，收到消息后触发状态转换，又要执行其step函数，此时step函数还是stepLeader，发现消息是MsgAppResp，准备处理：see 源码。

我们先考虑正常情况，leader收到响应发现follower在WAL中记录了发送的log entries，leader收到此响应后就会决定是否要更新该follower的next index（下次要发送的log entries开始索引）。然后判断是否可以更新leader的commit index了，更新了之后对client的读请求就可见了。leader更新了commit index之后也要通过bcastAppend通知followers更新commit index。

这些已提交索引之前的log entries会被发布到示例代码中的commitsC chan中，然后有一个goroutine专门读取这上面的commitC并把其中的entries读取出来，应用到我们的kvstore中，这样存储的一些数据就从WAL日志转化为了内存数据结构中的真实数据，可以对外提供查询服务了。

小结 ##

大致就是这些内容吧！感兴趣的继续深挖下raftexample+raft实现吧。感觉自己已经理解了raft的核心思想以及如何使用raft来开发强一致存储系统了，读者是不是也有同感呢:)