ETCD源碼閱讀（二）

閱讀raftexample：etcd/contrib/raftexample

raftexample 包括三個組件：一個基于 raft 的kv store、一個 REST API Server、一個基于 etcd raft 實現(xiàn)的 Raft Node。其中Raft Node也擁有一個Http server，用于與peer節(jié)點進行通信。REST API Server則是這個Raft Node的client。

kv store 是一個k-v map，它保存了所有 committed 的鍵值。Raft Node 和REST API Server通過該kv store進行通信。數(shù)據(jù)更新通過kv store發(fā)送到 Raft Node。存儲庫在 Raft Node 回復更新 committed 后，更新內(nèi)部的k-v map。

REST API Server 通過訪問kv store來展示當前的 raft 共識狀況。GET 命令用于在kv store中查找鍵并返回值（如果有）。鍵值 PUT 命令向kv store發(fā)出更新proposal。

Raft Node與其集群peer一起參與共識。當 REST API Server器提交proposal時，Raft Node將這個proposal傳輸給其同伴。當raft集群達成共識時，Raft Node通過commit channel發(fā)布所有committed的proposal。對于 raftexample這個例子來說，這個channel的消費者是kv store。


raftexample組成部分
go build -o raftexample

啟動單節(jié)點

./raftexample --id 1 --cluster http://127.0.0.1:12379 --port 12380

id是指這個RAFT節(jié)點的id索引，--cluster后面是所有的peer列表，使用逗號分隔，port是k-v存儲server的地址

store a value ("hello") to a key ("my-key"):

curl -L http://127.0.0.1:12380/my-key -XPUT -d hello

retrieve the stored key:

curl -L http://127.0.0.1:12380/my-key

多節(jié)點集群以及配置變更

啟動一個三節(jié)點集群，配置信息參看Procfile

goreman start

使用一個POST請求，告知集群，一個ID為4的節(jié)點可以加入集群

curl -L http://127.0.0.1:12380/4 -XPOST -d http://127.0.0.1:42379

啟動這個ID為4的節(jié)點，并附上--join標簽

raftexample --id 4 --cluster http://127.0.0.1:12379,http://127.0.0.1:22379,http://127.0.0.1:32379,http://127.0.0.1:42379 --port 42380 --join

通過一個DELETE請求，可以刪除一個節(jié)點

curl -L http://127.0.0.1:12380/3 -XDELETE

閱讀RAFTEXAMPLE的main函數(shù)

package main
// etcd/contrib/raftexample/main.go

import (
    "flag"
    "strings"

    "go.etcd.io/etcd/raft/v3/raftpb"
)

func main() {
    cluster := flag.String("cluster", "http://127.0.0.1:9021", "comma separated cluster peers")
    id := flag.Int("id", 1, "node ID")
    kvport := flag.Int("port", 9121, "key-value server port")
    join := flag.Bool("join", false, "join an existing cluster")
    flag.Parse()

    proposeC := make(chan string)
    defer close(proposeC)
    confChangeC := make(chan raftpb.ConfChange)
    defer close(confChangeC)

    // raft provides a commit stream for the proposals from the http api
    var kvs *kvstore
    getSnapshot := func() ([]byte, error) { return kvs.getSnapshot() }

    // newRaftNode initiates a raft instance and returns a committed log entry
    // channel and error channel. Proposals for log updates are sent over the
    // provided the proposal channel. All log entries are replayed over the
    // commit channel, followed by a nil message (to indicate the channel is
    // current), then new log entries. To shutdown, close proposeC and read errorC.
    commitC, errorC, snapshotterReady := newRaftNode(*id, strings.Split(*cluster, ","), *join, getSnapshot, proposeC, confChangeC)

    kvs = newKVStore(<-snapshotterReady, proposeC, commitC, errorC)

    // the key-value http handler will propose updates to raft
    serveHttpKVAPI(kvs, *kvport, confChangeC, errorC)
}

  1. 創(chuàng)建了兩個通道 proposeC 和 confChangeC，分別用來傳遞 Proposal 和 raft 配置變更信息，并設(shè)置了 defer 關(guān)鍵字以確保這兩個channel會被關(guān)閉。

  2. 創(chuàng)建了一個名為 kvs 的指向 kvstore 結(jié)構(gòu)體的指針，并定義了 getSnapshot 函數(shù)，該函數(shù)用于獲取快照。

  3. 通過 newRaftNode 函數(shù)創(chuàng)建了一個 raft 節(jié)點，該節(jié)點與集群中的其他節(jié)點通信。將 proposeC 和 confChangeC 通道傳遞給 newRaftNode 函數(shù)，以便將來可以通過HTTP API向 raft 節(jié)點提交更新建議和配置變更請求。返回的 commitC 和 errorC 通道接收已提交的更新和錯誤信息。

  4. 通過 raft 節(jié)點的 snapshotterReady 通道，傳遞一個snap.Snapshotter指針給 kvs 結(jié)構(gòu)體，使得 kvs 結(jié)構(gòu)體能夠接收到快照并進行恢復。

  5. 使用新創(chuàng)建的 kvs 結(jié)構(gòu)體、kvport 端口、confChangeC 和 errorC 通道啟動了一個 HTTP 服務(wù)器，該服務(wù)器用于接收 key-value 更新的請求，同時將這些請求轉(zhuǎn)發(fā)到 raft 節(jié)點進行處理。

