以etcd源码中的一个kvstore的例子来分析基于raft算法的kvstore的实现，在etcd/contrib/raftexample目录下，启动代码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()
//当kvstore中收到配置添加请求时会向proposeC通道发送kv数据，在raft中会得到proposeC通道的事件进行处理
proposeC := make(chan string)
defer close(proposeC)
//当kvstore中收到集群节点变更请求时会向confChangeC通道发送集群变更数据，在raft中会得到confChangeC通道的事件进行处理
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() }
//当raft中数据可以提交时会向commitC通道发送消息，这样kvstore就可以监听该通道消息，当收到提交消息时会修改kvstore内存中的值
commitC, errorC, snapshotterReady := newRaftNode(*id, strings.Split(*cluster, ","), *join, getSnapshot, proposeC, confChangeC)
//直到snapshotterReady通道有数据了，即snapshot可用了，才可以创建kvstore实例
kvs = newKVStore(<-snapshotterReady, proposeC, commitC, errorC)
// the key-value http handler will propose updates to raft
serveHttpKVAPI(kvs, *kvport, confChangeC, errorC)
}
启动流程如下图：
server启动.png
raftNode的启动方法如下：
func (rc *raftNode) startRaft() {
//如果snapshot目录不存在，则创建snapshot目录，目录命名规则: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]})
}
}
//监听并处理raft停止命令
go rc.serveRaft()
//监听配置添加等命令
go rc.serveChannels()
}
在server启动过程中主要有4个处理模块，每个模块都有对应的routine处理事件：
- node:用于处理raft算法相关逻辑。
- raftNode:用于调度node产生的消息，node模块的执行是通过消息机制，将产生的消息交给raftNode进行调度。
- raftNode.transport:主要处理其他server的连接。
- kvstore:用于监听kvstore的配置添加和集群变更请求，会将请求转换为消息交给raftNode来调度处理。
在创建各个模块的routine处理各个模块的事件之前，需要把server的状态恢复到上次停止时的状态。恢复到之前的状态需要两种文件，即wal日志和snapshot文件，wal日志是预写日志，在生成日志条目运用到raft算法前，需要先写入到wal文件，避免server突然宕机丢失raft相关数据，而snapshot用于保存server的状态和数据快照。wal文件默认每存储10000条日志进行一次切分，即生成一个新的wal文件，然后执行snapshot操作，执行完之前的wal文件便没有用处了。wal文件与snapshot文件的区别是wal文件保存日志的追加记录，记录着数据和状态的变化过程，而snapshot保存日志变化后产生的状态或结果(即对于kvstore的配置数据，保存快照时的各个配置的值，对于之前该配置值的变化过程不保存，对于server的状态，保存snapshot时的状态，不保存之前状态变化的过程)，所以wal文件会比较大，而snapshot文件相对较小。
那如何恢复到之前server关闭时的状态呢？要结合wal和snapshot来做，首先通过snapshot文件恢复到server关闭前上一次snapshot时的数据和状态，然后根据snapshot的最后一条日志索引，找到wal文件中snaoshot时的日志位置，再把该位置之后的wal文件里的日志重放，因为wal文件记录着server数据和状态的变化过程，通过重放这个过程，可以恢复到之前server关闭时的数据和状态。
网友评论