彭榮新，茄子科技老司機一枚，長期專注基礎架構領域，對中間件的的研發和治理以及穩定性保障有豐富的實踐經驗。

背景

Consul 是一個非常強大的服務發現和配置管理工具，可以幫助您簡化服務管理流程，提高系統的可用性和可擴展性，是目前非常流行的服務發現和配置管理系統，支持高可用，可擴展，多數據中心的分布式系統，是很多公司的基礎實施組件，這些架構的優點的背後是基于分布式協議raft的實現，raft協議的理論有很多，以前需要根據paxos來實現，像zk自己實現了一套zap的協議來實現數據的複制和一致性，hdfs的namenode 日志高可用也是基于paxos實現，技術發展就是快，現在要實現一致性，高可靠，多副本基本上都是采用raft協議來實現，真正講raft 日志複制實現的比較少，比如k8s用的etcd，nacos cp模型也有raft的實現，rocketmq也有raft實現slave選舉，這篇文章主要分享consul raft 協議日志複制的實現原理，嘗試講明白寫日志複制，日志順序，過半提交，一致性檢查等相關的知識點和實現原理。

Consul Raft

raft 算法主要包含兩個部分，分別是leader選舉和日志複制，leader選舉我們不分析，我們主要分析日志複制的實現原理，下面我們以consul的key value 存儲的寫場景入手，一步步分析寫請求的實現邏輯，是怎麽實現raft 日志複制保證一致性的, 內容會比較長，會涉及到如下知識點：

客戶端發起請求

Server端接受請求

拆包處理

心跳機制

批量發送

過半提交

一致性檢查

總結

Consul Agent 請求

客戶端發起一個put key value的http請求，由kvs_endpoint.go 的KVSEndpoint func 處理，put的方法會路由給KVSPut 處理，除了一些校驗外和請求標識，比如是否有獲取鎖acquire或者release，這裏提下一個檢查，就是value的大小檢查，和web 容器一樣檢查防止請求數據太大，可以通過參數kv_max_value_size 控制，如果超過返回狀態碼413，標准的http 狀態碼。

檢查都OK後，consul agent就開始請求consul server了，當然還是rpc 操作

// Copy the valuebuf := bytes.NewBuffer(nil)// 這裏才開始讀請求的數據。if _, err := io.Copy(buf, req.Body); err != nil { return nil, err}applyReq.DirEnt.Value = buf.Bytes()// Make the RPCvar out bool// 開始請求serverif err := s.agent.RPC("KVS.Apply", &applyReq, &out); err != nil { return nil, err}// Only use the out value if this was a CAS// 沒有出錯的話，這裏就成功返回了if applyReq.Op == api.KVSet { return true, nil}

agent請求server時，會默認請求server list的第一個節點，只有在失敗的請求下，回滾動節點，把失敗的添加到最後，原來的第二個節點做爲第一個節點，請求的是consul 下面的kvs_endpoint.go 下面的Apply 方法，所以我們的重點要來了

Server Apply

consul server的 apply方法，代碼還是show下，這裏還有兩個邏輯說明下。

// Apply is used to apply a KVS update request to the data store.func (k *KVS) Apply(args *structs.KVSRequest, reply *bool) error { // 檢查機房dc是否匹配，不是就轉發到對應到dc的server。 if done, err := k.srv.forward("KVS.Apply", args, args, reply); done { return err } // 中間不重要的去了，省得太多... // 對權限token 應用ACL policy ok, err := kvsPreApply(k.logger, k.srv, authz, args.Op, &args.DirEnt) if err != nil { return err } if !ok { *reply = false return nil } // Apply the update. // 這裏是開啓raft 算法的之旅的入口。 resp, err := k.srv.raftApply(structs.KVSRequestType, args) if err != nil { k.logger.Error("Raft apply failed", "error", err) return err } if respErr, ok := resp.(error); ok { return respErr } // Check if the return type is a bool. if respBool, ok := resp.(bool); ok { *reply = respBool } return nil}

