缘起

最近阅读<> (刘金亮, 2021.1)
本系列笔记拟采用golang练习之

Saga模式

  • saga模式将分布式长事务切分为一系列独立短事务
  • 每个短事务是可通过补偿动作进行撤销的
  • 事务动作和补动作偿都是幂等的, 允许重复执行而不会有副作用 ``` Saga由一系列的子事务“Ti”组成， 每个Ti都有对应的补偿“Ci”， 当Ti出现问题时Ci用于处理Ti执行带来的问题。

可以通过下面的两个公式理解Saga模式。 T = T1 T2 … Tn T = TCT

Saga模式的核心理念是避免使用长期持有锁（如14.2.2节介绍的两阶段提交）的长事务， 而应该将事务切分为一组按序依次提交的短事务， Saga模式满足ACD（原子性、一致性、持久性）特征。

摘自 <> 刘金亮, 2021.1

  1. <a name="fGPkq"></a>
  2. # 目标
  4. - 为实现saga模式的分布式事务, 先撸一个pub/sub事务消息队列服务
  5. - 事务消息队列服务的功能性要求
  6.    - 消息不会丢失: 消息的持久化
  7.    - 消息的唯一性: 要求每个消息有全局ID和子事务ID
  8.    - 确保投递成功: 投递队列持久化, 投递状态持久化, 失败重试
  10. <a name="naUyZ"></a>
  11. # 子目标(Day 3)
  13. - 有序,可靠地投递消息
  14.    - 有序: 每个订阅者绑定独立的投递worker, 按时间戳投递
  15.    - 可靠: 投递状态的持久化
  17. <a name="XyCWB"></a>
  18. # 设计
  20. - QueuedMsg: 把投递队列的消息设计得胖一点, 以减少join的使用
  21. - IEventBus: 设计一个内部消息总线, 以减少逻辑耦合
  22. - tDeliveryService: 投递服务, 管理若干个投递worker
  23. - tWorkerInfo: 投递worker的初始化参数
  24. - tDeliveryWorker: 投递worker(未完成)
  26. <a name="HbILr"></a>
  27. # QueuedMsg.go
  28. 把待投递消息设计得胖一点, 以减少join的使用
  29. ```go
  30. package models
  32. type QueuedMsg struct {
  33.     ID int
  34.     ClientID string
  35.     NotifyURL string
  37.     MsgID int
  38.     GlobalID string
  39.     SubID string
  40.     SenderID string
  41.     CreateTime int64
  42.     Topic string
  43.     Content string
  45.     StatusFlag int
  46.     FailedCount int
  47. }

IEventBus.go

设计一个内部消息总线, 以减少逻辑耦合

  1. package eventbus
  3. import "sync"
  5. type EventHandleFunc func(e string, args interface{})
  7. type IEventBus interface {
  8.     Pub(e string, args interface{})
  9.     Sub(e string, handler EventHandleFunc)
  10. }
  13. type tEventBus struct {
  14.     rwmutex *sync.RWMutex
  15.     items map[string][]EventHandleFunc
  16. }
  19. func newEventBus() IEventBus {
  20.     it := new(tEventBus)
  21.     it.init()
  22.     return it
  23. }
  25. func (me *tEventBus) init() {
  26.     me.rwmutex = new(sync.RWMutex)
  27.     me.items = make(map[string][]EventHandleFunc)
  28. }
  30. func (me *tEventBus) Pub(e string, args interface{}) {
  31.     me.rwmutex.RLock()
  32.     defer me.rwmutex.RUnlock()
  34.     handlers,ok := me.items[e]
  35.     if ok {
  36.         for _,it := range handlers {
  37.             go it(e, args)
  38.         }
  39.     }
  40. }
  42. func (me *tEventBus) Sub(e string, handler EventHandleFunc) {
  43.     me.rwmutex.Lock()
  44.     defer me.rwmutex.Unlock()
  46.     handlers,ok := me.items[e]
  47.     if ok {
  48.         me.items[e] = append(handlers, handler)
  49.     } else {
  50.         me.items[e] = []EventHandleFunc{handler }
  51.     }
  52. }
  55. var GlobalEventBus = newEventBus()

tDeliveryService.go

