缘起

最近阅读 [云原生分布式存储基石：etcd深入解析] (杜军 , 2019.1)
本系列笔记拟采用golang练习之
gitee: https://gitee.com/ioly/learning.gooop

raft分布式一致性算法

  1. 分布式存储系统通常会通过维护多个副本来进行容错，
  2. 以提高系统的可用性。
  3. 这就引出了分布式存储系统的核心问题——如何保证多个副本的一致性？
  5. Raft算法把问题分解成了四个子问题：
  6. 1. 领袖选举（leader election）、
  7. 2. 日志复制（log replication）、
  8. 3. 安全性（safety）
  9. 4. 成员关系变化（membership changes）
  10. 这几个子问题。

目标

  • 根据raft协议，实现高可用分布式强一致的kv存储

子目标（Day 7）

  • 实现各raft节点之间的rpc通讯
    • 定义IRaftClientService服务，管理所有节点的tcp长连接
    • 定义IRaftClient接口，封装节点间的rpc调用
      • 基于状态模式，区分已连接状态和已断开状态
      • 基于事件驱动的逻辑编排
      • 基于读写分离的字段管理

设计

  • model/IEventDirvenModel: 事件驱动的逻辑编排基类
  • IRaftClientService：管理所有的节点间rpc连接
  • IRaftClient：管理当前节点与某个节点间的rpc连接
  • iClientState：基于状态模式的rpc连接状态接口
  • iStateContext：状态模式下的连接状态上下文接口
  • tRaftClient：IRaftClient接口的具体实现，并实现iStateContext接口。
  • tConnectedState: 管理已连接状态的rpc连接
    • 定时Ping以检测连接状态
    • 基于事件驱动的逻辑编排
    • 基于读写分离的字段管理
  • tBrokenState：管理已断开状态的rpc连接
    • 定时Dial以尝试重连接
    • 基于事件驱动的逻辑编排
    • 基于读写分离的字段管理

model/IEventDirvenModel.go

事件驱动的逻辑编排基类

  1. package model
  3. type TEventHandleFunc func(e string, args ...interface{})
  5. type IEventDrivenModel interface {
  6.     hook(e string, handleFunc TEventHandleFunc)
  7.     raise(e string, args ...interface{})
  8. }
  10. type TEventDrivenModel struct {
  11.     items map[string][]TEventHandleFunc
  12. }
  14. func (me *TEventDrivenModel) Hook(e string, handler TEventHandleFunc) {
  15.     arr, ok := me.items[e]
  16.     if ok {
  17.         me.items[e] = append(arr, handler)
  18.     } else {
  19.         me.items[e] = []TEventHandleFunc{handler}
  20.     }
  21. }
  23. func (me *TEventDrivenModel) Raise(e string, args ...interface{}) {
  24.     if handlers, ok := me.items[e]; ok {
  25.         for _, it := range handlers {
  26.             it(e, args...)
  27.         }
  28.     }
  29. }

IRaftClientService.go

管理所有的节点间rpc连接

  1. package client
  3. import (
  4.     "learning/gooop/etcd/raft/config"
  5.     "learning/gooop/etcd/raft/rpc"
  6.     netrpc "net/rpc"
  7.     "sync"
  8. )
  10. type tRaftClientService struct {
  11.     cfg config.IRaftConfig
  12.     rwmutex *sync.RWMutex
  13.     clients map[string]IRaftClient
  14. }
  16. func NewRaftClientService(cfg config.IRaftConfig) IRaftClientService {
  17.     it := new(tRaftClientService)
  18.     it.init(cfg)
  19.     return it
  20. }
  23. func (me *tRaftClientService) init(cfg config.IRaftConfig) {
  24.     me.cfg = cfg
  25.     me.rwmutex = new(sync.RWMutex)
  26.     me.clients = make(map[string]IRaftClient)
  27. }
  30. func (me *tRaftClientService) Using(peerID string, action func(client rpc.IRaftRPC) error) error {
  31.     // check client exists?
  32.     me.rwmutex.RLock()
  33.     it,ok := me.clients[peerID]
  34.     if ok {
  35.         return action(it)
  36.     }
  38.     var nodeCfg config.IRaftNodeConfig
  39.     for _,it := range me.cfg.Nodes() {
  40.         if it.ID() == peerID {
  41.             nodeCfg = it
  42.             break
  43.         }
  44.     }
  45.     me.rwmutex.RUnlock()
  47.     // dial to peer
  48.     conn, err := netrpc.Dial("tcp", nodeCfg.Endpoint())
  49.     if err != nil {
  50.         return err
  51.     }
  53.     // to create new client
  54.     me.rwmutex.Lock()
  55.     defer me.rwmutex.Unlock()
  57.     // recheck client
  58.     _,ok = me.clients[peerID]
  59.     if ok {
  60.         defer conn.Close()
  61.         return action(it)
  62.     }
  64.     // create new client
  65.     return action(newRaftClient(nodeCfg, conn))
  66. }

