前言
选举是Raft实现数据一致性的安全保证，一个raft集群能够正常运行，必须有且仅有一个Leader存在，一次成功选举是集群能够正常运行的前提。Raft协议对选举的定义和安全性保证请参考之前的Raft选举原理解析[传送门]。这篇文章通过解析etcd的源码来看一下Raft集群选举的具体实现。
心跳
Raft集群在正常运行中是不会触发选举的，选举只会发生在集群初次启动或者其它节点无法收到Leader心跳的情况下。初次启动比较好理解，因为raft节点在启动时，默认都是将自己设置为Follower。收不到Leader心跳有两种情况，一种是原来的Leader机器Crash了，还有一种是发生网络分区，Follower跟Leader之间的网络断了，Follower以为Leader宕机了。下面先看一下集群一切正常时，心跳是怎么流转的。
心跳触发
EtcdServer定时触发
Raft集群运行时Leader需要定时的发送心跳给所有Follower。在etcd中，通过EtcdServer中的定时器定时触发Raft模块来实现，这个定时器在raftNode.start()中启动的。
func (r *raftNode) start(rh *raftReadyHandler) {
internalTimeout := time.Second
go func() {
defer r.onStop()
islead := false
for {
select {
//通过go中的ticker触发
case <-r.ticker.C:
r.tick()
case rd := <-r.Ready():
...
...
}
}
...
}()
}
func (r *raftNode) tick() {
r.tickMu.Lock()
r.Tick() //调用node.Tick()
r.tickMu.Unlock()
}
raftNode在启动时会启动一个go routine循环监听ticker定时器的channel，时间到时则调用它的tick()方法。这个ticker定时器的周期即用户设置的raft集群的心跳周期。raftNode中组合了一个raft.Node，所以tick()方法只是加了互斥锁之后就调用的raft.Node.Tick()方法。也就是说触发由EtcdServer来做，逻辑由raft模块来处理。
raft模块处理
Node的Tick()方法只是简单的写个空对象到tickc的channel中，这个前一篇讲过，Node接口的实现类大部分操作都是异步完成的。
func (n *node) Tick() {
select {
//写空对象到tickc channel异步执行
case n.tickc <- struct{}{}:
case <-n.done:
default:
n.rn.raft.logger.Warningf("%x (leader %v) A tick missed to fire. Node blocks too long!", n.rn.raft.id, n.rn.raft.id == n.rn.raft.lead)
}
}
node在启动时会启动一个go routine监听tickc（在node.run()方法中）。收到触发后直接调用的RawNode.Tick()方法，而RawNode又调用了raft.tick()。
//node启动时会在一个新的go routine中运行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
//一直循环查看各个channel
for {
...
...
select {
//监听到触发，则调用RawNode.Tick()
case <-n.tickc:
n.rn.Tick()
...
}
}
}
//RawNode.Tick()
func (rn *RawNode) Tick() {
//直接调用raft.tick()
rn.raft.tick()
}
Leader tick()逻辑
在上一篇文章中讲过，raft的tick属性是函数类型，当节点的角色是Leader时，tick指向的是raft.tickHeartbeat()。
func (r *raft) tickHeartbeat() {
//1. 心跳计数+1
r.heartbeatElapsed++
r.electionElapsed++
// 选举超时控制
if r.electionElapsed >= r.electionTimeout {
r.electionElapsed = 0
if r.checkQuorum {
r.Step(pb.Message{From: r.id, Type: pb.MsgCheckQuorum})
}
// Leader转让
if r.state == StateLeader && r.leadTransferee != None {
r.abortLeaderTransfer()
}
}
//2. 如果当前已经不是Leader了，跳过
if r.state != StateLeader {
return
}
//3. 如果心跳计数大于心跳超时，则发送心跳消息
if r.heartbeatElapsed >= r.heartbeatTimeout {
//心跳计数清0
r.heartbeatElapsed = 0
//调用Step()发送心跳
r.Step(pb.Message{From: r.id, Type: pb.MsgBeat})
}
}
- 第1步首先Leader将心跳计数加1，上一篇讲过，etcd在记录超时时不是以标准时间记录的，而是记录的心跳间隔的倍数。所以EtcdServer每触发一次
tick()，心跳计数+1，代表距离上次发送心跳又过了1个心跳时间 - 第2步关于选举超时和转让的逻辑先跳过
- 第3步中判断是否应该发送心跳，
heartbeatTimeout的意思是Leader应该经过几次心跳时间后必须发送一次心跳。etcd中heartbeatTimeout的默认值是1，也就是说其实每次进到这个方法heartbeatElapsed >= heartbeatTimeout都是成立的。当判断需要发送心跳时，会封装一个MsgBeat的消息提交Step方法处理，处理逻辑下面再说，先看下Follower的tick()方法。
用户也可以把
heartbeatTimeout这个值设的很大，当然这样在Leader宕机到触发重新选举的间隔会长一些。在网络状况不好的时候可以这样设置。
Follower tick()逻辑
当节点角色是Follower或者Candidate的时候，tick指向的是tickElection()
func (r *raft) tickElection() {
//选举计数加1
r.electionElapsed++
//判断是否超时，要发起重新选举
if r.promotable() && r.pastElectionTimeout() {
r.electionElapsed = 0
r.Step(pb.Message{From: r.id, Type: pb.MsgHup})
}
}
以上的逻辑很简单，因为对于Follower来说，唯一需要关心得就是是不是很久都没收到Leader的心跳了。所以每次tick都将选举计数+1，当Follower收到Leader心跳的时候会将electionElapsed清0。如果Follower收不到Leader的心跳，electionElapsed就会一直加到超过选举超时，就发起选举。发起选举的逻辑下面再说。
Leader心跳发送
raft消息封装
上面讲到Leader在收到Tick请求后，会提交一个MsgBeat的消息给到Step()方法，对于心跳消息，会直接调用step指向的函数。跟tick一样,step也是个函数类型，在节点为Leader时，它指向的是stepLeader()，该函数中对于MsgBeat的消息会调用bcastHeartbeat()来给集群中每个Follower发送心跳消息。
func stepLeader(r *raft, m pb.Message) error {
switch m.Type {
//判断对于心跳消息，则广播心跳
case pb.MsgBeat:
r.bcastHeartbeat()
return nil
case pb.MsgCheckQuorum:
...
...
}
func (r *raft) bcastHeartbeat() {
lastCtx := r.readOnly.lastPendingRequestCtx()
if len(lastCtx) == 0 {
r.bcastHeartbeatWithCtx(nil)
} else {
r.bcastHeartbeatWithCtx([]byte(lastCtx))
}
}
// 广播心跳消息至所有节点
func (r *raft) bcastHeartbeatWithCtx(ctx []byte) {
r.prs.Visit(func(id uint64, _ *tracker.Progress) {
//排除Leader自己
if id == r.id {
return
}
//发送心跳
r.sendHeartbeat(id, ctx)
})
}
前一篇讲过raft在prs属性中保存了所有Follower的进度信息，包含Follower的id、同步日志的进度等。所以上面的方法就是遍历prs所有节点，发送心跳消息。
func (r *raft) sendHeartbeat(to uint64, ctx []byte) {
//计算消息中带的commitIndex
commit := min(r.prs.Progress[to].Match, r.raftLog.committed)
//封装成一个Type是MsgHeartbeat的消息，并带上commitIndex
m := pb.Message{
To: to,
Type: pb.MsgHeartbeat,
Commit: commit,
Context: ctx,
}
//发送消息
r.send(m)
}
Raft协议中定义心跳消息和日志消息其实是一个格式的，只是心跳消息没有带日志条目，只会携带CommitIndex。Leader首先看Follower已经接收成功的日志条目的Index，即Progress.Match字段，然后跟自己的CommitIndex比较，取值较小的那个。这是为了防止Follower的日志同步落后太多，CommitIndex处的日志还没有同步到。
封装好消息后，调用send()方法发送。raft本身并不负责消息发送，所以这个方法只是把消息放到一个队列中，等待EtcdServer来获取。
func (r *raft) send(m pb.Message) {
m.From = r.id
//数据校验，选举类消息必须带term属性
if m.Type == pb.MsgVote || m.Type == pb.MsgVoteResp || m.Type == pb.MsgPreVote || m.Type == pb.MsgPreVoteResp {
if m.Term == 0 {
panic(fmt.Sprintf("term should be set when sending %s", m.Type))
}
} else {
//其它类消息不能带term属性
if m.Term != 0 {
panic(fmt.Sprintf("term should not be set when sending %s (was %d)", m.Type, m.Term))
}
//除了日志和MsgReadIndex消息外，设置term为raft当前周期
if m.Type != pb.MsgProp && m.Type != pb.MsgReadIndex {
m.Term = r.Term
}
}
//将消息放入队列
r.msgs = append(r.msgs, m)
}
EtcdServer消息传输
上一步中，心跳消息被放入队列，那这些消息是什么时候被发给集群中其它节点呢？发送操作是由EtcdServer启动时的go routine处理的。具体实现还是在raftNode.start()中。
func (r *raftNode) start(rh *raftReadyHandler) {
internalTimeout := time.Second
go func() {
defer r.onStop()
islead := false
for {
select {
case <-r.ticker.C:
r.tick()
//调用Node.Ready()，从返回的channel中获取数据
case rd := <-r.Ready():
if rd.SoftState != nil {
// SoftState不为空的处理逻辑
}
if len(rd.ReadStates) != 0 {
//ReadStates不为空的处理逻辑
}
// 如果是Leader发送消息给Follower
if islead {
r.transport.Send(r.processMessages(rd.Messages))
}
...
...
//处理完毕调用Advance()方法
r.Advance()
case <-r.stopped:
return
}
}
}()
}
以上的逻辑中只包含心跳相关的，当从Ready channel中读到数据后，直接通过transport发送出去，这里的processMessages()除了对消息封装成传输协议要求的格式，还会做超时控制。
发送完毕后无论成功失败都会调用raft的Advance()方法处理后续逻辑。Leader一次心跳发送就算结束了。
Ready数据获取
上面的逻辑中，心跳的message是被打包在Ready数据结构中返回的，下面看一下数据打包的过程。既然Node.Ready()返回的是个channel，则必然有地方将Ready塞进channel中，这段逻辑是在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 advancec != nil {
readyc = nil
} else if n.rn.HasReady() { //判断是否有Ready数据
// 获取Ready数据
rd = n.rn.readyWithoutAccept()
readyc = n.readyc
}
....
select {
....
case readyc <- rd: //数据放入ready channel中
n.rn.acceptReady(rd) //告诉raft，ready数据已被接收
advancec = n.advancec //赋值Advance channel等待Ready处理完成的消息
}
}
}
上面的代码中Ready数据是通过调用RawNode.readyWithoutAccept()获取到的。
func (rn *RawNode) readyWithoutAccept() Ready {
return newReady(rn.raft, rn.prevSoftSt, rn.prevHardSt)
}
func newReady(r *raft, prevSoftSt *SoftState, prevHardSt pb.HardState) Ready {
rd := Ready{
Entries: r.raftLog.unstableEntries(), //未持久化的日志
CommittedEntries: r.raftLog.nextEnts(), //已提交可以apply的日志
Messages: r.msgs, //raft队列中所有的message
}
//判断softState有没有变化，有则赋值
if softSt := r.softState(); !softSt.equal(prevSoftSt) {
rd.SoftState = softSt
}
//判断hardState有没有变化，有则赋值
if hardSt := r.hardState(); !isHardStateEqual(hardSt, prevHardSt) {
rd.HardState = hardSt
}
//判断是不是收到snapshot
if r.raftLog.unstable.snapshot != nil {
rd.Snapshot = *r.raftLog.unstable.snapshot
}
if len(r.readStates) != 0 {
rd.ReadStates = r.readStates
}
//处理该Ready后是否需要做fsync，将数据强制刷盘
rd.MustSync = MustSync(r.hardState(), prevHardSt, len(rd.Entries))
return rd
}
对于心跳来说，上面最关键的操作就是生成Ready的时候，将msg放到Ready中。
Follower心跳处理
现在来到心跳的接收方，心跳消息到达Follower后，传输层会回调EtcdServer.Process方法，将心跳消息交给raft状态机处理。
func (s *EtcdServer) Process(ctx context.Context, m raftpb.Message) error {
if s.cluster.IsIDRemoved(types.ID(m.From)) {
//发送方已经从集群中移除
}
if m.Type == raftpb.MsgApp {
//收到日志消息记录metrics
}
//调用raft.Step处理消息
return s.r.Step(ctx, m)
}
对于Follower来说Step仍然进入的是stepFollower方法，第一步是将选举计时清0，防止发起选举流程。
func stepFollower(r *raft, m pb.Message) error {
switch m.Type {
...
case pb.MsgHeartbeat:
r.electionElapsed = 0 //选举超时清0
r.lead = m.From //设置Lead为心跳发送方ID
r.handleHeartbeat(m) //处理心跳消息
...
}
return nil
}
func (r *raft) handleHeartbeat(m pb.Message) {
//设置commitIndex为Leader传来的最新值
r.raftLog.commitTo(m.Commit)
//发送Response给Leader
r.send(pb.Message{To: m.From, Type: pb.MsgHeartbeatResp, Context: m.Context})
}
Follower对心跳消息的处理很简单，1）选举超时计时清0；2）设置commitIndex（会检查本地的commitIndex是不是比leader发过来的小）；3）回复Leader，回复的时候按照raft协议的要求带上自己日志的进度。
Leader心跳回复处理
Leader收到Follower的心跳回复后，跟所有消息的处理逻辑一样，会进入stepLeader()方法处理
func stepLeader(r *raft, m pb.Message) error {
switch m.Type {
...
...
case pb.MsgHeartbeatResp:
//记录Follower为Active状态
pr.RecentActive = true
pr.ProbeSent = false
if pr.State == tracker.StateReplicate && pr.Inflights.Full() {
pr.Inflights.FreeFirstOne()
}
//有日志要发送，继续发送
if pr.Match < r.raftLog.lastIndex() {
r.sendAppend(m.From)
}
if r.readOnly.option != ReadOnlySafe || len(m.Context) == 0 {
return nil
}
//处理线性读的逻辑
if r.prs.Voters.VoteResult(r.readOnly.recvAck(m.From, m.Context)) != quorum.VoteWon {
return nil
}
rss := r.readOnly.advance(m)
for _, rs := range rss {
req := rs.req
if req.From == None || req.From == r.id { // from local member
r.readStates = append(r.readStates, ReadState{Index: rs.index, RequestCtx: req.Entries[0].Data})
} else {
r.send(pb.Message{To: req.From, Type: pb.MsgReadIndexResp, Index: rs.index, Entries: req.Entries})
}
}
}
}
Leader收到心跳回复后，会判断是否有新的日志要发给Follower，有的话就继续发送。线性读的逻辑放在后面的文章解析。
选举
正常情况下，Leader在每次tick()方法时发送心跳，网络一切正常，Follower收到心跳后将选举计时清0，集群就这样愉快的运行下去了。但是分布式系统中，意外时必然会发生的，这时候Leader宕机了，就需要集群中其它节点站出来，竞选成为新的Leader。
选举触发
首先来看一下Follower是怎么认定Leader挂了的。当raft节点角色是Follower的时候，EtcdServer每次触发tick()，进入的是tickElection()方法：
func (r *raft) tickElection() {
r.electionElapsed++
//判断是否要发起选举
if r.promotable() && r.pastElectionTimeout() {
r.electionElapsed = 0
r.Step(pb.Message{From: r.id, Type: pb.MsgHup})
}
}
func (r *raft) promotable() bool {
pr := r.prs.Progress[r.id]
return pr != nil && !pr.IsLearner
}
func (r *raft) pastElectionTimeout() bool {
//每次tick加1后和随机选举超时比较
return r.electionElapsed >= r.randomizedElectionTimeout
}
在每次tick()时，都会检查是否符合发起一次新的选举的条件。其中promotable()比较简单，就是判断自己当前是不是还在集群中并且不能是Learner。pastElectionTimeout()判断是否已经超过选举超时时间还没收到Leader的心跳。这里比较的值是randomizedElectionTimeout，代表一个随机选举超时时间，使用随机时间的原因是防止Leader心跳超时后所有Follower同时发起选举。我们看下etcd中这个时间是怎么算的：
func (r *raft) resetRandomizedElectionTimeout() {
r.randomizedElectionTimeout = r.electionTimeout + globalRand.Intn(r.electionTimeout)
}
可以看到这个时间是1个选举超时到2个选举超时之间的随机值。每次开启一个新的term，这个reset方法都会被调用一次，所以在每个选举周期这个随机值都是不同的，最大限度防止重复。
发送选票
状态转换
上一步中Follower发现过了选举超时还没收到Leader心跳，触发Step()方法让raft状态机进行状态转换。
func (r *raft) Step(m pb.Message) error {
...
...
switch m.Type {
case pb.MsgHup:
if r.state != StateLeader {
if !r.promotable() {
//write log
return nil
}
ents, err := r.raftLog.slice(r.raftLog.applied+1, r.raftLog.committed+1, noLimit)
if err != nil {
r.logger.Panicf("unexpected error getting unapplied entries (%v)", err)
}
//判断有配置变更日志，如果集群正在做配置变更，则不发起选举
if n := numOfPendingConf(ents); n != 0 && r.raftLog.committed > r.raftLog.applied {
//write log
return nil
}
//发起选举，根据配置判断选举之前是否要做预投票
if r.preVote {
r.campaign(campaignPreElection)
} else {
r.campaign(campaignElection)
}
} else {
r.logger.Debugf("%x ignoring MsgHup because already leader", r.id)
}
}
Step方法收到要发起选举的消息后（MsgHup），会首先判断已经提交的还没生效的日志中有没有集群变更，有的话说明集群正在变更，则不发起选举。这么做的原因是有可能当前节点在集群变更后已经被从集群中移除了。
然后，根据配置中设置的选举是否需要先预选，来调用campaign()方法发起选举。预选的原理放在本文最后说。
func (r *raft) campaign(t CampaignType) {
var term uint64
var voteMsg pb.MessageType
if t == campaignPreElection {
//需要预选的情况
r.becomePreCandidate()
voteMsg = pb.MsgPreVote
term = r.Term + 1
} else {
//1. 变为候选人，并初始化一条拉票的消息
r.becomeCandidate()
voteMsg = pb.MsgVote
term = r.Term
}
//2.判断是否已经赢得选举
if _, _, res := r.poll(r.id, voteRespMsgType(voteMsg), true); res == quorum.VoteWon {
if t == campaignPreElection {
r.campaign(campaignElection)
} else {
r.becomeLeader()
}
return
}
//3. 收集所有集群中节点id
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] })
}
//4. 轮询所有节点，给除自己之外的节点发送一条拉票的消息
for _, id := range ids {
if id == r.id {
continue
}
var ctx []byte
if t == campaignTransfer {
ctx = []byte(t)
}
//5.根据raft协议的定义组合投票消息
r.send(pb.Message{Term: term, To: id, Type: voteMsg, Index: r.raftLog.lastIndex(), LogTerm: r.raftLog.lastTerm(), Context: ctx})
}
}
上面的代码分为如下几步：
- 在发送投票之前，节点首先将自己的状态置为候选人，这一步会把term加1，然后修改自己的Vote属性为自己的id，表示当前周期选票投给自己。becomeCandidate() 方法如下：
func (r *raft) becomeCandidate() {
if r.state == StateLeader {
panic("invalid transition [leader -> candidate]")
}
r.step = stepCandidate //step方法指向stepCandidate
r.reset(r.Term + 1) //选举周期+1
r.tick = r.tickElection //tick方法指向tickElection
r.Vote = r.id //投票给自己
r.state = StateCandidate //节点状态转为候选人
}
- 第二步的判断是为了兼容单节点集群的场景，不是真正的判断是否收到半数以上选票。对于单个节点，只要自己给自己投票了就已经是Leader了。
3-5. 给集群中所有节点发送选票， 根据raft协议的定义，投票请求需要包含新的选举周期，节点id和最新日志的Index。然后跟其它消息一样调用send方法提交一条消息。
选票发送
消息发送和心跳消息一样，也是放到raft的msg队列中。EtcdServer拿到Ready后发送给集群中其它的节点，整个步骤中没有针对voteMsg做特殊处理。
Follower投票
当候选人节点将选票消息发出以后，在node中会放入recvc，最终会调用raft.Step(m pb.Message)处理这条消息。到Step()方法之前，逻辑跟心跳没有区别，就不重复了，下面看下Step方法的处理。
func (r *raft) Step(m pb.Message) error {
switch {
case m.Term == 0:
// local message
case m.Term > r.Term:
//1. 消息的term比当前节点的大
if m.Type == pb.MsgVote || m.Type == pb.MsgPreVote {
//2. 判断是否是人工操作的强制transfer
force := bytes.Equal(m.Context, []byte(campaignTransfer))
//3. 根据本地记录判断Leader是否已超时
inLease := r.checkQuorum && r.lead != None && r.electionElapsed < r.electionTimeout
if !force && inLease {
// 如果没超时，则直接不回复候选人
return nil
}
}
switch {
case m.Type == pb.MsgPreVote:
// 预选不修改term
case m.Type == pb.MsgPreVoteResp && !m.Reject:
// 预选的response
default:
if m.Type == pb.MsgApp || m.Type == pb.MsgHeartbeat || m.Type == pb.MsgSnap {
//4-1. 如果收到了新的term的心跳、append或者snapshot，代表新的周期开始，自己变成Follower
r.becomeFollower(m.Term, m.From)
} else {
//4-2. 如果收到了候选人 的投票请求，则说明当前进入重新选举阶段，将Leader设置成None
r.becomeFollower(m.Term, None)
}
}
case m.Term < r.Term:
//处理收到的消息term小于当前节点
}
switch m.Type {
case pb.MsgHup:
...
case pb.MsgVote, pb.MsgPreVote:
// 5-1. 判断是否可以投票给候选人
canVote := r.Vote == m.From ||
(r.Vote == None && r.lead == None) ||
(m.Type == pb.MsgPreVote && m.Term > r.Term)
// 5-2. 判断候选人的日志比当前节点的新
if canVote && r.raftLog.isUpToDate(m.Index, m.LogTerm) {
//6-1. 回复候选人同意
r.send(pb.Message{To: m.From, Term: m.Term, Type: voteRespMsgType(m.Type)})
if m.Type == pb.MsgVote {
// 将Vote属性改为候选人的id
r.electionElapsed = 0
r.Vote = m.From
}
} else {
//6-2. 回复候选人不同意
r.send(pb.Message{To: m.From, Term: r.Term, Type: voteRespMsgType(m.Type), Reject: true})
}
default:
...
}
return nil
}
节点收到投票的消息，处理前提是收到的Term比当前的Term要大。
- ，如果是候选人发送的投票消息，首先会做一次校验。1到3步是判断这次投票的消息不是Transfer消息，并且选举也没有超时，则直接忽略掉。再次检查最小选举超时是为了防止集群中只有少部分节点收不到心跳，而其它节点心跳正常的情况，减少重新选举的次数。
Transfer消息是etcd支持的人工发起的Leader转移请求，这是为了在Leader机器性能不够或者准备下线时，人工发起切换Leader
- 收到term比自己大的消息时，有可能是有新的Leader当选了，发送日志或者心跳消息出来。这种情况当前节点无论处于什么状态都应该切换成Follower。如果是候选人的投票消息，则将自己的Leader设置成None，进入选举中阶段。
5-1. 在收到候选人的投票消息后，必须满足3种情况下的一种才可以投同意票，① 之前已经投给这个候选人了，可能由于网络的原因再次收到重复的消息；②当前未给任何节点投过票，而且当前的Leader是None(在低4步中设置的)；③ 预选消息只需要判断term就可以了
5-2.投同意票还有一个条件就是候选人的日志比当前节点的新，raft中新的标准就是最后一条日志要么term更大，要么term相同Index更大
6-1. 第5步的条件都满足后就可以回复同意给候选人了，同时将自己Vote值改为候选人的ID，这一步很关键，保证了同一个term中，只能投票给一个候选人
6-2. 如果第5步中的条件不满足则拒绝候选人
选举完成
对于候选人来说，选票发出去之后无非面临如下2种结果：
- 失败，规定时间内没有收到超过半数同意票
- 成功，规定时间内收到超过半数同意票
处理选票回复
在集群网络正常时，候选人应该很快会收到各个节点对她选票的回复。消息的处理在stepCandidate()函数中：
func stepCandidate(r *raft, m pb.Message) error {
var myVoteRespType pb.MessageType
if r.state == StatePreCandidate {
myVoteRespType = pb.MsgPreVoteResp
} else {
myVoteRespType = pb.MsgVoteResp
}
switch m.Type {
case pb.MsgProp:
//cadidate状态下不接受客户端数据修改请求
return ErrProposalDropped
case pb.MsgApp:
//投票期间收到日志，说明其它节点成为Leader
r.becomeFollower(m.Term, m.From) // always m.Term == r.Term
r.handleAppendEntries(m)
case pb.MsgHeartbeat:
//投票期间收到心跳，说明其它节点成为Leader
r.becomeFollower(m.Term, m.From) // always m.Term == r.Term
r.handleHeartbeat(m)
case pb.MsgSnap:
//投票期间收到日志快照，说明其它节点成为Leader
r.becomeFollower(m.Term, m.From) // always m.Term == r.Term
r.handleSnapshot(m)
case myVoteRespType:
//收到投票反馈，则记录并判断是否超过半数
gr, rj, res := r.poll(m.From, m.Type, !m.Reject)
switch res {
case quorum.VoteWon:
if r.state == StatePreCandidate {
r.campaign(campaignElection)
} else {
//赢得选举，则成为Leader，开始广播心跳和日志
r.becomeLeader()
r.bcastAppend()
}
case quorum.VoteLost:
// 输掉选举，重新变成Follower
r.becomeFollower(r.Term, None)
}
case pb.MsgTimeoutNow:
...
}
return nil
}
当发起选举的节点收到消息时，如果消息是日志、快照或者心跳消息，说明别的节点已经成为Leader，它已经输掉了，则直接成为Follower。
如果收到的是其它节点的投票回复，则会统计自己的选票，如果超过半数支持，是则成为Leader；超过半数拒绝，则回到Follower状态等待下个选举超时。如果都没到，则继续等。
在赢得选举成为Leader的情况下，根据raft协议，需要马上开始发送心跳，以防止其它Follower开始新的选举。
超时失败
除了以上候选人节点在收到明确的消息时，可以判断自己是否成功之外，还有另外一种场景。比如一个节点和其它节点网络状况不佳或者是多个节点同时成为候选人。这种场景下，即不会收到足够的投票，也没有收到别人成为Leader消息，为了防止节点一直等下去，需要一个超时的机制。
etcd在整个选票发送及等待选票的过程中，tick()方法是一直在运行的，如果自己一直没有当选，别人也没当选超时的话。候选人会发起重新一轮的选举，逻辑跟第一轮是一样的。回顾下这个方法，其实无论是处于Follower还是Candidate状态，tick的逻辑是一样的。
func (r *raft) tickElection() {
r.electionElapsed++
if r.promotable() && r.pastElectionTimeout() {
r.electionElapsed = 0
r.Step(pb.Message{From: r.id, Type: pb.MsgHup})
}
}
预选举
什么是预选
etcd中，raft状态机新增了一个状态叫做预候选人。如果用户在启动etcd时配置了PreVote属性为true，则每次选举开始之前，都会先来一轮预选。所谓预选，就是节点在成为正式候选人之前，先发送一个预选的消息给集群内所有节点（MsgPreVote），如果超过半数节点都同意，候选人才会开始一次正式的选举。
在预选阶段，候选节点的状态变为PreCadidate。而其它节点仍然保持原来的状态，也就是说这时候又有其它Follower要发起选举并发送预选请求，其它节点也是会同意的。
当进入预选状态的节点，收到超过半数同意后，则正式进入候选人状态(Candidate)
为什么需要预选
添加预选的原因是为了在网络状况不佳时，减少选举次数。举个具体场景，当集群中的网络不稳定时，会有部分Follower不能及时地收到Leader的心跳，这时候就会有Follower发起选举。但是网络原因，它自身也很难拿到超过半数选票当选，或者当选之后也很快就会有别的节点因为收不到心跳而再次发起选举，这就导致了集群经常处于选举状态而不可用。为了防止这种情况的频繁发生，添加预选阶段，等于把Leader挂掉这件事从单个节点自己判断，变成了半数节点一起判断，大大减少了误判。
凡事都有利有弊，当Leader虽然没挂掉，但性能有问题时，可能只影响了不到一半的节点。添加预选之后可能会导致性能不佳的Leader很难被选下去，从而影响读写性能。
总结
选举是保证raft安全性的基础，心跳是保证集群能够尽快从Leader宕机或者网络分区中恢复的关键。etcd中将心跳和计时做了集成，抽象成tick。tick操作在Leader端用来触发raft定时发送心跳，而在Follower端是为了触发检查Leader是否超时。
Raft模块通过tick操作来触发状态机在不同状态中的转换，通过绑定不同的函数来对消息进行处理和反馈。
选举完成后，etcd就可以通过在集群中复制日志来保证用户对数据读写的分布式保存和一致性保证了。下一篇将重点解析etcd中日志复制和提交生效的过程。