投递服务, 管理若干个投递worker

  1. package delivery
  3. import (
  4.     "github.com/jmoiron/sqlx"
  5.     "learning/gooop/saga/mqs/database"
  6.     "learning/gooop/saga/mqs/eventbus"
  7.     "learning/gooop/saga/mqs/logger"
  8.     "learning/gooop/saga/mqs/models"
  9.     "sync"
  10.     "time"
  11. )
  13. type tDeliveryService struct {
  14.     rwmutex *sync.RWMutex
  15.     workers map[string]*tDeliveryWorker
  16. }
  18. func newDeliveryService() *tDeliveryService {
  19.     it := new(tDeliveryService)
  20.     it.init()
  21.     return it
  22. }
  24. func (me *tDeliveryService) init() {
  25. }
  27. func (me *tDeliveryService) handleBootEvent(e string, args interface{}) {
  28.     go me.beginCreatingWorkers()
  29.     go me.beginCleanExpiredWorkers()
  30. }
  33. func (me *tDeliveryService) beginCreatingWorkers() {
  34.     for {
  35.         e := database.DB(func(db *sqlx.DB) error {
  36.             now := time.Now().UnixNano()
  37.             rows, err := db.Queryx("select client_id, notify_url, expire_time from subscriber where expire_time>?", now)
  38.             if err != nil {
  39.                 return err
  40.             }
  42.             for rows.Next() {
  43.                 it := new(tWorkerInfo)
  44.                 err = rows.StructScan(it)
  45.                 if err != nil {
  46.                     return err
  47.                 }
  49.                 me.createWorker(it)
  50.             }
  52.             return nil
  53.         })
  55.         if e != nil {
  56.             logger.Logf("tDeliveryService.beginCreatingWorkers, error = %s", e.Error())
  57.         }
  58.         time.Sleep(time.Duration(5) * time.Second)
  59.     }
  60. }
  63. func (me *tDeliveryService) beginCleanExpiredWorkers() {
  64.     for range time.Tick(time.Duration(30) * time.Second) {
  65.         me.clean()
  66.     }
  67. }
  69. func (me *tDeliveryService) clean() {
  70.     me.rwmutex.RLock()
  71.     var keys []string
  72.     for k,v := range me.workers {
  73.         if v.isExpired() {
  74.             keys = append(keys, k)
  75.         }
  76.     }
  77.     me.rwmutex.RUnlock()
  79.     if len(keys) == 0 {
  80.         return
  81.     }
  83.     me.rwmutex.Lock()
  84.     defer me.rwmutex.Unlock()
  85.     for _,k := range keys {
  86.         delete(me.workers, k)
  87.     }
  88. }
  90. func (me *tDeliveryService) handleNewSubscriber(args interface{}) {
  91.     it, ok := args.(*models.SubMsg)
  92.     if !ok {
  93.         return
  94.     }
  96.     me.createWorker(&tWorkerInfo{
  97.         it.ClientID, it.NotifyUrl, it.ExpireTime,
  98.     })
  99. }
  101. func maxInt64(a, b int64) int64 {
  102.     if a >= b {
  103.         return a
  104.     }
  105.     return b
  106. }
  108. func (me *tDeliveryService) createWorker(info *tWorkerInfo) {
  109.     me.rwmutex.RLock()
  110.     w,ok := me.workers[info.ClientID]
  111.     me.rwmutex.RUnlock()
  112.     if ok {
  113.         w.info.ExpireTime = maxInt64(w.info.ExpireTime, info.ExpireTime)
  114.         return
  115.     }
  117.     me.rwmutex.Lock()
  118.     defer me.rwmutex.Unlock()
  119.     me.workers[info.ClientID] = newDeliveryWorker(info)
  120. }
  122. var gDeliveryService = newDeliveryService()
  124. func init() {
  125.     eventbus.GlobalEventBus.Sub("boot", gDeliveryService.handleBootEvent)
  126.     eventbus.GlobalEventBus.Sub("subscriber.new", gDeliveryService.handleBootEvent)
  127. }

tWorkerInfo.go

投递worker的初始化参数

  1. package delivery
  3. type tWorkerInfo struct {
  4.     ClientID string
  5.     NotifyURL string
  6.     ExpireTime int64
  7. }

tDeliveryWorker.go

