etcd-raft学习

type Message struct {
Type             MessageType `protobuf:"varint,1,opt,name=type,enum=raftpb.MessageType" json:"type"` // 消息类型
To               uint64      `protobuf:"varint,2,opt,name=to" json:"to"` // 消息接收者的节点ID
From             uint64      `protobuf:"varint,3,opt,name=from" json:"from"` // 消息发送者的节点 ID
Term             uint64      `protobuf:"varint,4,opt,name=term" json:"term"` // 发送消息的节点的Term值。如果Term值为0，则为本地消息，在etcd-raft模块的实现中，对本地消息进行特殊处理。
LogTerm          uint64      `protobuf:"varint,5,opt,name=logTerm" json:"logTerm"` // 该消息携带的第一条Entry记录的Term值，日志所处的任期ID
Index            uint64      `protobuf:"varint,6,opt,name=index" json:"index"` // 日志索引ID，用于节点向 Leader 汇报自己已经commit的日志数据ID
Entries          []Entry     `protobuf:"bytes,7,rep,name=entries" json:"entries"` // 如果是MsgApp类型的消息，则该字段中保存了Leader节点复制到Follower节点的Entry记录
Commit           uint64      `protobuf:"varint,8,opt,name=commit" json:"commit"` // 消息发送节点提交日志索引
Snapshot         Snapshot    `protobuf:"bytes,9,opt,name=snapshot" json:"snapshot"` // 在传输快照时，该字段保存了快照数据
Reject           bool        `protobuf:"varint,10,opt,name=reject" json:"reject"` // 主要用于响应类型的消息，表示是否拒绝收到的消息
RejectHint       uint64      `protobuf:"varint,11,opt,name=rejectHint" json:"rejectHint"` //在Follower节点拒绝Leader节点的消息之后，会在该字段记录一个Entry索引值供Leader节点
Context          []byte      `protobuf:"bytes,12,opt,name=context" json:"context,omitempty"` // 消息携带的一些上下文信息。例如，该消息是否与Leader节点转移相关
XXX_unrecognized []byte      `json:"-"`
}

    MsgHup            MessageType = 0  //当Follower节点的选举计时器超时，会发送MsgHup消息
 MsgBeat           MessageType = 1  //Leader发送心跳，主要作用是探活，Follower接收到MsgBeat会重置选举计时器，防止Follower发起新一轮选举
 MsgProp           MessageType = 2  //客户端发往到集群的写请求是通过MsgProp消息表示的
 MsgApp            MessageType = 3  //当一个节点通过选举成为Leader时，会发送MsgApp消息
 MsgAppResp        MessageType = 4  //MsgApp的响应消息
 MsgVote           MessageType = 5  //当PreCandidate状态节点收到半数以上的投票之后，会发起新一轮的选举，即向集群中的其他节点发送MsgVote消息
 MsgVoteResp       MessageType = 6  //MsgVote选举消息响应的消息
 MsgSnap           MessageType = 7  //Leader向Follower发送快照信息
 MsgHeartbeat      MessageType = 8  //Leader发送的心跳消息
 MsgHeartbeatResp  MessageType = 9  //Follower处理心跳回复返回的消息类型
 MsgUnreachable    MessageType = 10 //Follower消息不可达
 MsgSnapStatus     MessageType = 11 //如果Leader发送MsgSnap消息时出现异常，则会调用Raft接口发送MsgUnreachable和MsgSnapStatus消息
 MsgCheckQuorum    MessageType = 12 //Leader检测是否保持半数以上的连接
 MsgTransferLeader MessageType = 13 //Leader节点转移时使用，本地消息
 MsgTimeoutNow     MessageType = 14 //Leader节点转移超时，会发该类型的消息，使Follower的选举计时器立即过期，并发起新一轮的选举
 MsgReadIndex      MessageType = 15 //客户端发往集群的只读消息使用MsgReadIndex消息（只读的两种模式：ReadOnlySafe和ReadOnlyLeaseBased）
 MsgReadIndexResp  MessageType = 16 //MsgReadIndex消息的响应消息
 MsgPreVote        MessageType = 17 //PreCandidate状态下的节点发送的消息
 MsgPreVoteResp    MessageType = 18 //预选节点收到的响应消息  