在真正開始執行raft 算法前，主要做了如下兩件事：

先檢查了dc是否是當前dc，如果不是會路由到正確的dc，這也是consul 支持多機房部署的一個很好的特性，路由很方便，這也是多機房部署consul是很好的選擇。

檢查是否啓用了acl策略，如果有，需要檢查，沒有對應的token是不能操作的。

leader的寫請求是從kvs的apply方法開始處理請求的，下面我們看下apply 方法的實現邏輯，在真正執行raft前，consul還做了一些加工，不能蠻搞，是非常嚴謹的，上面通過raftApply，經過幾跳後，會執行到raftApplyWithEncoder方法，這裏做的工作是很重要的，所以還是拿出來說下，是漲知識的地方，代碼如下：

// raftApplyWithEncoder is used to encode a message, run it through raft,// and return the FSM response along with any errors. Unlike raftApply this// takes the encoder to use as an argument.func (s *Server) raftApplyWithEncoder(t structs.MessageType, msg interface{}, encoder raftEncoder) (interface{}, error) { if encoder == nil { return nil, fmt.Errorf("Failed to encode request: nil encoder") } // 對請求編碼。 buf, err := encoder(t, msg) if err != nil { return nil, fmt.Errorf("Failed to encode request: %v", err) } // Warn if the command is very large if n := len(buf); n > raftWarnSize { s.rpcLogger().Warn("Attempting to apply large raft entry", "size_in_bytes", n) } var chunked bool var future raft.ApplyFuture switch { case len(buf) <= raft.SuggestedMaxDataSize || t != structs.KVSRequestType: //請求的數據大小如果小于512 * 1024 即512k，則做一次log執行。 future = s.raft.Apply(buf, enqueueLimit) default: //超過了512k，則需要分chunk，每個chunk做爲一個log來應用。 chunked = true //這裏就是每個log一次future。 future = raftchunking.ChunkingApply(buf, nil, enqueueLimit, s.raft.ApplyLog) } //阻塞，等待raft協議完成。 if err := future.Error(); err != nil { return nil, err } resp := future.Response() //... return resp, nil}

這裏通過注釋，你也可以看出，主要關心4件事情：

1. 把請求編碼，這個不是我們的重點，後面有時間可以單獨分析。

2. 檢查是否要拆包，是否要拆成多個raft command 來執行，這裏有個參數控制，SuggestedMaxDataSize consul 默認設置是512k，如果超過這個則拆，否則可以一次raft 協議搞定。

3. 有一個超時時間，默認是30秒，後面會用到。

4. 最後事阻塞等待完成，是logfuture。

爲什麽要拆包

這些事raft 算法不會提的，這個事工程實踐才會有的一些優化，此時你也和我一樣，爲啥要做這個優化呢，有什麽好處，解決什麽問題，這是我們做一個架構師必須要有的思考。

consul的官方就給出了解釋，所以閱讀優秀的代碼就是一種享受，看注釋就能知道爲啥這樣做，下面是他們對SuggestedMaxDataSize的注釋：

// Increasing beyond this risks RPC IO taking too long and preventing// timely heartbeat signals which are sent in serial in current transports,// potentially causing leadership instability.SuggestedMaxDataSize = 512 * 1024

理解就是單次log 數據不能太大，太大理解有下面幾個問題：

鏈接池：leader和follower默認最大3個鏈接，但新版本的代碼實現是沒有鏈接用時，因爲日志複制發送會獨占一個鏈接，直到follower應用完成返回成功，才會釋放鏈接，如果發送的數據量大，回影響鏈接釋放的時間，但是目前大版本通過fast path處理以及會新建一個鏈接來發送，就沒有了這個問題。

網絡帶寬：而且並發高的時候，都是批量發送的，還會同時發給多個follower，有可能導致leader的網絡帶寬過大，會影響心跳包出現延遲，另外如果heartbeat包也支持follower commit日志，follower端也會影響heartbeat包的處理。

上面說影響到心跳，那我們總要知道heartbeat是怎麽實現的，看看到底有什麽影響，所以我們把consul的心跳機制實現原理說明下

