手撸 golang etcd raft 协定之 2
缘起
最近浏览 [云原生分布式存储基石:etcd 深刻解析] (杜军 , 2019.1)
本系列笔记拟采纳 golang 练习之
gitee: https://gitee.com/ioly/learning.gooop
raft 分布式一致性算法
分布式存储系统通常会通过保护多个副原本进行容错,以进步零碎的可用性。这就引出了分布式存储系统的外围问题——如何保障多个正本的一致性?Raft 算法把问题分解成了首领选举(leader election)、日志复制(log replication)、安全性(safety)和成员关系变动(membership changes)这几个子问题。Raft 算法的基本操作只需 2 种 RPC 即可实现。RequestVote RPC 是在选举过程中通过旧的 Leader 触发的,AppendEntries RPC 是领导人触发的,目标是向其余节点复制日志条目和发送心跳(heartbeat)。
指标
- 依据 raft 协定,实现高可用分布式强统一的 kv 存储
子目标 (Day 2)
- 定义 raft rpc 接口
- 定义 raft lsm 无限状态自动机接口(状态模式)
设计
- config/IRaftConfig.go: 集群配置接口。简略起见,应用动态配置模式定义节点数量和地址。
- config/IRaftNodeConfig.go: 节点配置接口
- roles/roles.go:raft 三种角色常量
- timeout/timeout.go:超时工夫常量
- rpc/IRaftRPC.go: raft 协定的根本 RPC 接口及参数定义。简略起见,拟采纳 net/rpc 实现之。
- rpc/IRaftRPCServer.go: 反对 raft 协定的服务器接口。简略起见,拟采纳 net/rpc 实现之。
- lsm/IRaftLSM.go: raft 无限状态机接口
- lsm/IRaftState.go: 状态接口
- lsm/tRaftStateBase.go: 根本状态数据
- lsm/tFollowerState: follower 状态的实现,未实现
config/IRaftConfig.go
集群配置接口。简略起见,应用动态配置模式定义节点数量和地址。
package config
type IRaftConfig interface {ID() string
Nodes() []IRaftNodeConfig
}
config/IRaftNodeConfig.go
节点配置接口
package config
type IRaftNodeConfig interface {ID() string
Endpoint() string}
roles/roles.go
raft 三种角色常量
package roles
type RaftRole int
const Follower RaftRole = 1
const Candidate RaftRole = 2
const Leader RaftRole = 3
timeout/timeout.go
超时工夫常量
package timeout
import "time"
const HeartbeatInterval = 150 * time.Millisecond
const HeartbeatTimeout = 5 * HeartbeatInterval
const ElectionTimeout = HeartbeatTimeout
rpc/IRaftRPC.go
raft 协定的根本 RPC 接口及参数定义。简略起见,拟采纳 net/rpc 实现之。
package rpc
type IRaftRPC interface {RequestVote(cmd *RequestVoteCmd, ret *RequestVoteRet) error
AppendEntries(cmd *AppendEntriesCmd, ret *AppendEntriesRet) error
}
type RequestVoteCmd struct {
CandidateID string
Term int
LastLogIndex int
LastLogTerm int
}
type RequestVoteRet struct {
Term int
VoteGranted bool
}
type AppendEntriesCmd struct {
Term int
LeaderID string
PrevLogTerm int
PrevLogIndex int
LeaderCommit int
Entries []*LogEntry}
type LogEntry struct {
Tag int
Content []byte}
type AppendEntriesRet struct {
Term int
Success bool
}
rpc/IRaftRPCServer.go
反对 raft 协定的服务器接口。简略起见,拟采纳 net/rpc 实现之。
package rpc
type IRaftRPCServer interface {BeginServeTCP(port int, r IRaftRPC)
}
lsm/IRaftLSM.go
raft 无限状态机接口
package lsm
import "learning/gooop/etcd/raft/rpc"
// IRaftLSM raft 无限状态自动机
type IRaftLSM interface {
rpc.IRaftRPC
State() IRaftState}
lsm/IRaftState.go
状态接口
package lsm
import (
"learning/gooop/etcd/raft/roles"
"learning/gooop/etcd/raft/rpc"
)
type IRaftState interface {
rpc.IRaftRPC
Role() roles.RaftRole
Start()}
lsm/tRaftStateBase.go
根本状态数据
package lsm
import (
"learning/gooop/etcd/raft/config"
"learning/gooop/etcd/raft/roles"
)
//
type tRaftStateBase struct {
// 以后角色
role roles.RaftRole
// 以后任期号
term int
// leader.id
leaderID string
// 集群配置
cfg config.IRaftConfig
}
func newRaftStateBase(term int, cfg config.IRaftConfig) *tRaftStateBase {it := new(tRaftStateBase)
it.init(term, cfg)
return it
}
// init initialize self, with term and config specified
func (me *tRaftStateBase) init(term int, cfg config.IRaftConfig) {
me.cfg = cfg
me.role = roles.Follower
me.term = term
me.leaderID = ""
}
func (me *tRaftStateBase) Role() roles.RaftRole {return me.role}
lsm/tFollowerState
follower 状态的实现,未实现
package lsm
import (
"learning/gooop/etcd/raft/config"
"learning/gooop/etcd/raft/timeout"
"sync"
"time"
)
type tFollowerState struct {
tRaftStateBase
mInitOnce sync.Once
mStartOnce sync.Once
mEventMap map[tFollowerEvent][]tFollowerEventHandler}
type tFollowerEvent int
const evStart tFollowerEvent = 1
type tFollowerEventHandler func(e tFollowerEvent, args ...interface{})
func newFollowerState(term int, cfg config.IRaftConfig) *tFollowerState {it := new(tFollowerState)
it.init(term, cfg)
// todo: to implement IRaftState
return it
}
func (me *tFollowerState) init(term int, cfg config.IRaftConfig) {me.mInitOnce.Do(func() {me.tRaftStateBase = *newRaftStateBase(term, cfg)
// init event map
me.mEventMap = make(map[tFollowerEvent][]tFollowerEventHandler)
me.registerEventHandlers()})
}
func (me *tFollowerState) registerEventHandlers() {me.mEventMap[evStart] = []tFollowerEventHandler{me.afterStartThenBeginWatchLeaderTimeout,}
}
func (me *tFollowerState) raise(e tFollowerEvent, args ...interface{}) {if handlers, ok := me.mEventMap[e]; ok {
for _, it := range handlers {it(e, args...)
}
}
}
func (me *tFollowerState) Start() {me.mStartOnce.Do(func() {me.raise(evStart)
})
}
func (me *tFollowerState) afterStartThenBeginWatchLeaderTimeout(e tFollowerEvent, args ...interface{}) {go func() {
iCheckingTimeoutInterval := timeout.HeartbeatTimeout / 3
for range time.Tick(iCheckingTimeoutInterval) {// todo: watch leader.AppendEntries rpc timeout}
}()}
(未完待续)