etcd教程(十五)---leader选取源码分析

// raft/raft.go 318行
func newRaft(c *Config) *raft {
	r := &raft{
		id:                        c.ID,
		lead:                      None,
		isLearner:                 false,
		raftLog:                   raftlog,
		maxMsgSize:                c.MaxSizePerMsg,
		maxUncommittedSize:        c.MaxUncommittedEntriesSize,
		prs:                       tracker.MakeProgressTracker(c.MaxInflightMsgs),
		electionTimeout:           c.ElectionTick,
		heartbeatTimeout:          c.HeartbeatTick,
		logger:                    c.Logger,
		checkQuorum:               c.CheckQuorum,
		preVote:                   c.PreVote,
		readOnly:                  newReadOnly(c.ReadOnlyOption),
		disableProposalForwarding: c.DisableProposalForwarding,
	}

    if !IsEmptyHardState(hs) {
		r.loadState(hs)
	}
    
	r.becomeFollower(r.Term, None)
	return r
}

// raft/raft.go 686行
func (r *raft) becomeFollower(term uint64, lead uint64) {
    // 这是一个func，主要是状态机的处理逻辑
	r.step = stepFollower
	r.reset(term)
    // 这就是 选举相关的逻辑
	r.tick = r.tickElection
	r.lead = lead
	r.state = StateFollower
}

// raft/raft.go 645行
func (r *raft) tickElection() {
    // 每次计数加一
    r.electionElapsed++
    // 如果条件允许(并不是所有节点都可以参与Leader选举的)，并且已经超时，那么就开始选举
    if r.promotable() && r.pastElectionTimeout() {
        r.Step(pb.Message{From: r.id, Type: pb.MsgHup})
    }
}
// raft/raft.go 1714行
func (r *raft) pastElectionTimeout() bool {
 // 超时时间到了就可以开始去选举Leader了
 return r.electionElapsed >= r.randomizedElectionTimeout
}
// raft/raft.go 1718行
func (r *raft) resetRandomizedElectionTimeout() {
    // 再固定的超时时间上增加一个随机值，避免出现所有节点同时超时的情况
 r.randomizedElectionTimeout = r.electionTimeout + globalRand.Intn(r.electionTimeout)
}

// raft/raft.go 847行
func (r *raft) Step(m pb.Message) error {
  switch m.Type {
        case pb.MsgHup:
            if r.preVote { // 如果开启了预选机制，则进入预选流程
                r.hup(campaignPreElection)
            } else { // 否则直接进入选举流程
                r.hup(campaignElection)
            }        
    }
}

// raft/raft.go 760行
func (r *raft) hup(t CampaignType) {
	if r.state == StateLeader { // 已经是 Leader 就不能再发起选举了
		return
	}

	if !r.promotable() { // 不满足参与 Leader 选举的条件也是不让选
		return
	}
    // 如果还有配置修改没有应用，也不能选
	ents, err := r.raftLog.slice(r.raftLog.applied+1, r.raftLog.committed+1, noLimit)
	if n := numOfPendingConf(ents); n != 0 && r.raftLog.committed > r.raftLog.applied {
		return
	}
    // 到此，正式开始选举
	r.campaign(t)
}

// raft/raft.go 785行
func (r *raft) campaign(t CampaignType) {
	var term uint64
	var voteMsg pb.MessageType
    // 根据始预选还是分别处理
	if t == campaignPreElection {
		r.becomePreCandidate() // 切换到 PreCandidate 状态
		voteMsg = pb.MsgPreVote
		term = r.Term + 1
	} else {
		r.becomeCandidate() // 切换到 Candidate 状态
		voteMsg = pb.MsgVote
		term = r.Term
	}
    
	if _, _, res := r.poll(r.id, voteRespMsgType(voteMsg), true); res == quorum.VoteWon {
		// We won the election after voting for ourselves (which must mean that
		// this is a single-node cluster). Advance to the next state.
		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
		}
			r.id, r.raftLog.lastTerm(), r.raftLog.lastIndex(), voteMsg, id, r.Term)

		var ctx []byte
		if t == campaignTransfer {
			ctx = []byte(t)
		}
		r.send(pb.Message{Term: term, To: id, Type: voteMsg, Index: r.raftLog.lastIndex(), LogTerm: r.raftLog.lastTerm(), Context: ctx})
	}
}

if t == campaignPreElection {
    r.becomePreCandidate()
    voteMsg = pb.MsgPreVote
    term = r.Term + 1
} else {
    r.becomeCandidate()
    voteMsg = pb.MsgVote
    term = r.Term
}

// raft/raft.go 695行
func (r *raft) becomeCandidate() {
	r.step = stepCandidate
	r.reset(r.Term + 1)
	r.tick = r.tickElection
	r.Vote = r.id
	r.state = StateCandidate

}
// raft/raft.go 708行
func (r *raft) becomePreCandidate() {
	r.step = stepCandidate
	r.prs.ResetVotes()
	r.tick = r.tickElection
	r.lead = None
	r.state = StatePreCandidate
}

// raft/raft.go 1376行
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 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)
		}
	}
	return nil
}

// raft/raft.go 724行
func (r *raft) becomeLeader() {
    // 成为 Leader 后设置了另一个 tick
    r.tick = r.tickHeartbeat
    // ...
}

// raft/raft.go 657行
func (r *raft) tickHeartbeat() {
	r.heartbeatElapsed++ // 可以看到同样是用的计数来代表时间
	r.electionElapsed++


	if r.state != StateLeader {
		return
	}
	// 如果超了就给 Follower 发一个心跳消息过去
	if r.heartbeatElapsed >= r.heartbeatTimeout {
		r.heartbeatElapsed = 0
		if err := r.Step(pb.Message{From: r.id, Type: pb.MsgBeat}); err != nil {
		}
	}
}

// raft/raft.go 847行
func (r *raft) Step(m pb.Message) error {
	switch m.Type {
	default:
		err := r.step(r, m)
		if err != nil {
			return err
		}
	}
	return nil
}

// raft/raft.go 991行
func stepLeader(r *raft, m pb.Message) error {
	switch m.Type {
	case pb.MsgBeat:
		r.bcastHeartbeat()
		return nil
}

// raft/raft.go 525行
func (r *raft) bcastHeartbeat() {
	lastCtx := r.readOnly.lastPendingRequestCtx()
	if len(lastCtx) == 0 {
		r.bcastHeartbeatWithCtx(nil)
	} else {
		r.bcastHeartbeatWithCtx([]byte(lastCtx))
	}
}

// raft/raft.go 1421行
func stepFollower(r *raft, m pb.Message) error {
	switch m.Type {
	case pb.MsgHeartbeat:
		r.electionElapsed = 0 // 直接把计数清0
		r.lead = m.From
		r.handleHeartbeat(m)  // 然后回复心跳消息
}
// raft/raft.go 1513行
func (r *raft) handleHeartbeat(m pb.Message) {
	r.raftLog.commitTo(m.Commit)
	r.send(pb.Message{To: m.From, Type: pb.MsgHeartbeatResp, Context: m.Context})
}