Raft启动后运行期间主要执行两个函数；
1、状态监测和转化
func (rf *Raft) Loop() {
// Set out as a follower
TimeOutConst := 0
for {
TimeOutConst = ElectionTimeoutConst()
if rf.state == "follower" {
// DO FOLLOWER STUFF
select {
case <-rf.heartbeatCH:
case <-time.After(time.Duration(TimeOutConst) * time.Millisecond):
if rf.state != "leader" {
rf.mu.Lock()
// println(strconv.Itoa(rf.me) + " panic, term: " + strconv.Itoa(rf.CurrentTerm) + " len(rf.Logs): " + strconv.Itoa(len(rf.Logs)) + " voteCount: " + strconv.Itoa(rf.voteCount))
rf.state = "candidate"
rf.mu.Unlock()
}
}
} else if rf.state == "candidate" {
// DO CANDIDIATE STUFF
rf.CandidateState()
} else {
// DO LEADER STUFF
rf.LeaderState()
}
}
}
2、如果自己是一个leader，把client传过来的command编程log commit到各个server(通过RequestAppendEntries)和自己之后，向客户端返回结果表示 当前command已经被整个集群正式序列化
func (rf *Raft) FeedStateMachine(applyCh chan ApplyMsg) {
for {
if rf.state == "leader" {
rf.UpdateCommit()
}
time.Sleep(30 * time.Millisecond)
if rf.LastApplied < rf.CommitIndex {
go func() {
rf.mu.Lock()
oldApplied := rf.LastApplied
commitIdx := rf.CommitIndex
rf.LastApplied = commitIdx
rf.mu.Unlock()
if len(rf.Logs)-1 < commitIdx {
return
}
time.Sleep(10 * time.Millisecond)
for i := oldApplied+1; i <= commitIdx; i++ {
Msg := new(ApplyMsg)
Msg.Index = i
Msg.Command = rf.Logs[i].Command
applyCh <- *Msg
}
}()
}
}
}