早在2013年11月份，在raft論文還只能在網(wǎng)上下載到草稿版時，我曾經(jīng)寫過一篇blog對其進行簡要分析。4年過去了，各種raft協(xié)議的講解鋪天蓋地，raft也確實得到了廣泛的應用。其中最知名的應用莫過于etcd。etcd將raft協(xié)議本身實現(xiàn)為一個library，位于https://github.com/coreos/etcd/tree/master/raft，然后本身作為一個應用使用它。

本文不講解raft協(xié)議核心內(nèi)容，而是站在一個etcd raft library使用者的角度，講解要用上這個library需要了解的東西。

這個library使用起來相對來說還是有點麻煩。官方有一個使用示例在 https://github.com/coreos/etcd/tree/master/contrib/raftexample。整體來說，這個庫實現(xiàn)了raft協(xié)議核心的內(nèi)容，比如append log的邏輯，選主邏輯，snapshot，成員變更等邏輯。需要明確的是：library沒有實現(xiàn)消息的網(wǎng)絡傳輸和接收，庫只會把一些待發(fā)送的消息保存在內(nèi)存中，用戶自定義的網(wǎng)絡傳輸層取出消息并發(fā)送出去，并且在網(wǎng)絡接收端，需要調一個library的函數(shù)，用于將收到的消息傳入library，后面會詳細說明。同時，library定義了一個Storage接口，需要library的使用者自行實現(xiàn)。

Storage接口如下:

// Storage is an interface that may be implemented by the application // to retrieve log entries from storage. // // If any Storage method returns an error, the raft instance will // become inoperable and refuse to participate in elections; the // application is responsible for cleanup and recovery in this case. type Storage interface {// InitialState returns the saved HardState and ConfState information.InitialState() (pb.HardState, pb.ConfState, error)// Entries returns a slice of log entries in the range [lo,hi).// MaxSize limits the total size of the log entries returned, but// Entries returns at least one entry if any.Entries(lo, hi, maxSize uint64) ([]pb.Entry, error)// Term returns the term of entry i, which must be in the range// [FirstIndex()-1, LastIndex()]. The term of the entry before// FirstIndex is retained for matching purposes even though the// rest of that entry may not be available.Term(i uint64) (uint64, error)// LastIndex returns the index of the last entry in the log.LastIndex() (uint64, error)// FirstIndex returns the index of the first log entry that is// possibly available via Entries (older entries have been incorporated// into the latest Snapshot; if storage only contains the dummy entry the// first log entry is not available).FirstIndex() (uint64, error)// Snapshot returns the most recent snapshot.// If snapshot is temporarily unavailable, it should return ErrSnapshotTemporarilyUnavailable,// so raft state machine could know that Storage needs some time to prepare// snapshot and call Snapshot later.Snapshot() (pb.Snapshot, error) }

這些接口在library中會被用到。熟悉raft協(xié)議的人不難理解。上面提到的官方示例https://github.com/coreos/etcd/tree/master/contrib/raftexample中使用了library自帶的MemoryStorage，和etcd的wal和snap包做持久化，重啟的時候從wal和snap中獲取日志恢復MemoryStorage。

要提供這種IO/網(wǎng)絡密集型的東西，提高吞吐最好的手段就是batch加批處理了。etcd raft library正是這么做的。

下面看一下為了做這事，etcd提供的核心抽象Ready結構體：

// Ready encapsulates the entries and messages that are ready to read, // be saved to stable storage, committed or sent to other peers. // All fields in Ready are read-only. type Ready struct {// The current volatile state of a Node.// SoftState will be nil if there is no update.// It is not required to consume or store SoftState.*SoftState// The current state of a Node to be saved to stable storage BEFORE// Messages are sent.// HardState will be equal to empty state if there is no update.pb.HardState// ReadStates can be used for node to serve linearizable read requests locally// when its applied index is greater than the index in ReadState.// Note that the readState will be returned when raft receives msgReadIndex.// The returned is only valid for the request that requested to read.ReadStates []ReadState// Entries specifies entries to be saved to stable storage BEFORE// Messages are sent.Entries []pb.Entry// Snapshot specifies the snapshot to be saved to stable storage.Snapshot pb.Snapshot// CommittedEntries specifies entries to be committed to a// store/state-machine. These have previously been committed to stable// store.CommittedEntries []pb.Entry// Messages specifies outbound messages to be sent AFTER Entries are// committed to stable storage.// If it contains a MsgSnap message, the application MUST report back to raft// when the snapshot has been received or has failed by calling ReportSnapshot.Messages []pb.Message// MustSync indicates whether the HardState and Entries must be synchronously// written to disk or if an asynchronous write is permissible.MustSync bool }

