今天想要介紹的 etcd 其實是在生產環境中經常被使用的協調服務，它與 Zookeeper 一樣，也能夠為整個集群提供服務發現、配置以及分佈式協調的功能。

etcd-logo

這篇文章將會介紹 etcd 的實現原理，其中包括 Raft 協議、存儲兩大模塊，在最後我們也會簡單介紹 etcd 一些具體應用場景。

簡介

etcd 的官方將它定位成一個可信賴的分佈式鍵值存儲服務，它能夠為整個分佈式集群存儲一些關鍵數據，協助分佈式集群的正常運轉。

etcd-keywords

我們可以簡單看一下 etcd 和 Zookeeper 在定義上有什麼不同：

• etcd is a distributed reliable key-value store for the most critical data of a distributed system…
• ZooKeeper is a centralized service for maintaining configuration information, naming, providing distributed synchronization, and providing group services.

其中前者是一個用於存儲關鍵數據的鍵值存儲，後者是一個用於管理配置等信息的中心化服務。

etcd 的使用其實非常簡單，它對外提供了 gRPC 接口，我們可以通過 Protobuf 和 gRPC 直接對 etcd 中存儲的數據進行管理，也可以使用官方提供的 etcdctl 操作存儲的數據。

service KV {
 rpc Range(RangeRequest) returns (RangeResponse) {
 option (google.api.http) = {
 post: "/v3beta/kv/range"
 body: "*"
 };
 }
 rpc Put(PutRequest) returns (PutResponse) {
 option (google.api.http) = {
 post: "/v3beta/kv/put"
 body: "*"
 };
 }
}

文章並不會展開介紹 etcd 的使用方法，這一小節將逐步介紹幾大核心模塊的實現原理，包括 etcd 使用 Raft 協議同步各個節點數據的過程以及 etcd 底層存儲數據使用的結構。

Raft

在每一個分佈式系統中，etcd 往往都扮演了非常重要的地位，由於很多服務配置發現以及配置的信息都存儲在 etcd 中，所以整個集群可用性的上限往往就是 etcd 的可用性，而使用 3 ~ 5 個 etcd 節點構成高可用的集群往往都是常規操作。

etcd-cluste

正是因為 etcd 在使用的過程中會啟動多個節點，如何處理幾個節點之間的分佈式一致性就是一個比較有挑戰的問題了。

解決多個節點數據一致性的方案其實就是共識算法，在之前的文章中我們簡單介紹過 Zookeeper 使用的 Zab 協議 以及常見的 共識算法 Paxos 和 Raft，etcd 使用的共識算法就是 Raft，這一節我們將詳細介紹 Raft 以及 etcd 中 Raft 的一些實現細節。

介紹

Raft 從一開始就被設計成一個易於理解和實現的共識算法，它在容錯和性能上與 Paxos 協議比較類似，區別在於它將分佈式一致性的問題分解成了幾個子問題，然後一一進行解決。

每一個 Raft 集群中都包含多個服務器，在任意時刻，每一臺服務器只可能處於 Leader、Follower 以及 Candidate 三種狀態；在處於正常的狀態時，集群中只會存在一個 Leader，其餘的服務器都是 Follower。

raft-server-states

上述圖片修改自 In Search of an Understandable Consensus Algorithm 一文 5.1 小結中圖四。

所有的 Follower 節點都是被動的，它們不會主動發出任何的請求，只會響應 Leader 和 Candidate 發出的請求，對於每一個用戶的可變操作，都會被路由給 Leader 節點進行處理，除了 Leader 和 Follower 節點之外，Candidate 節點其實只是集群運行過程中的一個臨時狀態。

Raft 集群中的時間也被切分成了不同的幾個任期（Term），每一個任期都會由 Leader 的選舉開始，選舉結束後就會進入正常操作的階段，直到 Leader 節點出現問題才會開始新一輪的選擇。

raft-terms

每一個服務器都會存儲當前集群的最新任期，它就像是一個單調遞增的邏輯時鐘，能夠同步各個節點之間的狀態，當前節點持有的任期會隨著每一個請求被傳遞到其他的節點上。

Raft 協議在每一個任期的開始時都會從一個集群中選出一個節點作為集群的 Leader 節點，這個節點會負責集群中的日誌的複製以及管理工作。

raft-subproblems

我們將 Raft 協議分成三個子問題：節點選舉、日誌複製以及安全性，文章會以 etcd 為例介紹 Raft 協議是如何解決這三個子問題的。

節點選舉

使用 Raft 協議的 etcd 集群在啟動節點時，會遵循 Raft 協議的規則，所有節點一開始都被初始化為 Follower 狀態，新加入的節點會在 NewNode 中做一些配置的初始化，包括用於接收各種信息的 Channel：

