以etcd源碼中的一個(gè)kvstore的例子來(lái)分析基于raft算法的kvstore的實(shí)現(xiàn)，在etcd/contrib/raftexample目錄下，啟動(dòng)代碼main.go如下：

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

  //當(dāng)kvstore中收到配置添加請(qǐng)求時(shí)會(huì)向proposeC通道發(fā)送kv數(shù)據(jù)，在raft中會(huì)得到proposeC通道的事件進(jìn)行處理
  proposeC := make(chan string)
  defer close(proposeC)
  //當(dāng)kvstore中收到集群節(jié)點(diǎn)變更請(qǐng)求時(shí)會(huì)向confChangeC通道發(fā)送集群變更數(shù)據(jù)，在raft中會(huì)得到confChangeC通道的事件進(jìn)行處理
  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() }
  //當(dāng)raft中數(shù)據(jù)可以提交時(shí)會(huì)向commitC通道發(fā)送消息，這樣kvstore就可以監(jiān)聽(tīng)該通道消息，當(dāng)收到提交消息時(shí)會(huì)修改kvstore內(nèi)存中的值
  commitC, errorC, snapshotterReady := newRaftNode(*id, strings.Split(*cluster, ","), *join, getSnapshot, proposeC, confChangeC)

  //直到snapshotterReady通道有數(shù)據(jù)了，即snapshot可用了，才可以創(chuàng)建kvstore實(shí)例
  kvs = newKVStore(<-snapshotterReady, proposeC, commitC, errorC)

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

啟動(dòng)流程如下圖：

server啟動(dòng).png

raftNode的啟動(dòng)方法如下：

func (rc *raftNode) startRaft() {
//如果snapshot目錄不存在，則創(chuàng)建snapshot目錄，目錄命名規(guī)則:raftexample-id-snap
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(rc.snapdir)
rc.snapshotterReady <- rc.snapshotter

oldwal := wal.Exist(rc.waldir)
//重放wal日志，
rc.wal = rc.replayWAL()

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

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

ss := &stats.ServerStats{}
ss.Initialize()

rc.transport = &rafthttp.Transport{
    ID:          types.ID(rc.id),
    ClusterID:   0x1000,
    Raft:        rc,
    ServerStats: ss,
    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]})
    }
}

//監(jiān)聽(tīng)并處理raft停止命令
go rc.serveRaft()
//監(jiān)聽(tīng)配置添加等命令
go rc.serveChannels()
}

在server啟動(dòng)過(guò)程中主要有4個(gè)處理模塊，每個(gè)模塊都有對(duì)應(yīng)的routine處理事件：

• node:用于處理raft算法相關(guān)邏輯。
• raftNode:用于調(diào)度node產(chǎn)生的消息，node模塊的執(zhí)行是通過(guò)消息機(jī)制，將產(chǎn)生的消息交給raftNode進(jìn)行調(diào)度。
• raftNode.transport:主要處理其他server的連接。
• kvstore:用于監(jiān)聽(tīng)kvstore的配置添加和集群變更請(qǐng)求，會(huì)將請(qǐng)求轉(zhuǎn)換為消息交給raftNode來(lái)調(diào)度處理。

在創(chuàng)建各個(gè)模塊的routine處理各個(gè)模塊的事件之前，需要把server的狀態(tài)恢復(fù)到上次停止時(shí)的狀態(tài)?；謴?fù)到之前的狀態(tài)需要兩種文件，即wal日志和snapshot文件，wal日志是預(yù)寫(xiě)日志，在生成日志條目運(yùn)用到raft算法前，需要先寫(xiě)入到wal文件，避免server突然宕機(jī)丟失raft相關(guān)數(shù)據(jù)，而snapshot用于保存server的狀態(tài)和數(shù)據(jù)快照。wal文件默認(rèn)每存儲(chǔ)10000條日志進(jìn)行一次切分，即生成一個(gè)新的wal文件，然后執(zhí)行snapshot操作，執(zhí)行完之前的wal文件便沒(méi)有用處了。wal文件與snapshot文件的區(qū)別是wal文件保存日志的追加記錄，記錄著數(shù)據(jù)和狀態(tài)的變化過(guò)程，而snapshot保存日志變化后產(chǎn)生的狀態(tài)或結(jié)果(即對(duì)于kvstore的配置數(shù)據(jù)，保存快照時(shí)的各個(gè)配置的值，對(duì)于之前該配置值的變化過(guò)程不保存，對(duì)于server的狀態(tài)，保存snapshot時(shí)的狀態(tài)，不保存之前狀態(tài)變化的過(guò)程)，所以wal文件會(huì)比較大，而snapshot文件相對(duì)較小。

那如何恢復(fù)到之前server關(guān)閉時(shí)的狀態(tài)呢？要結(jié)合wal和snapshot來(lái)做，首先通過(guò)snapshot文件恢復(fù)到server關(guān)閉前上一次snapshot時(shí)的數(shù)據(jù)和狀態(tài)，然后根據(jù)snapshot的最后一條日志索引，找到wal文件中snaoshot時(shí)的日志位置，再把該位置之后的wal文件里的日志重放，因?yàn)閣al文件記錄著server數(shù)據(jù)和狀態(tài)的變化過(guò)程，通過(guò)重放這個(gè)過(guò)程，可以恢復(fù)到之前server關(guān)閉時(shí)的數(shù)據(jù)和狀態(tài)。