定時心跳

- raft 心跳理論

把日志提交的兩階段優化爲了一個階段，省去了commit階段，減少了一個rt，提升了吞吐量，爲什麽能這樣優化，是借助下次請求和心跳請求來告訴followe 當前leader的commit index，所以raft 算法認爲心跳包也是會帶上當前commit index 給follower，讓follower可以盡快提交，保持和leader一致，理論是這樣的，但是consul並沒有這樣實現，請繼續看下面

- Consul 心跳實現

但是consul 在實踐的時候並沒有這麽做，也可能是優化了實現，實現邏輯是每個follower有一個獨立的goroutine來負責發送heartbeat，consul leader 給follower發心跳時，只帶了一個當前leader的任期和leader自身的id和地址兩個信息，不帶log 相關的信息，所以follower 處理心跳請求就很簡單，只要更新下心跳時間即可，當然也會檢查任期，這樣就沒有了io請求，就能快速響應，也叫fast path，就時直接在io線程處理了，因爲follower 處理正常的rpc請求和心跳請求經過decode後，都會統一有follower的main goroutine來處理，如果有一個log append的rpc請求很大，即io操作會大，需要持久化log，很定影響後面的請求，即會影響到心跳請求，follower認爲在心跳超時時間內沒有收到心跳，則認爲leader出了問題，會觸發選舉，就影響了leader穩定。

- CommitTimeout

consul 沒有通過心跳機制來讓follower盡快和leader保持一致來commit log 到fsm狀態機，如果寫不連續，那最近一次寫follower就會一直不提交，本來是發心跳給follower時會讓follower提交，但現在心跳不幹這個活了，所以consul 需要一個新的機制來保證即使沒有新的寫請求的情況下，讓follower也盡快和leader保持一致的commit log，這個機制就是commit timeout隨機時間，每到了一個時間，consul leader就發一個

日志複制的請求給follower，該rpc請求除了帶上leader任期term和標識信息外，還會告訴follower leader的commitindex，follower 就能比較自己的commitindex，如果小于，則進行提交的流程，把沒有應用到狀態機的log commit掉。

批量發送

說完了拆包優化邏輯後，我們看下ApplyLog的邏輯，代碼如下：

// ApplyLog performs Apply but takes in a Log directly. The only values// currently taken from the submitted Log are Data and Extensions.func (r *Raft) ApplyLog(log Log, timeout time.Duration) ApplyFuture { metrics.IncrCounter([]string{"raft", "apply"}, 1) var timer <-chan time.Time if timeout > 0 { timer = time.After(timeout) } // Create a log future, no index or term yet logFuture := &logFuture{ log: Log{ Type: LogCommand, Data: log.Data, Extensions: log.Extensions, }, } logFuture.init() select { case <-timer: return errorFuture{ErrEnqueueTimeout} case <-r.shutdownCh: return errorFuture{ErrRaftShutdown} case r.applyCh <- logFuture: return logFuture }}

這裏主要關心這個applyCh channel，consul 在初始化leader的時候給創建的一個無緩沖區的通道，所以如果leader的協程在幹其他的事情，那這個提交log就阻塞了，時間最長30s，寫入成功，就返回了logFuture，也就事前面我們看到future的阻塞。

到這裏整個consul leader server的插入請求從接受到阻塞等待的邏輯就完成了，consul leader server 有個核心的go routine 在watch 這個applyCh，從定義可以看出，是應用raft log的channel，阻塞在applych 的go routine 代碼如下：

case newLog := <-r.applyCh://這個是前面我們提交log future的 if r.getLeadershipTransferInProgress() { r.logger.Debug(ErrLeadershipTransferInProgress.Error()) newLog.respond(ErrLeadershipTransferInProgress) continue } // Group commit, gather all the ready commits ready := []*logFuture{newLog}GROUP_COMMIT_LOOP: for i := 0; i < r.conf.MaxAppendEntries; i++ { select { case newLog := <-r.applyCh: ready = append(ready, newLog) default: break GROUP_COMMIT_LOOP } } // Dispatch the logs if stepDown { // we're in the process of stepping down as leader, don't process anything new //如果發現我們不是leader了，直接響應失敗 for i := range ready { ready[i].respond(ErrNotLeader) } } else { r.dispatchLogs(ready) }