type node struct {

propc      chan msgWithResult      //该通道用于接收MsgProp类型的消息

recvc      chan pb.Message         //除MsgProp外的其他类型的消息都是由该通道接收的

confc      chan pb.ConfChangeV2    //当节点收到EntryConfChange类型的Entry记录时，会转换成ConfChange，并写入该通道中等待处理。在ConfChange中封装了其唯一 ID、待处理的节点 ID （NodeID 字段）及处理类型（Type 字段，例如，ConfChangeAddNode类型表示添加节点）等信息
confstatec chan pb.ConfState       //在ConfState中封装了当前集群中所有节点的ID，该通道用于向上层模块返回ConfState实例

readyc     chan Ready              //Ready结构体的功能在上一小节已经介绍过了，该通道用于向上层模块返回Ready实例，即node.Ready（）方法的返回值

advancec   chan struct{}           //当上层模块处理完通过上述readyc通道获取到的Ready实例之后，会通过node.Advance（）方法向该通道写入信号，从而通知底层raft实例

tickc      chan struct{}                //用来接收逻辑时钟发出的信号，之后会根据当前节点的角色推进选举计时器和心跳计时器

done       chan struct{}           //当检测到done通道关闭后，在其上阻塞的goroutine会继续执行，并进行相应的关闭操作

stop       chan struct{}           //当node.Stop（）方法被调用时，会向该通道发送信号，在后续介绍中会提到，有另一个goroutine会尝试读取该通道中的内容，当读取到信息之后，会关闭done通道。

status     chan chan Status        //注意该通道的类型，其中传递的元素也是Channel类型，即node.Status（）方法的返回值

 rn        *RawNode

}

//在newRaft（）函数中完成初始化之后，会调用 becomeFollower（）方法将节点切换成 Follower状态，其中会设置raft实例的多个字段
func (r *raft) becomeFollower(term uint64, lead uint64) {
 r.step = stepFollower //设置函数处理Follower节点处理消息的行为
 r.reset(term) //在reset（）方法中会重置raft实例的多个字段
 r.tick = r.tickElection //将tick字段设置成tickElection函数
 r.lead = lead //设置当前节点的leader节点
    //修改当前节点的角色
 r.state = StateFollower
}

//如果当前集群开启了 PreVote 模式，当 Follower 节点的选举计时器超时时，会先调用becomePreCandidate（）方法切换到PreCandidate状态，becomePreCandidate（）
func (r *raft) becomePreCandidate() {
    //检查当前节点的状态，禁止leader直接切换到PreCandidate状态
 if r.state == StateLeader {
  panic("invalid transition [leader -> pre-candidate]")
 }
    //设置函数处理Candidate节点处理消息的行为
 r.step = stepCandidate 
 r.prs.ResetVotes()
 r.tick = r.tickElection
 r.lead = None
    //修改当前节点的角色
 r.state = StatePreCandidate 
}
//当节点可以连接到集群中半数以上的节点时，会调用 becomeCandidate（）方法切换到Candidate状态，becomeCandidate（）
func (r *raft) becomeCandidate() {
 // TODO(xiangli) remove the panic when the raft implementation is stable
 if r.state == StateLeader {
  panic("invalid transition [leader -> candidate]")
 }
    //在reset（）方法中会重置raft实例的多个字段
 r.step = stepCandidate
 r.reset(r.Term + 1) //在reset（）方法中会重置raft实例的多个字段
 r.tick = r.tickElection
 r.Vote = r.id //在此次的选举中，Candidate节点会将选票投给自己
    //修改当前节点的角色
 r.state = StateCandidate
}