上面這段代碼中使用到了以下幾個channel：


各個channel
  • proposeC：這個channel會作為參數(shù)傳給RaftNode，KVStore通過這個channel發(fā)送新的Proposal給RaftNode
  • commitC：這個channel由RaftNode創(chuàng)建，作為返回值，又在KVStore初始化的時候被一個獨立的Goroutine進行處理。因為只有RaftNode可以決定一個Proposal是否被commit，而committed的Proposal才會被寫入KVStore。 
    // read commits from raft into kvStore map until error
go s.readCommits(commitC, errorC)
  • confChangeC：這個channel同樣作為參數(shù)傳給RaftNode，Http API Server通過這個channel將配置變更的信息發(fā)送給RaftNode
  • snapshotterReady：根據(jù) contrib/raftexample/raft.go中定義的newRaftNode函數(shù), 創(chuàng)建RaftNode時會根據(jù)Node ID創(chuàng)建一個目錄：/raftexample-id-snap。 完成設(shè)置后，啟動一個Goroutine來執(zhí)行startRaft()函數(shù) 
        rc.snapshotter = snap.New(zap.NewExample(), rc.snapdir)
    // signal replay has finished
    rc.snapshotterReady <- rc.snapshotter
    這樣，newKVStore函數(shù)就可以從channel中讀到snapshot（如果存在）
  • errorC：由KVStore、HTTP API Server、RaftNode共享，一旦出現(xiàn)錯誤，都會log.Fatal()

針對newRaftNode, kvstore, serveHttpAPI這幾個關(guān)鍵代碼，我們逐一進行分析。

newRaftNode：

這個函數(shù)主要就是進行配置設(shè)置，最后用一個goroutine來啟動raft節(jié)點，所以我們直接看這段代碼就行。