// https://sourcegraph.com/github.com/etcd-io/etcd@1cab49e/-/blob/raft/node.go#L190-225
func StartNode(c *Config, peers []Peer) Node {
 r := newRaft(c)
 r.becomeFollower(1, None)
 r.raftLog.committed = r.raftLog.lastIndex()
 for _, peer := range peers {
 r.addNode(peer.ID)
 }
 n := newNode()
 go n.run(r)
 return &n
}

在做完這些初始化的節點和 Raft 配置的事情之後，就會進入一個由 for 和 select 組成的超大型循環，這個循環會從 Channel 中獲取待處理的事件：

// https://sourcegraph.com/github.com/etcd-io/etcd@1cab49e/-/blob/raft/node.go#L291-423
func (n *node) run(r *raft) {
 lead := None
 for {
 if lead != r.lead {
 lead = r.lead
 }
 select { 

 case m :=  r.Step(m)
 case  r.tick()
 case  close(n.done)
 return
 }
 }
}

作者對整個循環內的代碼進行了簡化，因為當前只需要關心三個 Channel 中的消息，也就是用於接受其他節點消息的 recvc、用於觸發定時任務的 tickc 以及用於暫停當前節點的 stop。

raft-etcd

除了 stop Channel 中介紹到的消息之外，recvc 和 tickc 兩個 Channel 中介紹到事件時都會交給當前節點持有 Raft 結構體處理。

定時器與心跳

當節點從任意狀態（包括啟動）調用 becomeFollower 時，都會將節點的定時器設置為 tickElection：

// https://sourcegraph.com/github.com/etcd-io/etcd@1cab49e/-/blob/raft/raft.go#L636-643
func (r *raft) tickElection() {
 r.electionElapsed++
 if r.promotable() && r.pastElectionTimeout() {
 r.electionElapsed = 0
 r.Step(pb.Message{From: r.id, Type: pb.MsgHup})
 }
}

如果當前節點可以成為 Leader 並且上一次收到 Leader 節點的消息或者心跳已經超過了等待的時間，當前節點就會發送 MsgHup 消息嘗試開始新的選舉。

但是如果 Leader 節點正常運行，就能夠同樣通過它的定時器 tickHeartbeat 向所有的 Follower 節點廣播心跳請求，也就是 MsgBeat 類型的 RPC 消息：

func (r *raft) tickHeartbeat() {
 r.heartbeatElapsed++
 r.electionElapsed++
 if r.heartbeatElapsed >= r.heartbeatTimeout {
 r.heartbeatElapsed = 0
 r.Step(pb.Message{From: r.id, Type: pb.MsgBeat})
 }
}

上述代碼段 Leader 節點中調用的 Step 函數，最終會調用 stepLeader 方法，該方法會根據消息的類型進行不同的處理：

// https://sourcegraph.com/github.com/etcd-io/etcd@1cab49e/-/blob/raft/raft.go#L931-1142
func stepLeader(r *raft, m pb.Message) error { 

 switch m.Type {
 case pb.MsgBeat:
 r.bcastHeartbeat()
 return nil
 // ...
 }
 //...
}

bcastHeartbeat 方法最終會向所有的 Follower 節點發送 MsgHeartbeat 類型的消息，通知它們目前 Leader 的存活狀態，重置所有 Follower 持有的超時計時器。

// https://sourcegraph.com/github.com/etcd-io/etcd@1cab49e/-/blob/raft/raft.go#L518-534
func (r *raft) sendHeartbeat(to uint64, ctx []byte) {
 commit := min(r.getProgress(to).Match, r.raftLog.committed)
 m := pb.Message{
 To: to,
 Type: pb.MsgHeartbeat,
 Commit: commit,
 Context: ctx,
 }
 r.send(m)
}

作為集群中的 Follower，它們會在 stepFollower 方法中處理接收到的全部消息，包括 Leader 節點發送的心跳 RPC 消息：

// https://sourcegraph.com/github.com/etcd-io/etcd@1cab49e/-/blob/raft/raft.go#L1191-1247
func stepFollower(r *raft, m pb.Message) error {
 switch m.Type {
 case pb.MsgHeartbeat:
 r.electionElapsed = 0
 r.lead = m.From
 r.handleHeartbeat(m)
 // ...
 }
 return nil
}

當 Follower 接受到了來自 Leader 的 RPC 消息 MsgHeartbeat 時，會將當前節點的選舉超時時間重置並通過 handleHeartbeat 向 Leader 節點發出響應 —— 通知 Leader 當前節點能夠正常運行。

而 Candidate 節點對於 MsgHeartBeat 消息的處理會稍有不同，它會先執行 becomeFollower 設置當前節點和 Raft 協議的配置：