IRaftClient.go

管理当前节点与某个节点间的rpc连接

  1. package client
  3. import "learning/gooop/etcd/raft/rpc"
  5. type IRaftClient interface {
  6.     rpc.IRaftRPC
  7.     iStateContext
  9.     Ping(cmd *PingCmd, ret *PingRet) error
  10. }
  13. type PingCmd struct {
  14.     SenderID string
  15.     Timestamp int64
  16. }
  18. type PingRet struct {
  19.     SenderID string
  20.     Timestamp int64
  21. }

iClientState.go

基于状态模式的rpc连接状态接口

  1. package client
  3. import "learning/gooop/etcd/raft/rpc"
  5. type iClientState interface {
  6.     rpc.IRaftRPC
  8.     Start()
  9.     Ping(cmd *PingCmd, ret *PingRet) error
  10. }

iStateContext.go

状态模式下的连接状态上下文接口

  1. package client
  3. import (
  4.     "learning/gooop/etcd/raft/config"
  5.     "net/rpc"
  6. )
  8. type iStateContext interface {
  9.     Config() config.IRaftNodeConfig
  10.     GetConn() *rpc.Client
  11.     SetConn(client *rpc.Client)
  12.     HandleStateChanged(state iClientState)
  13. }

tRaftClient.go

IRaftClient接口的具体实现，并实现iStateContext接口。

  1. package client
  3. import (
  4.     "learning/gooop/etcd/raft/config"
  5.     "net/rpc"
  6.     rrpc "learning/gooop/etcd/raft/rpc"
  7. )
  9. type tRaftClient struct {
  10.     cfg config.IRaftNodeConfig
  11.     conn *rpc.Client
  12.     state iClientState
  13. }
  15. func newRaftClient(cfg config.IRaftNodeConfig, conn *rpc.Client) IRaftClient {
  16.     it := new(tRaftClient)
  17.     it.init(cfg, conn)
  18.     return it
  19. }
  21. func (me *tRaftClient) init(cfg config.IRaftNodeConfig, conn *rpc.Client) {
  22.     me.cfg = cfg
  23.     me.conn = conn
  24. }
  26. func (me *tRaftClient) Config() config.IRaftNodeConfig {
  27.     return me.cfg
  28. }
  30. func (me *tRaftClient) GetConn() *rpc.Client {
  31.     return me.conn
  32. }
  34. func (me *tRaftClient) SetConn(conn *rpc.Client) {
  35.     me.conn = conn
  36. }
  38. func (me *tRaftClient) HandleStateChanged(state iClientState) {
  39.     me.state = state
  40.     state.Start()
  41. }
  43. func (me *tRaftClient) Heartbeat(cmd *rrpc.HeartbeatCmd, ret *rrpc.HeartbeatRet) error {
  44.     return me.state.Heartbeat(cmd, ret)
  45. }
  47. func (me *tRaftClient) AppendLog(cmd *rrpc.AppendLogCmd, ret *rrpc.AppendLogRet) error {
  48.     return me.state.AppendLog(cmd, ret)
  49. }
  51. func (me *tRaftClient) CommitLog(cmd *rrpc.CommitLogCmd, ret *rrpc.CommitLogRet) error {
  52.     return me.state.CommitLog(cmd, ret)
  53. }
  55. func (me *tRaftClient) RequestVote(cmd *rrpc.RequestVoteCmd, ret *rrpc.RequestVoteRet) error {
  56.     return me.state.RequestVote(cmd, ret)
  57. }
  59. func (me *tRaftClient) Ping(cmd *PingCmd, ret *PingRet) error {
  60.     return me.state.Ping(cmd, ret)
  61. }

tConnectedState.go