投递worker, 未完成

  1. package delivery
  3. import (
  4.     "bytes"
  5.     "encoding/json"
  6.     "errors"
  7.     "github.com/jmoiron/sqlx"
  8.     "io/ioutil"
  9.     "learning/gooop/saga/mqs/database"
  10.     "learning/gooop/saga/mqs/logger"
  11.     "learning/gooop/saga/mqs/models"
  12.     "net/http"
  13.     "time"
  14. )
  16. type tDeliveryWorker struct {
  17.     info *tWorkerInfo
  18. }
  20. func newDeliveryWorker(info *tWorkerInfo) *tDeliveryWorker {
  21.     it := new(tDeliveryWorker)
  22.     it.info = info
  23.     go it.beginMainLoop()
  24.     return it
  25. }
  27. func (me *tDeliveryWorker) beginMainLoop() {
  28.     for !me.isExpired() {
  29.         ok, msg := me.peek()
  30.         if ok {
  31.             switch msg.StatusFlag {
  32.             case 0:
  33.                 // 未处理的消息
  34.                 me.handleNewMsg(msg)
  35.                 break
  37.             case 1:
  38.                 // 处理中的消息
  39.                 me.handleDeliveringMsg(msg)
  40.                 break
  41.             }
  43.         } else {
  44.             time.Sleep(time.Duration(1) * time.Second)
  45.         }
  46.     }
  47. }
  50. func (me *tDeliveryWorker) isExpired() bool {
  51.     return time.Now().UnixNano() >= me.info.ExpireTime
  52. }
  54. var gEmptyRowsErr = errors.New("empty rows")
  56. func (me *tDeliveryWorker) peek() (bool, *models.QueuedMsg) {
  57.     msg := &models.QueuedMsg{}
  58.     e := database.DB(func(db *sqlx.DB) error {
  59.         rows, err := db.Queryx("select * from delivery_queue where client_id=? order by create_time asc limit 1", me.info.ClientID)
  60.         if err != nil {
  61.             return err
  62.         }
  64.         if rows.Next() {
  65.             err = rows.StructScan(msg)
  66.             if err != nil {
  67.                 return err
  68.             }
  69.             return nil
  71.         } else {
  72.             return gEmptyRowsErr
  73.         }
  74.     })
  76.     if e != nil {
  77.         return false, nil
  78.     } else {
  79.         return true, msg
  80.     }
  81. }
  85. func (me *tDeliveryWorker) handleNewMsg(msg *models.QueuedMsg) bool {
  86.     err := database.DB(func(db *sqlx.DB) error {
  87.         r,e := db.Exec("update delivery_queue set status_flag=1 where id=?", msg.ID)
  88.         if e != nil {
  89.             return e
  90.         }
  92.         rows, e := r.RowsAffected()
  93.         if e != nil {
  94.             return e
  95.         }
  96.         if rows != 1 {
  97.             return gEmptyRowsErr
  98.         }
  100.         return nil
  101.     })
  103.     if err != nil {
  104.         logger.Logf("tDeliveryWorker.handleNewMsg, id=%v, msg=%s/%s, err=%s", me.info.ClientID, msg.GlobalID, msg.SubID, err.Error())
  105.         return false
  106.     }
  108.     if me.deliver(msg) {
  109.         // todo: move msg to success queue
  110.     } else {
  111.         // todo: msg.failed_count++
  112.     }
  114.     return false
  115. }
  118. func (me *tDeliveryWorker) deliver(msg *models.QueuedMsg) bool {
  119.     t := &models.TxMsg{
  120.         GlobalID: msg.GlobalID,
  121.         SubID: msg.SubID,
  122.         Topic: msg.Topic,
  123.         CreateTime: msg.CreateTime,
  124.         Content: msg.Content,
  125.     }
  126.     j,e := json.Marshal(t)
  127.     if e != nil {
  128.         return false
  129.     }
  131.     r, e := http.Post(me.info.NotifyURL, "application/json", bytes.NewReader(j))
  132.     if e != nil {
  133.         return false
  134.     }
  136.     defer r.Body.Close()
  137.     rep, e := ioutil.ReadAll(r.Body)
  138.     if e != nil {
  139.         return false
  140.     }
  142.     m := &models.OkMsg{}
  143.     e = json.Unmarshal(rep, m)
  144.     if e != nil {
  145.         return false
  146.     }
  148.     return m.OK
  149. }
  151. func (me *tDeliveryWorker) handleDeliveringMsg(msg *models.QueuedMsg) bool {
  152.     // todo: check if delivery timeout
  153.     return false
  154. }

(未完待续)