這裏的一個重要的點就是組發送請求，就是讀applyCh的log，這個裏做了組提交的優化，最多一次發送MaxAppendEntries個，默認位64個，如果並發高的情況下，這裏是能讀到一個batch的，在網絡傳輸和io操作，分組提交是一個通用的優化技巧，比如dubbo在rpc網絡發送，rocketmq，mysql innodb log提交都用了分組提交技術來充分利用網絡io帶寬，減少網絡來回或者io次數的開銷，因爲一次大概率是用不完網絡或者io帶寬的，就像高速4車道的，我們可以一次發四個，而不是一個一個的發。

分組好了後，下面就開始dispatch log了，代碼如下：

// dispatchLog is called on the leader to push a log to disk, mark it// as inflight and begin replication of it.func (r *Raft) dispatchLogs(applyLogs []*logFuture) { now := time.Now() defer metrics.MeasureSince([]string{"raft", "leader", "dispatchLog"}, now) //獲取當前leader的任期編號，這個不會重複是遞增的，如果有心的leaer了，會比這個大。 term := r.getCurrentTerm() //log 編號，寫一個加1 lastIndex := r.getLastIndex() n := len(applyLogs) logs := make([]*Log, n) metrics.SetGauge([]string{"raft", "leader", "dispatchNumLogs"}, float32(n)) //設置每個log的編號和任期 for idx, applyLog := range applyLogs { applyLog.dispatch = now lastIndex++ applyLog.log.Index = lastIndex applyLog.log.Term = term logs[idx] = &applyLog.log r.leaderState.inflight.PushBack(applyLog) } // Write the log entry locally // log先寫入本地持久化，consul大部分的版本底層用的是boltdb，boltdb // 是一個支持事物的數據庫，非常方便，這裏會涉及io操作。 if err := r.logs.StoreLogs(logs); err != nil { r.logger.Error("failed to commit logs", "error", err) //如果寫失敗，則直接響應，前面的future阻塞就會喚醒。 for _, applyLog := range applyLogs { applyLog.respond(err) } //更新自己爲follower r.setState(Follower) return } //這裏很重要，好就才看明白，這個是log 複制成功後，最終應用到狀態機的一個機制 //這裏是記錄下leader自己的結果，因爲過半leader也算一份。 r.leaderState.commitment.match(r.localID, lastIndex) // Update the last log since it's on disk now // 更新最新log entry的編號，寫到這裏了。 r.setLastLog(lastIndex, term) // Notify the replicators of the new log // 開始異步發送給所有的follower，這個leader主go routine的活就幹完了。 for _, f := range r.leaderState.replState { asyncNotifyCh(f.triggerCh) }}

這個dispatchlog的邏輯注釋裏基本寫清楚了，核心的go routine 經過一頓操作後，最主要就是兩點：

- 本地持久化log

consul log持久化是通過boltdb來存儲的，boltdb可以看做一個簡單版的innodb實現，是一個支撐事務和mvcc的存儲引擎, consul 新版本自己實現了log持久化通過wal的方式，可以配置。

- 爲過半成功增加一次記錄，記錄自己寫成功，因爲計算過半時，leader自己這一份也算在裏面，這個很重要。

又異步交給了replicate go routine來處理，他就去繼續去分組提交了，大概率如此循環往複，不知疲倦的給replication routine 派活。

複制GoRoutine

replication routine 會監聽triggerCh channel，接受領導的任務，這個比較簡單，就開始真正發給各自的follower了，代碼如下：

case <-s.triggerCh: lastLogIdx, _ := r.getLastLog() //這個後面沒有異步了，就是這個rpc調用，判斷 shouldStop = r.replicateTo(s, lastLogIdx)