//当 Candidate 节点得到集群中半数以上节点的选票时，会调用 becomeLeader（）方法切换成Leader状态，becomeLeader（）
func (r *raft) becomeLeader() {
    //检查当前节点的状态，机制从follower直接切换成leader状态
 if r.state == StateFollower {
  panic("invalid transition [follower -> leader]")
 }
 r.step = stepLeader
 r.reset(r.Term) //在reset（）方法中会重置raft实例的多个字段
 r.tick = r.tickHeartbeat
 r.lead = r.id //将leader字段设置成当前节点的id
 r.state = StateLeader //更新当前节点的角色
    //检查未提交的记录中是否存在多条集群配置变更的Entry记录
 r.prs.Progress[r.id].BecomeReplicate()
 r.pendingConfIndex = r.raftLog.lastIndex()
 emptyEnt := pb.Entry{Data: nil}
    //向当前节点的raftLog中追加一条空的Entry记录
 if !r.appendEntry(emptyEnt) {
    }
 r.reduceUncommittedSize([]pb.Entry{emptyEnt})
}

func (r *raft) tickElection() {
 r.electionElapsed++ //递增electionElapsed计时器

 if r.promotable() && r.pastElectionTimeout() { //检查是否在集群中与检查单签的选举计时器是否超时
  r.electionElapsed = 0
  r.Step(pb.Message{From: r.id, Type: pb.MsgHup}) //发起step处理pb.MsgHup类型消息。
 }
}

func (r *raft) tickHeartbeat() {
 r.heartbeatElapsed++ //递增heartbeatElapsed计时器
 r.electionElapsed++ //递增electionElapsed计时器
 if r.electionElapsed >= r.electionTimeout {
  r.electionElapsed = 0 //重置选举计时器，leader节点不会主动发起选举
  if r.checkQuorum { //进行多数检查
   r.Step(pb.Message{From: r.id, Type: pb.MsgCheckQuorum}) //发起大多数检查。
  }
        //选举计时器处于electionElapsed~randomizedElectionTimeout时段之间时，不能进行leader转移
  if r.state == StateLeader && r.leadTransferee != None {
   r.abortLeaderTransfer() //清空raft.leadTransferee字段，放弃转移
  }
 }
 if r.state != StateLeader { //只有laeder能发送tickHeartbeat
  return
 }
 if r.heartbeatElapsed >= r.heartbeatTimeout { //心跳计时器超时
  r.heartbeatElapsed = 0 //重置心跳计时器
  r.Step(pb.Message{From: r.id, Type: pb.MsgBeat}) //发起step处理MsgBeat类型消息
 }
}

func (r *raft) appendEntry(es ...pb.Entry) (accepted bool) {
 li := r.raftLog.lastIndex()//获取raftLog中最后一条记录的索引值
 for i := range es {//更新待追加记录的Term值和索引值
  es[i].Term = r.Term//Entry记录的Term指定为当前leader节点的任期号
  es[i].Index = li + 1 + uint64(i) //为日志记录指定的Index
 }
 li = r.raftLog.append(es...)//向raft中追加记录
    //更新当前节点对应的Progress,主要是更新Next和Match
 r.prs.Progress[r.id].MaybeUpdate(li)
    //尝试提交Entry记录
 r.maybeCommit()
 return true
}

func (pr *Progress) MaybeUpdate(n uint64) bool {
 var updated bool
 if pr.Match < n {
  pr.Match = n //n之前所有的Entry记录都已经写入对应节点的raftLog中
  updated = true
        //下面将Progress.paused设置为false，表示leader节点可以继续向对应Follower
        //节点发送MsgApp消息
  pr.ProbeAcked()
 }
 pr.Next = max(pr.Next, n+1)//将Next值加一，下一次复制Entry记录开始的位置
 return updated
}

func (r *raft) maybeCommit() bool {
 mci := r.prs.Committed()
 return r.raftLog.maybeCommit(mci, r.Term)
}

func (p *ProgressTracker) Committed() uint64 {
 return uint64(p.Voters.CommittedIndex(matchAckIndexer(p.Progress)))
}
//将node分两个组，JointConfig是大多数的组，有兴趣的看一看quorum包的实现
func (c JointConfig) CommittedIndex(l AckedIndexer) Index {//比较大多数的node的前俩个Index，返回Match的值。
 idx0 := c[0].CommittedIndex(l)
 idx1 := c[1].CommittedIndex(l)
 if idx0 < idx1 {
  return idx0
 }
 return idx1
}
//更新raftLog.committed字段，完成提交
func (l *raftLog) maybeCommit(maxIndex, term uint64) bool {
 if maxIndex > l.committed && l.zeroTermOnErrCompacted(l.term(maxIndex)) == term {
  l.commitTo(maxIndex)
  return true
 }
 return false
}