// https://sourcegraph.com/github.com/etcd-io/etcd@1cab49e/-/blob/raft/raft.go#L1146-1189
func stepCandidate(r *raft, m pb.Message) error {
 // ...
 switch m.Type {
 case pb.MsgHeartbeat:
 r.becomeFollower(m.Term, m.From) // always m.Term == r.Term
 r.handleHeartbeat(m)
 }
 // ...
 return nil
}

Follower 與 Candidate 會根據節點類型的不同做出不同的響應，兩者收到心跳請求時都會重置節點的選舉超時時間，不過後者會將節點的狀態直接轉變成 Follower：

raft-heartbeat

當 Leader 節點收到心跳的響應時就會將對應節點的狀態設置為 Active，如果 Follower 節點在一段時間內沒有收到來自 Leader 節點的消息就會嘗試發起競選。

// https://sourcegraph.com/github.com/etcd-io/etcd@1cab49e/-/blob/raft/raft.go#L636-643
func (r *raft) tickElection() {
 r.electionElapsed++
 if r.promotable() && r.pastElectionTimeout() {
 r.electionElapsed = 0
 r.Step(pb.Message{From: r.id, Type: pb.MsgHup})
 }
}

到了這裡，心跳機制就起到了作用開始發送 MsgHup 嘗試重置整個集群中的 Leader 節點，接下來我們就會開始分析 Raft 協議中的競選流程了。

競選流程

如果集群中的某一個 Follower 節點長時間內沒有收到來自 Leader 的心跳請求，當前節點就會通過 MsgHup 消息進入預選舉或者選舉的流程。

// https://sourcegraph.com/github.com/etcd-io/etcd@1cab49e/-/blob/raft/raft.go#L785-927
func (r *raft) Step(m pb.Message) error {
 // ...
 switch m.Type {
 case pb.MsgHup:
 if r.state != StateLeader {
 if r.preVote {
 r.campaign(campaignPreElection)
 } else {
 r.campaign(campaignElection)
 }
 } else {
 r.logger.Debugf("%x ignoring MsgHup because already leader", r.id) 

 }
 }
 // ...
 return nil
}

如果收到 MsgHup 消息的節點不是 Leader 狀態，就會根據當前集群的配置選擇進入 PreElection 或者 Election 階段，PreElection 階段並不會真正增加當前節點的 Term，它的主要作用是得到當前集群能否成功選舉出一個 Leader 的答案，如果當前集群中只有兩個節點而且沒有預選舉階段，那麼這兩個節點的 Term 會無休止的增加，預選舉階段就是為了解決這一問題而出現的。

raft-cluster-states

在這裡不會討論預選舉的過程，而是將目光主要放在選舉階段，具體瞭解一下使用 Raft 協議的 etcd 集群是如何從眾多節點中選出 Leader 節點的。

我們可以繼續來分析 campaign 方法的具體實現，下面就是刪去預選舉相關邏輯後的代碼：

// https://sourcegraph.com/github.com/etcd-io/etcd@1cab49e/-/blob/raft/raft.go#L730-766
func (r *raft) campaign(t CampaignType) {
 r.becomeCandidate()
 if r.quorum() == r.poll(r.id, voteRespMsgType(voteMsg), true) {
 r.becomeLeader()
 return
 }
 for id := range r.prs {
 if id == r.id {
 continue
 }
 r.send(pb.Message{Term: r.Term, To: id, Type: pb.MsgVote, Index: r.raftLog.lastIndex(), LogTerm: r.raftLog.lastTerm(), Context: ctx})
 }
}

當前節點會立刻調用 becomeCandidate 將當前節點的 Raft 狀態變成候選人；在這之後，它會將票投給自己，如果當前集群只有一個節點，該節點就會直接成為集群中的 Leader 節點。

如果集群中存在了多個節點，就會向集群中的其他節點發出 MsgVote 消息，請求其他節點投票，在 Step 函數中包含不同狀態的節點接收到消息時的響應：

// https://sourcegraph.com/github.com/etcd-io/etcd@1cab49e/-/blob/raft/raft.go#L785-927
func (r *raft) Step(m pb.Message) error { 

 // ...
 switch m.Type {
 case pb.MsgVote, pb.MsgPreVote:
 canVote := r.Vote == m.From || (r.Vote == None && r.lead == None)
 if canVote && r.raftLog.isUpToDate(m.Index, m.LogTerm) {
 r.send(pb.Message{To: m.From, Term: m.Term, Type: pb.MsgVoteResp})
 r.electionElapsed = 0
 r.Vote = m.From
 } else {
 r.send(pb.Message{To: m.From, Term: r.Term, Type: pb.MsgVoteResp, Reject: true})
 }
 }
 // ...
 return nil
}

