說(shuō)明

raft-example是etcd中一個(gè)示例代碼租谈，閱讀后對(duì)于理解etcd整個(gè)工作原理有很大幫助蛹屿，這里面核心有幾個(gè)組件：httpKVAPI、raftNode娄蔼、KVstore

數(shù)據(jù)結(jié)構(gòu)

HttpKVAPI

type httpKVAPI struct {
    store       *kvstore
    confChangeC chan<- raftpb.ConfChange
}

該數(shù)據(jù)結(jié)構(gòu)中有兩個(gè)數(shù)據(jù)結(jié)構(gòu)：

• store:數(shù)據(jù)的持久化底層存儲(chǔ)
• confChangeC：這是個(gè)channle结窘，用于傳遞集群節(jié)點(diǎn)修改的請(qǐng)求
  其他的比較簡(jiǎn)單很洋，主要是提供HTTP的服務(wù)，接收請(qǐng)求并發(fā)送給raftNode

    case r.Method == "PUT":
        v, err := ioutil.ReadAll(r.Body)
        if err != nil {
            log.Printf("Failed to read on PUT (%v)\n", err)
            http.Error(w, "Failed on PUT", http.StatusBadRequest)
            return
        }
        //將接受到鍵值數(shù)據(jù)請(qǐng)求發(fā)送出去
        h.store.Propose(key, string(v))

        // Optimistic-- no waiting for ack from raft. Value is not yet
        // committed so a subsequent GET on the key may return old value
        w.WriteHeader(http.StatusNoContent)

func (s *kvstore) Propose(k string, v string) {
    var buf bytes.Buffer
    if err := gob.NewEncoder(&buf).Encode(kv{k, v}); err != nil {
        log.Fatal(err)
    }
    s.proposeC <- buf.String()
}

raftNode

raftNode是最核心的數(shù)據(jù)結(jié)構(gòu)組件隧枫，他起到承上啟下的作用喉磁，他接收HTTP發(fā)送過(guò)來(lái)的請(qǐng)求，由于封裝了etcd-raft模塊悠垛，使得上層不用直接跟raft模塊交互，另外他管理了snapshot和WAL日志娜谊，并最終發(fā)送持久化通道确买，使得數(shù)據(jù)持久化到kvstore，數(shù)據(jù)結(jié)構(gòu)如下：

type raftNode struct {
    proposeC    <-chan string            // proposed messages (k,v)
    confChangeC <-chan raftpb.ConfChange // proposed cluster config changes
    commitC     chan<- *string           // entries committed to log (k,v)
    errorC      chan<- error             // errors from raft session

    id          int      // client ID for raft session
    peers       []string // raft peer URLs
    join        bool     // node is joining an existing cluster
    waldir      string   // path to WAL directory
    snapdir     string   // path to snapshot directory
    getSnapshot func() ([]byte, error)
    lastIndex   uint64 // index of log at start

    confState     raftpb.ConfState
    snapshotIndex uint64
    appliedIndex  uint64

    // raft backing for the commit/error channel
    node        raft.Node
    raftStorage *raft.MemoryStorage
    wal         *wal.WAL

    snapshotter      *snap.Snapshotter
    snapshotterReady chan *snap.Snapshotter // signals when snapshotter is ready

    snapCount uint64
    transport *rafthttp.Transport
    stopc     chan struct{} // signals proposal channel closed
    httpstopc chan struct{} // signals http server to shutdown
    httpdonec chan struct{} // signals http server shutdown complete
}

啟動(dòng)流程

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() }
    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)
}

