raft 结构
raft.png
网络层
首先etcd最外层有一个网络层，负责与集群其他节点通信或者接受客户端的请求，这里我们主要学习raft模块不详细解读，使用就用网络层来代替。
node
node负责raft于网络层的交互，交互使用go的chan
propc: make(chan msgWithResult), //接收网络层MsgProp类型消息
recvc: make(chan pb.Message), //接收网络层除MsgProp类型以外的消息
confc: make(chan pb.ConfChange),//接收EntryConfChange类型消息比如动态添加节点
confstatec: make(chan pb.ConfState),
readyc: make(chan Ready),//向上层返回 ready
advancec: make(chan struct{}),//上层处理往ready后返回给raft的消息
// make tickc a buffered chan, so raft node can buffer some ticks when the node
// is busy processing raft messages. Raft node will resume process buffered
// ticks when it becomes idle.
tickc: make(chan struct{}, 128),//管理超时的管道
done: make(chan struct{}),
stop: make(chan struct{}),
status: make(chan chan Status),
可以看到定义来许多管道来交互信息
raft
raft主要处理raft算法的实现，和日志的复制。
raft会存储当前节点的id，任期，投票等。
其中raftlog用来管理日志，可以看到raftlog下有两个类。
一个日志的流程是这样的：
- 客户端发送请求，生成日志会先进入到unstable模块，如名字所表示，它是不安全的，因为日志还没有进行持久化。
- 将unstable中的日志同步给其他节点，同时交给上层持久化。node也负责和上层通信。
- 如果日志已经被半数以上的节点复制成功了，那么这部分日志将会被认为提交成功。raft会修改raftlog中的committed记录提交的index。
- 提交成功后raft会将提交成功的日志返回给上层，上层会应用日志，然后响应给客户端成功，同时raft也会同步给其他节点让他们也应用日志，然后修改自己的applied记录应用的index。
启动node
创建node
创建node之前先会读取持久化这磁盘中的日志，根据是否有日志和是否是一个新集群，有三种情况
- 没有日志并且是新集群，直接调用startNode
- 没有日志不是新集群，会先与其他节点通信然后更新配置调用straNode
- 有日志，说明是节点重启，读取日志调用restartNode
startNode 和 restartNode其实差别不大，都会构建一个raft的配置
c := &raft.Config{
ID: uint64(id),//当前节点id
ElectionTick: cfg.ElectionTicks,//选举用超时时间
HeartbeatTick: 1,//心跳间隔
Storage: s,//就是MemorySorage
MaxSizePerMsg: maxSizePerMsg, //每次发消息的最大size
MaxInflightMsgs: maxInflightMsgs,
CheckQuorum: true,
PreVote: cfg.PreVote,//上文提到的pre模式
}
除了Storage参数，其他都是一些配置的属性。在straNode中
Storage是调用 s = raft.NewMemoryStorage() 来创建的一个空的初始对象。而restartNode中会用Storage加载从日志中读取的数据。来还原服务宕机前的状态。
s := raft.NewMemoryStorage()
if snapshot != nil {
s.ApplySnapshot(*snapshot)
}
创建raft
接下来调用raft.StartNode(c, peers)来创建raft对象。c就是上面的raft.Config，peers记录集群中所有节点的id。
func StartNode(c *Config, peers []Peer) Node {
r := newRaft(c)
// become the follower at term 1 and apply initial configuration
// entries of term 1
r.becomeFollower(1, None)
for _, peer := range peers {
cc := pb.ConfChange{Type: pb.ConfChangeAddNode, NodeID: peer.ID, Context: peer.Context}
d, err := cc.Marshal()
if err != nil {
panic("unexpected marshal error")
}
e := pb.Entry{Type: pb.EntryConfChange, Term: 1, Index: r.raftLog.lastIndex() + 1, Data: d}
r.raftLog.append(e)
}
// Mark these initial entries as committed.
// TODO(bdarnell): These entries are still unstable; do we need to preserve
// the invariant that committed < unstable?
r.raftLog.committed = r.raftLog.lastIndex()
- 先调用newRaft创建raft对象，newRaft方法中会先Storage中读取信息。然后创建raft对象，然后判断Storage是不是空，如果是重启节点，那这里Storage就会有数据，然后更新raft的数据。
if !isHardStateEqual(hs, emptyState) {
r.loadState(hs)
}
if c.Applied > 0 {
raftlog.appliedTo(c.Applied)
}
/-------------------------------------------/
func (r *raft) loadState(state pb.HardState) {
if state.Commit < r.raftLog.committed || state.Commit > r.raftLog.lastIndex() {
r.logger.Panicf("%x state.commit %d is out of range [%d, %d]", r.id, state.Commit, r.raftLog.committed, r.raftLog.lastIndex())
}
r.raftLog.committed = state.Commit
r.Term = state.Term
r.Vote = state.Vote
/------------------------------------------/
func (l *raftLog) appliedTo(i uint64) {
if i == 0 {
return
}
if l.committed < i || i < l.applied {
l.logger.Panicf("applied(%d) is out of range [prevApplied(%d), committed(%d)]", i, l.applied, l.committed)
}
l.applied = i
}
}
r.loadState(hs) 里面会更新raft记录的任期和投票,raftLog中的提交index。
appliedTo 会一个raftLog的应用index。
之后会调用r.becomeFollower(1, None)修改节点状态变成follower
同样其他状态也有相应的函数：
func (r *raft) becomeFollower(term uint64, lead uint64)
func (r *raft) becomeCandidate()
func (r *raft) becomePreCandidate()
func (r *raft) becomeLeader()
然后如果是新集群，则会将集群的节点信息追加到日志中，并提交追加的日志，这部分用于重启时能从日志中读取到集群的节点信息。
最后创建node 开启协程启动node ,创建node没有什么操作就是创建对象。
n := newNode()
n.logger = c.Logger
go n.run(r)
return &n
run
func (n *node) run(r *raft) {
var propc chan msgWithResult
var readyc chan Ready
var advancec chan struct{}
var prevLastUnstablei, prevLastUnstablet uint64
var havePrevLastUnstablei bool
var prevSnapi uint64
var applyingToI uint64
var rd Ready
lead := None
prevSoftSt := r.softState()
prevHardSt := emptyState
for {
if advancec != nil {
readyc = nil
} else {
rd = newReady(r, prevSoftSt, prevHardSt)
if rd.containsUpdates() {
readyc = n.readyc
} else {
readyc = nil
}
}
if lead != r.lead {
if r.hasLeader() {
if lead == None {
propc = n.propc
} else {
propc = nil
}
lead = r.lead
}
select {
case pm := <-propc:
err := r.Step(m)
case m := <-n.recvc:
r.Step(m)
case cc := <-n.confc:
case <-n.tickc:
r.tick()
case readyc <- rd:
advancec = n.advancec
case <-advancec:
advancec = nil
case c := <-n.status:
c <- getStatus(r)
case <-n.stop:
close(n.done)
return
}
}
}
上面的代码是删除来很多逻辑只保留基本结构的。
主要用于从管道读取消息然后传递给raft。
这里重点就是，raft是不能主动给发消息的，只能是上层应用自己来拉取。
第一次进入循环advancec为空，会调用
newReady(r, prevSoftSt, prevHardSt) 这个函数的返回就是raft的状态变化和要发送的消息。
rd := Ready{
Entries: r.raftLog.unstableEntries(),//unstable中的日志交给上层持久化
CommittedEntries: r.raftLog.nextEnts(),//已经提交待应用的日志，交给上层应用
Messages: r.msgs,//raft要发送的消息
}
node拿到这个对象后会通过readyc通道发送给上层，然后记录这次的状态已用于下次调用Ready的时候判断状态是否变化。
之后会等待advancec通道，如果要消息则说明上次的消息已经处理完成，可以修改自己的状态了，比如日志应用完成修改自己的应用index。
这就是raft启动的过程。下一个文章讲解选举的流程