func (r *raft) Step(m pb.Message) error {
 switch {//首先根据消息的Term值进行分类处理
 case m.Term == 0://本地消息不做处理。MsgHup，MsgProp和MsgReadIndex是本地消息
 case m.Term > r.Term:
 case m.Term < r.Term://细节部分，可以自己研究源码
 }
 switch m.Type {//根据Message的Type进行分类处理
 case pb.MsgHup://这里针对MsgHup类型的消息进行处理。
  if r.preVote {//检查是不是开启了preVote，如果是开启了先调用raft.hup方法，发起preVote。
  } else {
   r.hup(campaignElection)//下面讲述
  }
 case pb.MsgVote, pb.MsgPreVote: //对MsgVote，MsgPreVote类型的消息进行处理。
  canVote := r.Vote == m.From ||
   (r.Vote == None && r.lead == None) ||
   (m.Type == pb.MsgPreVote && m.Term > r.Term)
  if canVote && r.raftLog.isUpToDate(m.Index, m.LogTerm) {
   r.send(pb.Message{To: m.From, Term: m.Term, Type: voteRespMsgType(m.Type)})
   if m.Type == pb.MsgVote {
    r.electionElapsed = 0
    r.Vote = m.From
   }
  } else {
   r.send(pb.Message{To: m.From, Term: r.Term, Type: voteRespMsgType(m.Type), Reject: true})
  }
 default://对于其他类型的消息处理，对应的node的step函数处理
  err := r.step(r, m)
  if err != nil {
   return err
  }
 }
 return nil
}

func (r *raft) hup(t CampaignType) {
 if r.state == StateLeader {//忽略leader
  return
 }
    //方法会检查prs字段中是否还存在当前节点对应的Progress实例，这是为了监测当前节点是否被从集群中移除了
    if !r.promotable() {
  return
 }
    //获取raftLog中已提交但未应用的Entry记录，异常处理
 ents, err := r.raftLog.slice(r.raftLog.applied+1, r.raftLog.committed+1, noLimit)
 r.campaign(t)
}
func (r *raft) campaign(t CampaignType) {
    //该方法的会发送一条包含Term值和类型
 var term uint64
 var voteMsg pb.MessageType
 if t == campaignPreElection {//切换的目标状态是Precandidate
  r.becomePreCandidate()
  voteMsg = pb.MsgPreVote
        //确定要发送的Term值，这里只是增加了消息的Term值，并未增加raft.term字段的值
  term = r.Term + 1
 } else {//切换的目标状态是Candidate
  r.becomeCandidate()
  voteMsg = pb.MsgVote
        //给raft.Term字段的值，并将当前节点的选票投给自身
  term = r.Term
 }
 if _, _, res := r.poll(r.id, voteRespMsgType(voteMsg), true); res == quorum.VoteWon {
        //当得到足够的选票时，则将PreCandidate状态的节点切换成Candidate状态
        //Candidate状态的节点则切换成Leader状态
  if t == campaignPreElection {
   r.campaign(campaignElection)
  } else {
   r.becomeLeader()
  }
  return
 }
 var ids []uint64
 {
  idMap := r.prs.Voters.IDs()
  ids = make([]uint64, 0, len(idMap))
  for id := range idMap {
   ids = append(ids, id)
  }
  sort.Slice(ids, func(i, j int) bool { return ids[i] < ids[j] })
 }
 for _, id := range ids {//状态切换完成之后，当前节点会向集群中所有节点发送指定类型的消息
  if id == r.id { //跳过当前节点自身
   continue
  }
        var ctx []byte
        //在进行Leader节点转移时，MsgPreVote或MsgVote消息会在Context字段中设置该特殊值
  if t == campaignTransfer {
   ctx = []byte(t)
  }
        //发送指定类型的消息，其中Index和LogTerm分别是当前节点的raftLog
        //最后一条消息的Index值和Term值
  r.send(pb.Message{Term: term, To: id, Type: voteMsg, Index: r.raftLog.lastIndex(), LogTerm: r.raftLog.lastTerm(), Context: ctx})
 }
}