核心有三個(gè)步驟
1纱皆、啟動(dòng)一個(gè)raftNode
newRaftNode代碼中實(shí)例后啟動(dòng)一個(gè)goroutine startRaft方法湾趾，代碼如下：

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)
    rc.snapshotterReady <- rc.snapshotter

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

    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.node = raft.RestartNode(c)
    } else {
        startPeers := rpeers
        if rc.join {
            startPeers = nil
        }
        rc.node = raft.StartNode(c, startPeers)
    }

    rc.transport = &rafthttp.Transport{
        Logger:      zap.NewExample(),
        ID:          types.ID(rc.id),
        ClusterID:   0x1000,
        Raft:        rc,
        ServerStats: stats.NewServerStats("", ""),
        LeaderStats: stats.NewLeaderStats(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()//負(fù)責(zé)監(jiān)聽(tīng)當(dāng)前節(jié)點(diǎn)的地址，完成和其他節(jié)點(diǎn)的通信
    //1派草、負(fù)責(zé)接收上層模塊到etcd-raft模塊之間的通信搀缠，
    //2、同時(shí)etcd-raft模塊返回給上層模塊的數(shù)據(jù)機(jī)其他相關(guān)的操作
    go rc.serveChannels()
}

這里面又開(kāi)啟了兩個(gè)groutine近迁，分別執(zhí)行serveRaft和serveChannels

• serveRaft
  這個(gè)方法主要負(fù)責(zé)和其他節(jié)點(diǎn)之間的通信
• serveChannels

func (rc *raftNode) serveChannels() {
    snap, err := rc.raftStorage.Snapshot()
    if err != nil {
        panic(err)
    }
    rc.confState = snap.Metadata.ConfState
    rc.snapshotIndex = snap.Metadata.Index
    rc.appliedIndex = snap.Metadata.Index

    defer rc.wal.Close()

    ticker := time.NewTicker(100 * time.Millisecond)
    defer ticker.Stop()

    // send proposals over raft
    go func() {
        confChangeCount := uint64(0)

        for rc.proposeC != nil && rc.confChangeC != nil {
            select {
            case prop, ok := <-rc.proposeC:
                fmt.Println("prop value is:",prop)
                if !ok {
                    rc.proposeC = nil
                } else {
                    // blocks until accepted by raft state machine
                    rc.node.Propose(context.TODO(), []byte(prop))
                }

            case cc, ok := <-rc.confChangeC:
                if !ok {
                    rc.confChangeC = nil
                } else {
                    confChangeCount++
                    cc.ID = confChangeCount
                    rc.node.ProposeConfChange(context.TODO(), cc)
                }
            }
        }
        // client closed channel; shutdown raft if not already
        close(rc.stopc)
    }()

    // event loop on raft state machine updates
    for {
        select {
        case <-ticker.C:
            rc.node.Tick()

        // store raft entries to wal, then publish over commit channel
        case rd := <-rc.node.Ready():
            fmt.Println("node is ready ,rd is ",rd.Entries)
            rc.wal.Save(rd.HardState, rd.Entries)
            if !raft.IsEmptySnap(rd.Snapshot) {
                rc.saveSnap(rd.Snapshot)//將新的快照數(shù)據(jù)寫(xiě)入快照文件中
                rc.raftStorage.ApplySnapshot(rd.Snapshot)//將新的快照持久化到raftNode
                rc.publishSnapshot(rd.Snapshot)//通知上層應(yīng)用加載新快照
            }
            rc.raftStorage.Append(rd.Entries)
            rc.transport.Send(rd.Messages)//將待發(fā)送的消息發(fā)送到指定節(jié)點(diǎn)
            //將已提交艺普、待應(yīng)用的Entry記錄應(yīng)用到上層應(yīng)用的狀態(tài)機(jī)中
            if ok := rc.publishEntries(rc.entriesToApply(rd.CommittedEntries)); !ok {
                rc.stop()
                return
            }
            //觸發(fā)日志壓縮
            rc.maybeTriggerSnapshot()
            rc.node.Advance()//上層應(yīng)用處理完Ready實(shí)例，通知etcd-raft組件準(zhǔn)備返回下一個(gè)Ready實(shí)例

        case err := <-rc.transport.ErrorC:
            rc.writeError(err)
            return

        case <-rc.stopc:
            rc.stop()
            return
        }
    }
}

這里方法完成兩個(gè)功能：
1鉴竭、負(fù)責(zé)接收上層模塊到etcd-raft模塊之間的通信
2歧譬、同時(shí)etcd-raft模塊返回給上層模塊的數(shù)據(jù)和其他相關(guān)的操作

for {
        select {
        case <-ticker.C:
            rc.node.Tick()

        // store raft entries to wal, then publish over commit channel
        case rd := <-rc.node.Ready():
            fmt.Println("node is ready ,rd is ",rd.Entries)
            rc.wal.Save(rd.HardState, rd.Entries)
            if !raft.IsEmptySnap(rd.Snapshot) {
                rc.saveSnap(rd.Snapshot)//將新的快照數(shù)據(jù)寫(xiě)入快照文件中
                rc.raftStorage.ApplySnapshot(rd.Snapshot)//將新的快照持久化到raftNode
                rc.publishSnapshot(rd.Snapshot)//通知上層應(yīng)用加載新快照
            }
            rc.raftStorage.Append(rd.Entries)
            rc.transport.Send(rd.Messages)//將待發(fā)送的消息發(fā)送到指定節(jié)點(diǎn)
            //將已提交、待應(yīng)用的Entry記錄應(yīng)用到上層應(yīng)用的狀態(tài)機(jī)中
            if ok := rc.publishEntries(rc.entriesToApply(rd.CommittedEntries)); !ok {
                rc.stop()
                return
            }
            //觸發(fā)日志壓縮
            rc.maybeTriggerSnapshot()
            rc.node.Advance()//上層應(yīng)用處理完Ready實(shí)例搏存，通知etcd-raft組件準(zhǔn)備返回下一個(gè)Ready實(shí)例

        case err := <-rc.transport.ErrorC:
            rc.writeError(err)
            return

        case <-rc.stopc:
            rc.stop()
            return
        }
    }

看這個(gè)方法瑰步，主要處理了兩個(gè)通道，node.Ready和ticker.C璧眠，
node.readyc通道是其他節(jié)點(diǎn)（peer）已經(jīng)保存接收到消息后的通知缩焦，此時(shí)保存wal日志，然后存儲(chǔ)entries责静，同時(shí)將待發(fā)送的消息發(fā)送到其他節(jié)點(diǎn)袁滥，調(diào)用publishEntries是的entry記錄應(yīng)用到上層應(yīng)用的狀態(tài)機(jī)，然后調(diào)用Advance方法通知底層etcd-raft模塊將unstable中對(duì)應(yīng)的記錄刪除灾螃。
publishEntries的核心代碼如下：

    s := string(ents[i].Data)
            select {
            case rc.commitC <- &s:
            case <-rc.stopc:
                return false
            }

// special nil commit to signal replay has finished
        if ents[i].Index == rc.lastIndex {
            select {
            case rc.commitC <- nil:
            case <-rc.stopc:
                return false
            }
        }

在wal日志保存后發(fā)送commit通道呻拌，通知kvstore持久化,如果發(fā)送nil，通知kvstore加載快照進(jìn)行重放睦焕，后面kvstore會(huì)提到相關(guān)的代碼邏輯藐握。
定時(shí)器用于定時(shí)更新邏輯時(shí)鐘靴拱，重新設(shè)置選舉時(shí)間和心跳時(shí)間的超時(shí)

    // send proposals over raft
    go func() {
        confChangeCount := uint64(0)

        for rc.proposeC != nil && rc.confChangeC != nil {
            select {
            case prop, ok := <-rc.proposeC:
                if !ok {
                    rc.proposeC = nil
                } else {
                    // blocks until accepted by raft state machine
                    rc.node.Propose(context.TODO(), []byte(prop))
                }

            case cc, ok := <-rc.confChangeC:
                if !ok {
                    rc.confChangeC = nil
                } else {
                    confChangeCount++
                    cc.ID = confChangeCount
                    rc.node.ProposeConfChange(context.TODO(), cc)
                }
            }
        }
        // client closed channel; shutdown raft if not already
        close(rc.stopc)
    }()

此方法主要處理前文提到的httpKVAPI接收請(qǐng)求到請(qǐng)求，通過(guò)proposeC和confChangeC這兩個(gè)通道通信猾普，
2袜炕、創(chuàng)建一個(gè)kvStore
3、啟動(dòng)HTTPAPI服務(wù)初家，用于接收請(qǐng)求

kvStore

type kvstore struct {
    proposeC    chan<- string // channel for proposing updates
    mu          sync.RWMutex
    kvStore     map[string]string // current committed key-value pairs
    snapshotter *snap.Snapshotter
}

kvStore的核心方法偎窘，readCommit（）:

func (s *kvstore) readCommits(commitC <-chan *string, errorC <-chan error) {
    for data := range commitC {
        // 接收到的 data 為 nil，從快照中加載數(shù)據(jù)
        if data == nil {
            // done replaying log; new data incoming
            // OR signaled to load snapshot
            snapshot, err := s.snapshotter.Load()
            if err == snap.ErrNoSnapshot {
                return
            }
            if err != nil {
                log.Panic(err)
            }
            log.Printf("loading snapshot at term %d and index %d", snapshot.Metadata.Term, snapshot.Metadata.Index)
            if err := s.recoverFromSnapshot(snapshot.Data); err != nil {
                log.Panic(err)
            }
            continue
        }
        // 2. 接收到的 data 不為 nil溜在，序列化 data 并存儲(chǔ)到 kvStore
        var dataKv kv
        dec := gob.NewDecoder(bytes.NewBufferString(*data))
        if err := dec.Decode(&dataKv); err != nil {
            log.Fatalf("raftexample: could not decode message (%v)", err)
        }
        fmt.Println("start save data:",dataKv)
        s.mu.Lock()
        s.kvStore[dataKv.Key] = dataKv.Val
        s.mu.Unlock()
    }
    if err, ok := <-errorC; ok {
        log.Fatal(err)
    }
}

如上文提到這邊接收到commitC通道后進(jìn)行持久化處理陌知，當(dāng)data為nil時(shí)從快照中加載數(shù)據(jù)并重放后放到存儲(chǔ)中，當(dāng)data不為nil時(shí)序列化 data 并存儲(chǔ)到 kvStore

工作流程

etcd-raft-example.jpg

1掖肋、httpKVAPI發(fā)送請(qǐng)求
2仆葡、raftnode接收后讀取數(shù)據(jù)并通過(guò)etcd-raft通知其他節(jié)點(diǎn)（這邊目前還有點(diǎn)疑問(wèn)，是發(fā)送其他節(jié)點(diǎn)什么數(shù)據(jù)志笼？其他節(jié)點(diǎn)做了什么沿盅？）
3、當(dāng)接收到readyc通道數(shù)據(jù)后纫溃，保存wal日志腰涧、生成快照、保存到內(nèi)存存儲(chǔ)中紊浩，然后將數(shù)據(jù)發(fā)送到其他節(jié)點(diǎn)窖铡，最后將待應(yīng)用的記錄應(yīng)用到狀態(tài)機(jī)中
4、kvstrore存儲(chǔ)接收到commitC消息后持久化存儲(chǔ)
5坊谁、觸發(fā)日志壓縮
6万伤、上層應(yīng)用處理完Ready實(shí)例，通知etcd-raft組件準(zhǔn)備返回下一個(gè)Ready實(shí)例

客戶端執(zhí)行過(guò)程

1呜袁、客戶端發(fā)送給leader請(qǐng)求后敌买，leader會(huì)把log entry append到日志中，然后發(fā)送給其他節(jié)點(diǎn)阶界，使用gRpc通信虹钮，AppendEntriesRPC
2、當(dāng)leader確定log entry被大多數(shù)節(jié)點(diǎn)寫(xiě)入日志膘融，apply這條log entry到狀態(tài)機(jī)中然后返回結(jié)果給客戶端