如果當前節點投的票就是消息的來源或者當前節點沒有投票也沒有 Leader，那麼就會向來源的節點投票，否則就會通知該節點當前節點拒絕投票。

raft-election

在 stepCandidate 方法中，候選人節點會處理來自其他節點的投票響應消息，也就是 MsgVoteResp：

// https://sourcegraph.com/github.com/etcd-io/etcd@1cab49e/-/blob/raft/raft.go#L1146-1189
func stepCandidate(r *raft, m pb.Message) error {
 switch m.Type {
 // ...
 case pb.MsgVoteResp:
 gr := r.poll(m.From, m.Type, !m.Reject)
 switch r.quorum() {
 case gr:
 r.becomeLeader()
 r.bcastAppend()
 // ...
 }
 }
 return nil
}

每當收到一個 MsgVoteResp 類型的消息時，就會設置當前節點持有的 votes 數組，更新其中存儲的節點投票狀態並返回投『同意』票的人數，如果獲得的票數大於法定人數 quorum，當前節點就會成為集群的 Leader 並向其他的節點發送當前節點當選的消息，通知其餘節點更新 Raft 結構體中的 Term 等信息。

節點狀態

對於每一個節點來說，它們根據不同的節點狀態會對網絡層發來的消息做出不同的響應，我們會分別介紹下面的四種狀態在 Raft 中對於配置和消息究竟是如何處理的。

raft-node-states

對於每一個 Raft 的節點狀態來說，它們分別有三個比較重要的區別，其中一個是在改變狀態時調用 becomeLeader、becomeCandidate、becomeFollower 和 becomePreCandidate 方法改變 Raft 狀態有比較大的不同，第二是處理消息時調用 stepLeader、stepCandidate 和 stepFollower 時有比較大的不同，最後是幾種不同狀態的節點具有功能不同的定時任務。

對於方法的詳細處理，我們在這一節中不詳細介紹和分析，如果一個節點的狀態是 Follower，那麼當前節點切換到 Follower 一定會通過 becomeFollower 函數，在這個函數中會重置節點持有任期，並且設置處理消息的函數為 stepFollower：

// https://sourcegraph.com/github.com/etcd-io/etcd@1cab49e/-/blob/raft/raft.go#L671-678
func (r *raft) becomeFollower(term uint64, lead uint64) {
 r.step = stepFollower
 r.reset(term)
 r.tick = r.tickElection
 r.lead = lead
 r.state = StateFollower
}
// https://sourcegraph.com/github.com/etcd-io/etcd@1cab49e/-/blob/raft/raft.go#L636-643
func (r *raft) tickElection() {
 r.electionElapsed++
 if r.promotable() && r.pastElectionTimeout() {
 r.electionElapsed = 0
 r.Step(pb.Message{From: r.id, Type: pb.MsgHup})
 }
}

除此之外，它還會設置一個用於在 Leader 節點宕機時觸發選舉的定時器 tickElection。

Candidate 狀態的節點與 Follower 的配置差不了太多，只是在消息處理函數 step、任期以及狀態上的設置有一些比較小的區別：

// https://sourcegraph.com/github.com/etcd-io/etcd@1cab49e/-/blob/raft/raft.go#L680-691
func (r *raft) becomeCandidate() {
 r.step = stepCandidate
 r.reset(r.Term + 1)
 r.tick = r.tickElection
 r.Vote = r.id
 r.state = StateCandidate
}

最後的 Leader 就與這兩者有其他的區別了，它不僅設置了處理消息的函數 step 而且設置了與其他狀態完全不同的 tick 函數：

// https://sourcegraph.com/github.com/etcd-io/etcd@1cab49e/-/blob/raft/raft.go#L708-728
func (r *raft) becomeLeader() {
 r.step = stepLeader 

 r.reset(r.Term)
 r.tick = r.tickHeartbeat
 r.lead = r.id
 r.state = StateLeader
 r.pendingConfIndex = r.raftLog.lastIndex()
 r.appendEntry(pb.Entry{Data: nil})
}

這裡的 tick 函數 tickHeartbeat 每隔一段時間會通過 Step 方法向集群中的其他節點發送 MsgBeat 消息：

// https://sourcegraph.com/github.com/etcd-io/etcd@1cab49e/-/blob/raft/raft.go#L646-669
func (r *raft) tickHeartbeat() {
 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})
 }
 } 
 if r.heartbeatElapsed >= r.heartbeatTimeout {
 r.heartbeatElapsed = 0
 r.Step(pb.Message{From: r.id, Type: pb.MsgBeat})
 }
}

上述代碼中的 MsgBeat 消息會在 Step 中被轉換成 MsgHeartbeat 最終發送給其他的節點，Leader 節點超時之後的選舉流程我們在前兩節中也已經介紹過了，在這裡就不再重複了。