replicateTo 就是rpc調用follower，真正遠程rpc給follower，等待響應，這裏consul 爲保障follower完全和leader的日志一致，需要做有序檢查，所以consul leader 在replicate log給follower時，有一個細節要注意下，就是leader 除發當前操作的log entry，還需要帶上上一條log entry，每條log entry的有兩個關鍵變量：

log.Term 是log所屬的leader的任期。

log.Index 是log的編號。

這裏要發送的日志是就是獲取當前最大的log index即lastIndex 做爲最大值，然後每個follower維護一個nextIndex,即從那裏開始讀log，replication goroutine 會從存儲裏獲取nextIndex-->lastIndex 的log，這裏可能涉及到io操作，就是把前面的持久化的log，再批量讀出來，nextIndex是在複制給follower成功後，會吧lastIndex+1 來更新nextIndex，下次就從新的地方開始讀了。

除了log外，還會帶上leader當前的CommitIndex，即leader已經應用到狀態機FSM的log 索引，follower通過這個來比較，判斷自己是否要提交log。

follower 節點通過這兩個變量來匹配log是否一致，下面log一致性檢查會說明具體怎麽用，也會說明爲啥要發前面一天log。

過半提交

raft協議要求寫操作，只有超過一半才能算成功，才能應用到狀態機FSM, 客戶端才能讀到這個數據，這個過半是leader自己也算在裏面的，也就是前面一篇文章我們提到的，leader在持久化log後，就標記自己寫成功了，我們沒有分析，現在我們來分析下這個邏輯，因爲follower 處理完日志複制後，也是有這個邏輯處理的。

//這裏很重要，好就才看明白，這個是log 複制成功後，最終應用到狀態機的一個機制//這裏是記錄下leader自己的結果，因爲過半leader也算一份。r.leaderState.commitment.match(r.localID, lastIndex)

我們上篇文章只是在這裏做了一個注釋，並沒有分析裏面怎麽實現的，我們就是要搞懂到底怎麽實現的，下面是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() }}

注釋也基本說明了這個方法的作用，就是我們上面說的，我們就不再重複了，要理解這個邏輯，先了解下這個數據結構matchIndexes，matchIndexes 是一個map，key就是server id，就是consul 集群每個節點有一個id，value就是上次應用log到狀態機的編號commitIndex，recalculate的代碼如下：

// 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) } //這個排序是降序，才能保證下面取中間索引位置的值來判斷是否過半已經複制成功。 sort.Sort(uint64Slice(matched)) quorumMatchIndex := matched[(len(matched)-1)/2] //如果超過一半的follower成功了，則開始commit，即應用到狀態機 if quorumMatchIndex > c.commitIndex && quorumMatchIndex >= c.startIndex { c.commitIndex = quorumMatchIndex //符合條件，觸發commit，通知leader執行apply log asyncNotifyCh(c.commitCh) }}

這個recalculate的邏輯單獨看有點晦澀，先不急于理解，先舉一個例子來說明下recalculate的邏輯：

假如集群三個節點，server id分別爲1,2,3，上次寫log的編號是3，就是leader和follower都成功了，這個matchIndexes的數據如下：

1(leader) --> 3

2(follower) --> 3

3(follower) --> 3

假如這個時候新來一個put請求，leader本地持久化成功，就要更新這個數據結構了matchIndexes了, 因爲leader是先更新，再並發請求follower的，所以這個時候matchIndexes數據如下，因爲一個log，所以logIndex是加1。

1(leader) --> 4

2(follower) --> 3

3(follower) --> 3

因爲leader本地完成和follower遠程完成一樣，都要通過這個邏輯來判斷是否commit 該log 請求，即是否應用到FSM，所以就是要判斷是否過半完成了，邏輯是這樣的：

1. 先創建一個數組matched，長度爲集群節點數，我們的例子是3，

2. 然後把matchIndexes的commitIndex 起出來，放到matched中，matched的數據就是[4,3,3]

3. 排序，爲啥要排序，因爲map 是無序的，下面要通過中間索引的值來判斷是否變化。