管理已连接状态的rpc连接

  • 定时Ping以检测连接状态
  • 基于事件驱动的逻辑编排
  • 基于读写分离的字段管理 ```go package client

import ( “learning/gooop/etcd/raft/model” “learning/gooop/etcd/raft/rpc” “learning/gooop/etcd/raft/timeout” “sync” “time” )

type tConnectedState struct { model.TEventDrivenModel context iStateContext

  1. mInitOnce sync.Once
  2. mStartOnce sync.Once
  4. // update: ceInit, ceDisposing
  5. mDisposedFlag bool

}

// trigger: init() // args: empty const ceInit = “connected.init”

// trigger: Start() // args: empty const ceStart = “connected.Start”

// trigger: // args: empty const ceDisposing = “connected.Disposing”

// trigger: whenStartThenBeginPing() // args: empty const cePingFailed = “connected.PingFailed”

func newConnectedState(ctx iStateContext) iClientState { it := new(tConnectedState) it.init(ctx) return it }

func (me *tConnectedState) init(ctx iStateContext) { me.mInitOnce.Do(func() { me.context = ctx me.initEventHandlers() me.Raise(ceInit) }) }

func (me *tConnectedState) initEventHandlers() { // write only logic me.hookEventsForDisposedFlag()

  1. // read only logic
  2. me.Hook(ceStart,
  3.     me.whenStartThenBeginPing)
  5. me.Hook(cePingFailed,
  6.     me.whenPingFailedThenSwitchToBrokenState)
  8. me.Hook(ceDisposing,
  9.     me.whenDisposingThenCloseConn)

}

func (me *tConnectedState) Start() { me.mStartOnce.Do(func() { me.Raise(ceStart) }) }

func (me *tConnectedState) hookEventsForDisposedFlag() { me.Hook(ceInit, func(e string, args …interface{}) { me.mDisposedFlag = false })

  1. me.Hook(ceDisposing, func(e string, args ...interface{}) {
  2.     me.mDisposedFlag = true
  3. })

}

func (me *tConnectedState) whenStartThenBeginPing( string, …interface{}) { go func() { cmd := &PingCmd{ SenderID: me.context.Config().ID(), Timestamp: time.Now().UnixNano(), } ret := &PingRet{} for range time.Tick(timeout.ClientPingInterval) { if me.mDisposedFlag { return }

  1.         cmd.Timestamp = time.Now().UnixNano()
  2.         err := me.Ping(cmd, ret)
  3.         if err != nil {
  4.             me.Raise(cePingFailed)
  5.         }
  6.     }
  7. }()

}

func (me *tConnectedState) whenPingFailedThenSwitchToBrokenState( string, …interface{}) { me.Raise(ceDisposing) me.context.HandleStateChanged(newBrokenState(me.context)) }

func (me *tConnectedState) whenDisposingThenCloseConn( string, …interface{}) { it := me.context.GetConn() if it != nil { it.Close() }

  1. me.context.SetConn(nil)

}

func (me tConnectedState) Heartbeat(cmd rpc.HeartbeatCmd, ret *rpc.HeartbeatRet) error { return me.context.GetConn().Call(“TRaftRPCServer.Heartbeat”, cmd, ret) }

func (me tConnectedState) AppendLog(cmd rpc.AppendLogCmd, ret *rpc.AppendLogRet) error { return me.context.GetConn().Call(“TRaftRPCServer.AppendLog”, cmd, ret) }

func (me tConnectedState) CommitLog(cmd rpc.CommitLogCmd, ret *rpc.CommitLogRet) error { return me.context.GetConn().Call(“TRaftRPCServer.CommitLog”, cmd, ret) }

func (me tConnectedState) RequestVote(cmd rpc.RequestVoteCmd, ret *rpc.RequestVoteRet) error { return me.context.GetConn().Call(“TRaftRPCServer.RequestVote”, cmd, ret) }

func (me tConnectedState) Ping(cmd PingCmd, ret *PingRet) error { return me.context.GetConn().Call(“TRaftRPCServer.Ping”, cmd, ret) }

  1. <a name="YeHX8"></a>
  2. # tBrokenState.go
  3. 管理已断开状态的rpc连接
  5. - 定时Dial以尝试重连接
  6. - 基于事件驱动的逻辑编排
  7. - 基于读写分离的字段管理
  8. ```go
  9. package client
  11. import (
  12.     "errors"
  13.     "learning/gooop/etcd/raft/model"
  14.     rrpc "learning/gooop/etcd/raft/rpc"
  15.     "learning/gooop/etcd/raft/timeout"
  16.     "sync"
  17.     "net/rpc"
  18.     "time"
  19. )
  21. type tBrokenState struct {
  22.     model.TEventDrivenModel
  23.     context iStateContext
  25.     mInitOnce sync.Once
  26.     mStartOnce sync.Once
  28.     mDisposedFlag bool
  29. }
  31. // trigger : init()
  32. // args: empty
  33. const beInit = "broken.init"
  35. // trigger: Start()
  36. // args: empty
  37. const beStart = "broken.Start"
  39. // trigger: whenStartThenBeginDial
  40. // args: *rpc.Client
  41. const beDialOK  = "broken.DialOK"
  43. // trigger: whenDialOKThenSwitchToConnectedState
  44. // args: empty
  45. const beDisposing = "broken.Disposing"
  47. func newBrokenState(ctx iStateContext) iClientState {
  48.     it := new(tBrokenState)
  49.     it.init(ctx)
  50.     return it
  51. }
  53. func (me *tBrokenState) init(ctx iStateContext) {
  54.     me.mInitOnce.Do(func() {
  55.         me.context = ctx
  56.         me.initEventHandlers()
  57.         me.Raise(beInit)
  58.     })
  59. }
  61. func (me *tBrokenState) initEventHandlers() {
  62.     // write only logic
  63.     me.hookEventsForDisposedFlag()
  65.     // read only logic
  66.     me.Hook(beStart,
  67.         me.whenStartThenBeginDial)
  69.     me.Hook(beDialOK,
  70.         me.whenDialOKThenSetConn)
  72.     me.Hook(beDialOK,
  73.         me.whenDialOKThenSwitchToConnectedState)
  74. }
  77. func (me *tBrokenState) hookEventsForDisposedFlag() {
  78.     me.Hook(beInit, func(e string, args ...interface{}) {
  79.         me.mDisposedFlag = false
  80.     })
  82.     me.Hook(beDisposing, func(e string, args ...interface{}) {
  83.         me.mDisposedFlag = true
  84.     })
  85. }
  87. func (me *tBrokenState) Start() {
  88.     me.mStartOnce.Do(func() {
  89.         me.Raise(beStart)
  90.     })
  91. }
  94. func (me *tBrokenState) whenStartThenBeginDial(_ string, _ ...interface{}) {
  95.     go func() {
  96.         for !me.mDisposedFlag {
  97.             conn, err := rpc.Dial("tcp", me.context.Config().Endpoint())
  98.             if err == nil {
  99.                 me.Raise(beDialOK, conn)
  100.                 break
  102.             } else {
  103.                 time.Sleep(timeout.ClientRedialInterval)
  104.             }
  105.         }
  106.     }()
  107. }
  110. func (me *tBrokenState) whenDialOKThenSetConn(_ string, args ...interface{}) {
  111.     conn := args[0].(*rpc.Client)
  112.     me.context.SetConn(conn)
  113. }
  116. func (me *tBrokenState) whenDialOKThenSwitchToConnectedState(_ string, _ ...interface{}) {
  117.     me.Raise(beDisposing)
  118.     me.context.HandleStateChanged(newConnectedState(me.context))
  119. }
  121. func (me *tBrokenState) Heartbeat(cmd *rrpc.HeartbeatCmd, ret *rrpc.HeartbeatRet) error {
  122.     return gErrorConnectionBroken
  123. }
  125. func (me *tBrokenState) AppendLog(cmd *rrpc.AppendLogCmd, ret *rrpc.AppendLogRet) error {
  126.     return gErrorConnectionBroken
  127. }
  129. func (me *tBrokenState) CommitLog(cmd *rrpc.CommitLogCmd, ret *rrpc.CommitLogRet) error {
  130.     return gErrorConnectionBroken
  131. }
  133. func (me *tBrokenState) RequestVote(cmd *rrpc.RequestVoteCmd, ret *rrpc.RequestVoteRet) error {
  134.     return gErrorConnectionBroken
  135. }
  137. func (me *tBrokenState) Ping(cmd *PingCmd, ret *PingRet) error {
  138.     return gErrorConnectionBroken
  139. }
  141. var gErrorConnectionBroken = errors.New("peer connection broken")

（未完待续）