可以說，這個Ready結構體封裝了一批更新，這些更新包括：

• pb.HardState: 包含當前節(jié)點見過的最大的term，以及在這個term給誰投過票，已經(jīng)當前節(jié)點知道的commit index
• Messages: 需要廣播給所有peers的消息
• CommittedEntries:已經(jīng)commit了，還沒有apply到狀態(tài)機的日志
• Snapshot:需要持久化的快照

庫的使用者從node結構體提供的一個ready channel中不斷的pop出一個個的Ready進行處理，庫使用者通過如下方法拿到Ready channel：

func (n *node) Ready() <-chan Ready { return n.readyc }

應用需要對Ready的處理包括:

將HardState, Entries, Snapshot持久化到storage。

將Messages(上文提到的msgs)非阻塞的廣播給其他peers

將CommittedEntries(已經(jīng)commit還沒有apply)應用到狀態(tài)機。

如果發(fā)現(xiàn)CommittedEntries中有成員變更類型的entry，調用node的ApplyConfChange()方法讓node知道(這里和raft論文不一樣，論文中只要節(jié)點收到了成員變更日志就應用)

調用Node.Advance()告訴raft node，這批狀態(tài)更新處理完了，狀態(tài)已經(jīng)演進了，可以給我下一批Ready讓我處理。

應用通過raft.StartNode()來啟動raft中的一個副本，函數(shù)內(nèi)部通過啟動一個goroutine運行

func (n *node) run(r *raft)

來啟動服務。

應用通過調用

func (n *node) Propose(ctx context.Context, data []byte) error

來Propose一個請求給raft，被raft開始處理后返回。

增刪節(jié)點通過調用

func (n *node) ProposeConfChange(ctx context.Context, cc pb.ConfChange) error

node結構體包含幾個重要的channel:

// node is the canonical implementation of the Node interface type node struct {propc chan pb.Messagerecvc chan pb.Messageconfc chan pb.ConfChangeconfstatec chan pb.ConfStatereadyc chan Readyadvancec chan struct{}tickc chan struct{}done chan struct{}stop chan struct{}status chan chan Statuslogger Logger }

• propc: propc是一個沒有buffer的channel，應用通過Propose接口寫入的請求被封裝成Message被push到propc中，node的run方法從propc中pop出Message，append自己的raft log中，并且將Message放入mailbox中(raft結構體中的msgs []pb.Message)，這個msgs會被封裝在Ready中，被應用從readyc中取出來，然后通過應用自定義的transport發(fā)送出去。

• recvc: 應用自定義的transport在收到Message后需要調用

  func (n *node) Step(ctx context.Context, m pb.Message) error 來把Message放入recvc中，經(jīng)過一些處理后，同樣，會把需要發(fā)送的Message放入到對應peers的mailbox中。后續(xù)通過自定義transport發(fā)送出去。

• readyc／advancec: readyc和advancec都是沒有buffer的channel，node.run()內(nèi)部把相關的一些狀態(tài)更新打包成Ready結構體(其中一種狀態(tài)就是上面提到的msgs)放入readyc中。應用從readyc中pop出Ready中，對相應的狀態(tài)進行處理，處理完成后，調用

  rc.node.Advance() 往advancec中push一個空結構體告訴raft，已經(jīng)對這批Ready包含的狀態(tài)進行了相應的處理，node.run()內(nèi)部從advancec中得到通知后，對內(nèi)部一些狀態(tài)進行處理，比如把已經(jīng)持久化到storage中的entries從內(nèi)存(對應type unstable struct)中刪除等。
• tickc:應用定期往tickc中push空結構體，node.run()會調用tick()函數(shù)，對于leader來說，tick()會給其他peers發(fā)心跳，對于follower來說，會檢查是否需要發(fā)起選主操作。

• confc/confstatec:應用從Ready中拿出CommittedEntries，檢查其如果含有成員變更類型的日志，則需要調用

  func (n *node) ApplyConfChange(cc pb.ConfChange) *pb.ConfState

  這個函數(shù)會push ConfChange到confc中，confc同樣是個無buffer的channel，node.run()內(nèi)部會從confc中拿出ConfChange，然后進行真正的增減peers操作，之后將最新的成員組push到confstatec中，而ApplyConfChange函數(shù)從confstatec pop出最新的成員組返回給應用。

可以說，要想用上etcd的raft library還是需要了解不少東西的。

轉載于:https://www.cnblogs.com/foxmailed/p/7137431.html

總結

以上是生活随笔為你收集整理的etcd raft library设计原理和使用的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。