func (rc *raftNode) startRaft() {
    if !fileutil.Exist(rc.snapdir) {
        if err := os.Mkdir(rc.snapdir, 0750); err != nil {
            log.Fatalf("raftexample: cannot create
                 dir for snapshot (%v)", err)
        }
    }
    rc.snapshotter = snap.New(zap.NewExample(), rc.snapdir)

    oldwal := wal.Exist(rc.waldir)
    rc.wal = rc.replayWAL()

    // signal replay has finished
    rc.snapshotterReady <- rc.snapshotter

    rpeers := make([]raft.Peer, len(rc.peers))
    for i := range rpeers {
        rpeers[i] = raft.Peer{ID: uint64(i + 1)}
    }
    c := &raft.Config{
        ID:                        uint64(rc.id),
        ElectionTick:              10,
        HeartbeatTick:             1,
        Storage:                   rc.raftStorage,
        MaxSizePerMsg:             1024 * 1024,
        MaxInflightMsgs:           256,
        MaxUncommittedEntriesSize: 1 << 30,
    }

    if oldwal || rc.join {
        rc.node = raft.RestartNode(c)
    } else {
        rc.node = raft.StartNode(c, rpeers)
    }

    rc.transport = &rafthttp.Transport{
        Logger:      rc.logger,
        ID:          types.ID(rc.id),
        ClusterID:   0x1000,
        Raft:        rc,
        ServerStats: stats.NewServerStats("", ""),
        LeaderStats: stats.NewLeaderStats(zap.NewExample(), 
            strconv.Itoa(rc.id)),
        ErrorC:      make(chan error),
    }

    rc.transport.Start()
    for i := range rc.peers {
        if i+1 != rc.id {
            rc.transport.AddPeer(types.ID(i+1), []string{rc.peers[i]})
        }
    }

    go rc.serveRaft()
    go rc.serveChannels()
}
  1. 先檢查快照、重放預寫日志，使Raft節(jié)點恢復到之前的狀態(tài)。
  2. 根據(jù)傳遞進來的 peers 數(shù)組構(gòu)造出 raft.Peer 列表 rpeers，然后創(chuàng)建一個 Raft config對象 c，包括節(jié)點的 ID、選舉周期、心跳周期、存儲、最大消息大小等等
  3. 如果存在預寫日志或者需要加入一個現(xiàn)有集群（join==true），就會調(diào)用raft.RestartNode，不然就啟動一個新的Raft節(jié)點raft.StartNode
  4. 接著，創(chuàng)建一個 HTTP 傳輸對象 rafthttp.Transport，用于處理 Raft 節(jié)點之間的通信。然后，調(diào)用 rc.transport.Start 方法啟動，并添加peer節(jié)點。
  5. 最后，啟動兩個goroutine，用于處理RaftNode的事件請求與channel讀寫。
go rc.serveRaft()
go rc.serveChannels()

serveRaft()

func (rc *raftNode) serveRaft() {
    url, err := url.Parse(rc.peers[rc.id-1])
    if err != nil {
        log.Fatalf("raftexample: Failed parsing URL (%v)", err)
    }

    ln, err := newStoppableListener(url.Host, rc.httpstopc)
    if err != nil {
        log.Fatalf("raftexample: Failed to listen rafthttp (%v)", err)
    }
    // Transport implements Transporter interface. 
    // It provides the functionality to send raft messages to peers, 
    // and receive raft messages from peers. 
    // User should call Handler method to get a handler 
    // to serve requests received from peerURLs. 
    // User needs to call Start before calling other functions, 
    // and call Stop when the Transport is no longer used.
    err = (&http.Server{Handler: rc.transport.Handler()}).Serve(ln)
    select {
    case <-rc.httpstopc:
    default:
        log.Fatalf("raftexample: Failed to serve rafthttp (%v)", err)
    }
    close(rc.httpdonec)
}

  1. 在 serveRaft 函數(shù)中，首先通過id解析 rc.peers，得到當前節(jié)點的地址，Parse出一個URL對象。然后創(chuàng)建一個帶由stop channel的 listener 對象，并將其與 http.Server 綁定。通過調(diào)用 http.Server 的 Serve 方法來開始在監(jiān)聽地址上提供服務(wù)。直到出錯才會返回一個err。

  2. 當從 httpstopc 接收到停止服務(wù)的信號時，該方法會結(jié)束，然后關(guān)閉 httpdonec 以通知所有其他goroutine，http server已經(jīng)完成shutdown。這里的寫法非常有意思，我們來具體分析一下

    err = (&http.Server{Handler: rc.transport.Handler()}).Serve(ln)
    select {
    case <-rc.httpstopc:
    default:
        log.Fatalf("raftexample: Failed to serve rafthttp (%v)", err)
    }
    close(rc.httpdonec)
    // httpstopc：signals http server to shutdown
    // httpdonec：signals http server shutdown complete

第一行只有當http server發(fā)生錯誤才會返回，然后執(zhí)行select語句。

如果在執(zhí)行select語句前，httpstopc這個channel已經(jīng)被關(guān)閉，就會跳過default語句，直接關(guān)閉httpdonec這個channel。

如果httpstopc沒有被關(guān)閉，并且讀到了什么東西（意味著有別的goroutine在往里面寫，具體看下面的serveChannels分析），也會跳過default語句，直接關(guān)閉httpdonec這個channel。

如果httpstopc沒有任何響應，那么會執(zhí)行default分支，輸出log然后退出程序。

最后編輯于 ：
?著作權(quán)歸作者所有,轉(zhuǎn)載或內(nèi)容合作請聯(lián)系作者
【社區(qū)內(nèi)容提示】社區(qū)部分內(nèi)容疑似由AI輔助生成，瀏覽時請結(jié)合常識與多方信息審慎甄別。
平臺聲明：文章內(nèi)容（如有圖片或視頻亦包括在內(nèi)）由作者上傳并發(fā)布，文章內(nèi)容僅代表作者本人觀點，簡書系信息發(fā)布平臺，僅提供信息存儲服務(wù)。

相關(guān)閱讀更多精彩內(nèi)容

友情鏈接更多精彩內(nèi)容