4. 然後計算 quorumMatchIndex := matched[(len(matched)-1)/2]，這個就是取中間索引下標的值，也是因爲這點，需要第三步排序.

5. 比較quorumMatchIndex 是否大于當前的commitIndex,如果大于，說明滿足過半的條件，則更新，然後應用到狀態機。

通過上面5步，來實現了一個過半的邏輯，我們再以兩個場景來理下，

假如一個follower失敗了，一個成功，成功的follower會更新matched的數據是[4,3,4],或者是[4,4,3],排序後爲都是[4,4,3], 第4步計算的結果是4大于3，就可以提交了，經過上面的詳細，再回看上面的代碼就好理解了。

一致性檢查：

raft協議日志複制是需要嚴格保證順序的，所以在日志複制的時候follower需要對日志做檢查，主要有兩種情況：

log有gap，比如follower停了一段時間，重新加入集群，這個時候follower的log 編號很多事和leader有差距的，對這種情況，就是日志一致性的保證。

follower之前有其他的leader寫了日志，需要覆蓋，以新的leader爲准。

leader網絡出現問題，集群已經有新的leader，老的leader又活過來後，重新發起日志到follower，這個時候任期比會新的leader小，follower直接返回不處理該請求，老的leader 在檢查響應時，先判斷follower的term是不是比自己大，就停止複制的工作。

每次log append給follower時，follower會把自己當前的logindex 編號和當前leader的任期term返回給leader，leader獲取到對應的編號時，會更新發送logNext，也就是從這裏開始發生日志給follower，就進入重試的的流程，重新發日志。

這裏consul 根據raft協議做了一個優化，raft協議描述的是每次遞減一個logindex 編號，來回確認，直到找到follower匹配的編號，再開始發日志，這樣性能就很差，所有基本上沒有那個分布式系統是那樣實現落地的。

Commit Log

只要超過一半的日志 複制成功，consul 就進入日志commit階段，也就是將修改應用到狀態機，通過recalculate 方法給leader監聽的commitCh 發一個消息，通知leader開始執行apply log 到FSM, leader 的代碼如下：

case <-r.leaderState.commitCh: // Process the newly committed entries //上次執行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 { //應用的邏輯在這裏。groupFutures 就是寫入go routine wait的future r.processLogs(lastIdxInGroup, groupFutures) //清理inflight集合中已經commit過的log，防止重複commit for _, e := range groupReady { r.leaderState.inflight.Remove(e) } }

這裏比較簡單，就是從leaderState.inflight 中取出log，就是我們之前寫入的，循環判斷，如果log的編號大于commitIndex,說明是後面新寫入的log，還沒有同步到follower的log，不能提交。這裏應該是有序的，lastIdxInGroup 應該就是需要commit的log的最大的一個編號。

processLogs的邏輯就是支持分批提交支持，發給consul 的runFSM的go routine，consul raft專門有一個go routine來負責commit log到狀態機，支持批量和一個一個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() //將日志應用到FSM的關鍵在這裏。 resp = r.fsm.Apply(req.log) metrics.MeasureSince([]string{"raft", "fsm", "apply"}, start) .... } // Update the indexes lastIndex = req.log.Index lastTerm = req.log.Term}

主要就是三點，應用log 到fsm，然後跟新下fsm的logindex和任期，最後就是要通知還在wait的go routine。

整個日志複制的流程很長，最後再上一張圖總結下整個過程：

總結

這篇文章主要基于consul 目前版本的實現和基于個人的理解，以consul 一個寫請求的整個過程爲線索，介紹了consul 基于raft協議的實現日志複制的基本過程，重點介紹了日志順序保證措施，日志一致性檢查，過半提交log，心跳機制的實現原理，以及相關的幾個優化措施，比如大請求分拆，分組批量發送等一些實踐優化措施，如有不正確的地方，歡迎交流和指正，個人微博@絕塵駒

技術原創及架構實踐文章，歡迎通過公衆號菜單「聯系我們」進行投稿