背景
上一篇日志復(fù)制我們分析了consul leader 接受一個key value的put請求丑念，leader經(jīng)過一頓操作芳誓，把日志都發(fā)給了follower酬姆，但是還沒有提交咏闪，插入的go routine 也還wait在哪里等是否成功commit了，即應(yīng)用到狀態(tài)機(jī)赴涵，那我們這篇文章就來說一說consul 是怎么判斷過半日志復(fù)制成功的和怎么應(yīng)用到狀態(tài)機(jī)的媒怯。
過半提交
raft協(xié)議要求寫操作，只有超過一半才能算成功句占，才能應(yīng)用到狀態(tài)機(jī)FSM, 客戶端才能讀到這個數(shù)據(jù)沪摄，這個過半是leader自己也算在里面的躯嫉，也就是前面一篇文章我們提到的纱烘，leader在持久化log后，就標(biāo)記自己寫成功了祈餐，我們沒有分析擂啥，現(xiàn)在我們來分析下這個邏輯，因?yàn)閒ollower 處理完日志復(fù)制后帆阳，也是有這個邏輯處理的哺壶。
//這里很重要，好就才看明白蜒谤，這個是log 復(fù)制成功后山宾，最終應(yīng)用到狀態(tài)機(jī)的一個機(jī)制
//這里是記錄下leader自己的結(jié)果，因?yàn)檫^半leader也算一份鳍徽。
r.leaderState.commitment.match(r.localID, lastIndex)
我們上篇文章只是在這里做了一個注釋资锰，并沒有分析里面怎么實(shí)現(xiàn)的，我們就是要搞懂到底怎么實(shí)現(xiàn)的阶祭，下面是match的代碼：
// Match is called once a server completes writing entries to disk: either the
// leader has written the new entry or a follower has replied to an
// AppendEntries RPC. The given server's disk agrees with this server's log up
// through the given index.
func (c *commitment) match(server ServerID, matchIndex uint64) {
c.Lock()
defer c.Unlock()
if prev, hasVote := c.matchIndexes[server]; hasVote && matchIndex > prev {
c.matchIndexes[server] = matchIndex
c.recalculate()
}
}
注釋也基本說明了這個方法的作用绷杜，就是我們上面說的，我們就不再重復(fù)了濒募，要理解這個邏輯鞭盟，先了解下這個數(shù)據(jù)結(jié)構(gòu)matchIndexes，matchIndexes 是一個map瑰剃，key就是server id齿诉，就是consul 集群每個節(jié)點(diǎn)有一個id，value就是上次應(yīng)用log到狀態(tài)機(jī)的編號commitIndex，我們舉一個例子來說明下recalculate的邏輯：
假如集群三個節(jié)點(diǎn)粤剧，server id分別為1,2,3遗座，上次寫log的編號是3，就是leader和follower都成功了俊扳，這個matchIndexes的數(shù)據(jù)如下：
1(leader) --> 3
2(follower) --> 3
3(follower) --> 3
假如這個時候新來一個put請求途蒋，leader本地持久化成功，就要更新這個數(shù)據(jù)結(jié)構(gòu)了matchIndexes了, 因?yàn)閘eader是先更新馋记，再并發(fā)請求follower的号坡，所以這個時候matchIndexes數(shù)據(jù)如下，因?yàn)橐粋€log梯醒，所以logIndex是加1宽堆。
1(leader) --> 4
2(follower) --> 3
3(follower) --> 3
因?yàn)閘eader本地完成和follower遠(yuǎn)程完成一樣，都要通過這個邏輯來判斷是否commit 該log 請求茸习，即是否應(yīng)用到FSM畜隶，所以就是要判斷是否過半完成了，邏輯是這樣的：
先創(chuàng)建一個數(shù)組matched号胚，長度為集群節(jié)點(diǎn)數(shù)籽慢，我們的例子是3，
然后把matchIndexes的commitIndex 起出來猫胁，放到matched中箱亿，matched的數(shù)據(jù)就是[4,3,3]
排序，為啥要排序弃秆，因?yàn)閙ap 是無序的届惋，下面要通過中間索引的值來判斷是否變化。
然后計(jì)算 quorumMatchIndex := matched[(len(matched)-1)/2]菠赚，這個就是取中間索引下標(biāo)的值脑豹，也是因?yàn)檫@點(diǎn)，需要第三步排序.
比較quorumMatchIndex 是否大于當(dāng)前的commitIndex,如果大于衡查，說明滿足過半的條件瘩欺，則更新，然后應(yīng)用到狀態(tài)機(jī)峡捡。
通過上面5步击碗，來實(shí)現(xiàn)了一個過半的邏輯，我們再以兩個場景來理下们拙，
假如一個follower失敗了稍途，一個成功，成功的follower會更新matched的數(shù)據(jù)是[4,3,4],或者是[4,4,3],排序后為都是[4,4,3], 第4步計(jì)算的結(jié)果是4大于3砚婆，就可以提交了械拍，經(jīng)過上面的詳細(xì)突勇，再看下面的代碼就好理解了：
// Internal helper to calculate new commitIndex from matchIndexes.
// Must be called with lock held.
func (c *commitment) recalculate() {
if len(c.matchIndexes) == 0 {
return
}
matched := make([]uint64, 0, len(c.matchIndexes))
for _, idx := range c.matchIndexes {
matched = append(matched, idx)
}
//這個排序是降序，才能保證下面取中間索引位置的值來判斷是否過半已經(jīng)復(fù)制成功坷虑。
sort.Sort(uint64Slice(matched))
quorumMatchIndex := matched[(len(matched)-1)/2]
//如果超過一半的follower成功了甲馋，則開始commit，即應(yīng)用到狀態(tài)機(jī)
if quorumMatchIndex > c.commitIndex && quorumMatchIndex >= c.startIndex {
c.commitIndex = quorumMatchIndex
//符合條件迄损，觸發(fā)commit定躏，通知leader執(zhí)行apply log
asyncNotifyCh(c.commitCh)
}
}
失敗的處理情況
網(wǎng)絡(luò)失敗：
則本次寫就失敗了芹敌，就失敗了痊远，就返回了，這里也沒有看到怎么通知前面的future.wait 退出的氏捞。
數(shù)據(jù)不一致：
如果是log和follower的有g(shù)ap碧聪，比如follower停了一段時間，重新加入集群液茎，這個時候follower的log 編號很多事和leader有差距的逞姿，對這種情況，就是日志一致性的保證捆等。
follower會響應(yīng)leader自己當(dāng)前的logindex 編號滞造，leader獲取到對應(yīng)的編號時，會更新發(fā)送logNext楚里，也就是從這里開始發(fā)生日志給follower断部，就進(jìn)入重試的的流程猎贴，重新發(fā)日志班缎。
這里consul 根據(jù)raft協(xié)議做了一個優(yōu)化，raft協(xié)議描述的是每次遞減一個logindex 編號她渴，來回確認(rèn)达址，直到找到follower匹配的編號，再開始發(fā)日志趁耗，這樣性能就很差沉唠，所有基本上沒有那個分布式系統(tǒng)是那樣實(shí)現(xiàn)落地的。
Commit Log
只要超過一半的日志 復(fù)制成功苛败，consul 就進(jìn)入日志commit階段满葛，也就是將修改應(yīng)用到狀態(tài)機(jī)，通過recalculate 方法給leader監(jiān)聽的commitCh 發(fā)一個消息罢屈，通知leader開始執(zhí)行apply log 到FSM, leader 的代碼如下：
case <-r.leaderState.commitCh:
// Process the newly committed entries
//上次執(zhí)行commit log index
oldCommitIndex := r.getCommitIndex()
//新的log需要commit的log index嘀韧，在判斷是過半時，會更新commitindex
commitIndex := r.leaderState.commitment.getCommitIndex()
r.setCommitIndex(commitIndex)
....
start := time.Now()
var groupReady []*list.Element
var groupFutures = make(map[uint64]*logFuture)
var lastIdxInGroup uint64
// Pull all inflight logs that are committed off the queue.
for e := r.leaderState.inflight.Front(); e != nil; e = e.Next() {
commitLog := e.Value.(*logFuture)
idx := commitLog.log.Index
//idx 大于commitIndex缠捌，說明是后面新寫入的锄贷，還沒有同步到follower的日志。
if idx > commitIndex {
// Don't go past the committed index
break
}
// Measure the commit time
metrics.MeasureSince([]string{"raft", "commitTime"}, commitLog.dispatch)
groupReady = append(groupReady, e)
groupFutures[idx] = commitLog
lastIdxInGroup = idx
}
// Process the group
if len(groupReady) != 0 {
//應(yīng)用的邏輯在這里。groupFutures 就是寫入go routine wait的future
r.processLogs(lastIdxInGroup, groupFutures)
//清理inflight集合中已經(jīng)commit過的log谊却，防止重復(fù)commit
for _, e := range groupReady {
r.leaderState.inflight.Remove(e)
}
}
這里比較簡單柔昼，就是從leaderState.inflight 中取出log，就是我們之前寫入的炎辨，循環(huán)判斷捕透，如果log的編號大于commitIndex,說明是后面新寫入的log，還沒有同步到follower的log碴萧，不能提交激率。這里應(yīng)該是有序的，lastIdxInGroup 應(yīng)該就是需要commit的log的最大的一個編號勿决。
processLogs的邏輯就是支持分批提交支持乒躺，發(fā)給consul 的runFSM的go routine，consul raft專門有一個go routine來負(fù)責(zé)commit log到狀態(tài)機(jī)低缩，支持批量和一個一個commit嘉冒，我們看下單個commit的情況，代碼如下：
commitSingle := func(req *commitTuple) {
// Apply the log if a command or config change
var resp interface{}
// Make sure we send a response
defer func() {
// Invoke the future if given
if req.future != nil {
req.future.response = resp
req.future.respond(nil)
}
}()
switch req.log.Type {
case LogCommand:
start := time.Now()
//將日志應(yīng)用到FSM的關(guān)鍵在這里咆繁。
resp = r.fsm.Apply(req.log)
metrics.MeasureSince([]string{"raft", "fsm", "apply"}, start)
....
}
// Update the indexes
lastIndex = req.log.Index
lastTerm = req.log.Term
}
主要就是三點(diǎn)讳推，應(yīng)用log 到fsm，然后跟新下fsm的logindex和任期玩般，最后就是要通知還在wait的go routine银觅。
log有序
這里還補(bǔ)充下上篇日志復(fù)制一個重要點(diǎn)沒有說明，就是最后發(fā)起日志復(fù)制的時候坏为，因?yàn)閞aft要嚴(yán)格保證log的順序性究驴，所以發(fā)送日志除了當(dāng)前新產(chǎn)生的log 的編號和任期，還需要帶上該log的上一條log的編號和任期匀伏，follower需要檢查這個洒忧，因?yàn)閒ollower只要保證上一條也和leader的上一條匹配，即log 編號和任期term都相同够颠，才可以熙侍，如果有一個不相同，就是follower和leader出現(xiàn)了gap履磨，或者中間有其他leader寫了數(shù)據(jù)蛉抓，需要告訴leader直接覆蓋自己本地log。
總結(jié)
到這里基本上把consul log提交的整個流程摸了一遍剃诅。我們大致盤清楚了consul raft log 從log 復(fù)制到log commit的整個過程巷送，這中間還有一種情況沒有理清楚，就是兩個follower都掛了的情況综苔，前面的寫go routine 的wait的future是怎么通知到的惩系，后面有時間再單獨(dú)開一篇來說明位岔。
目前正在看機(jī)會中，關(guān)注基礎(chǔ)架構(gòu)堡牡，中間件抒抬，架構(gòu)師，技術(shù)經(jīng)理等相關(guān)的機(jī)會晤柄。
