缘起

最近阅读<> (刘金亮, 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 4)
  13. - 完善投递worker
  14.    - 未处理消息: 标记, 并尝试投递
  15.    - 已处理消息: 判断是否超时, 并重试投递
  16.    - 投递成功: 移动到成功投递表
  17.    - 投递失败: 重置标记, 下轮重试
  18. - 数据库表相应的细节调整
  19.    - delivery_queue: 去掉failed_count, 增加update_time时间戳
  20.    - success_queue: 去掉sub_id, 改为client_id, 并增加create_time时间戳
  21.    - failed_queue: 因为不允许失败, 因此删除失败投递表
  23. <a name="beZzA"></a>
  24. # tDeliveryWorker.go
  26. - 完善投递worker
  27.    - 未处理消息: 标记, 并尝试投递
  28.    - 已处理消息: 判断是否超时, 并重试投递
  29.    - 投递成功: 移动到成功投递表
  30.    - 投递失败: 重置标记, 下轮重试
  31. ```go
  32. package delivery
  34. import (
  35.     "bytes"
  36.     "encoding/json"
  37.     "errors"
  38.     "github.com/jmoiron/sqlx"
  39.     "io/ioutil"
  40.     "learning/gooop/saga/mqs/database"
  41.     "learning/gooop/saga/mqs/logger"
  42.     "learning/gooop/saga/mqs/models"
  43.     "net/http"
  44.     "time"
  45. )
  47. type tDeliveryWorker struct {
  48.     info *tWorkerInfo
  49. }
  51. func newDeliveryWorker(info *tWorkerInfo) *tDeliveryWorker {
  52.     it := new(tDeliveryWorker)
  53.     it.info = info
  54.     go it.beginMainLoop()
  55.     return it
  56. }
  58. func (me *tDeliveryWorker) beginMainLoop() {
  59.     for !me.isExpired() {
  60.         ok, msg := me.peek()
  61.         if ok {
  62.             switch msg.StatusFlag {
  63.             case 0:
  64.                 // 未处理的消息
  65.                 me.handleUndeliveredMsg(msg)
  66.                 break
  68.             case 1:
  69.                 // 处理中的消息
  70.                 me.handleDeliveringMsg(msg)
  71.                 break
  72.             }
  74.         } else {
  75.             time.Sleep(time.Duration(1) * time.Second)
  76.         }
  77.     }
  78. }
  81. func (me *tDeliveryWorker) isExpired() bool {
  82.     return time.Now().UnixNano() >= me.info.ExpireTime
  83. }
  85. // peek: 从待投递队列中获取最早的一条记录
  86. func (me *tDeliveryWorker) peek() (bool, *models.QueuedMsg) {
  87.     msg := &models.QueuedMsg{}
  88.     e := database.DB(func(db *sqlx.DB) error {
  89.         rows, err := db.Queryx(
  90.             "select * from delivery_queue where client_id=? order by create_time asc limit 1",
  91.             me.info.ClientID,
  92.         )
  93.         if err != nil {
  94.             return err
  95.         }
  97.         if rows.Next() {
  98.             err = rows.StructScan(msg)
  99.             if err != nil {
  100.                 return err
  101.             }
  102.             return nil
  104.         } else {
  105.             return gEmptyRowsErr
  106.         }
  107.     })
  109.     if e != nil {
  110.         return false, nil
  111.     } else {
  112.         return true, msg
  113.     }
  114. }
  117. // handleUndeliveredMsg: if msg unhandled, then try to deliver it
  118. func (me *tDeliveryWorker) handleUndeliveredMsg(msg *models.QueuedMsg) {
  119.     err := database.DB(func(db *sqlx.DB) error {
  120.         now := time.Now().UnixNano()
  121.         r,e := db.Exec(
  122.             "update delivery_queue set status_flag=1, update_time=? where id=? and status_flag=0 and update_time=?",
  123.             now,
  124.             msg.ID,
  125.             msg.UpdateTime,
  126.         )
  127.         if e != nil {
  128.             return e
  129.         }
  131.         rows, e := r.RowsAffected()
  132.         if e != nil {
  133.             return e
  134.         }
  135.         if rows != 1 {
  136.             return gOneRowsErr
  137.         }
  139.         msg.UpdateTime = now
  140.         return nil
  141.     })
  143.     if err != nil {
  144.         logger.Logf("tDeliveryWorker.handleNewMsg, id=%v, msg=%s/%s, err=%s", me.info.ClientID, msg.GlobalID, msg.SubID, err.Error())
  145.         return
  146.     }
  148.     if me.deliver(msg) {
  149.         me.afterDeliverySuccess(msg)
  151.     } else {
  152.         me.afterDeliveryFailed(msg)
  153.     }
  154. }
  156. // deliver: use http.Post function to delivery msg
  157. func (me *tDeliveryWorker) deliver(msg *models.QueuedMsg) bool {
  158.     t := &models.TxMsg{
  159.         GlobalID: msg.GlobalID,
  160.         SubID: msg.SubID,
  161.         Topic: msg.Topic,
  162.         CreateTime: msg.CreateTime,
  163.         Content: msg.Content,
  164.     }
  165.     j,e := json.Marshal(t)
  166.     if e != nil {
  167.         logger.Logf("tDeliveryWorker.deliver, failed json.Marshal, id=%v, msg=%s/%s", me.info.ClientID, msg.GlobalID, msg.SubID)
  168.         return false
  169.     }
  171.     r, e := http.Post(me.info.NotifyURL, "application/json", bytes.NewReader(j))
  172.     if e != nil {
  173.         logger.Logf("tDeliveryWorker.deliver, failed http.Post, id=%v, msg=%s/%s", me.info.ClientID, msg.GlobalID, msg.SubID)
  174.         return false
  175.     }
  177.     defer r.Body.Close()
  178.     rep, e := ioutil.ReadAll(r.Body)
  179.     if e != nil {
  180.         logger.Logf("tDeliveryWorker.deliver, failed ioutil.ReadAll, id=%v, msg=%s/%s", me.info.ClientID, msg.GlobalID, msg.SubID)
  181.         return false
  182.     }
  184.     m := &models.OkMsg{}
  185.     e = json.Unmarshal(rep, m)
  186.     if e != nil {
  187.         logger.Logf("tDeliveryWorker.deliver, failed json.Unmarshal, id=%v, msg=%s/%s", me.info.ClientID, msg.GlobalID, msg.SubID)
  188.         return false
  189.     }
  191.     if m.OK {
  192.         return true
  193.     } else {
  194.         logger.Logf("tDeliveryWorker.deliver, failed OkMsg.OK, id=%v, msg=%s/%s", me.info.ClientID, msg.GlobalID, msg.SubID)
  195.         return false
  196.     }
  197. }
  199. // handleDeliveringMsg: if delivery timeout, then retry delivery
  200. func (me *tDeliveryWorker) handleDeliveringMsg(msg *models.QueuedMsg) {
  201.     now := time.Now().UnixNano()
  202.     if msg.UpdateTime + gDeliveryTimeoutNanos > now {
  203.         return
  204.     }
  206.     // delivery timeout
  207.     me.afterDeliveryTimeout(msg)
  208. }
  210. // afterDeliverySuccess: if done, move msg to success queue
  211. func (me *tDeliveryWorker) afterDeliverySuccess(msg *models.QueuedMsg) {
  212.     err := database.TX(func(db *sqlx.DB, tx *sqlx.Tx) error {
  213.         r,e := db.Exec(
  214.     "delete from delivery_queue where id=? and update_time=? and status_flag=1",
  215.             msg.ID,
  216.             msg.UpdateTime,
  217.         )
  218.         if e != nil {
  219.             return e
  220.         }
  222.         rows, e := r.RowsAffected()
  223.         if e != nil {
  224.             return e
  225.         }
  226.         if rows != 1 {
  227.             return gOneRowsErr
  228.         }
  230.         r, e = db.Exec(
  231.             "insert into success_queue (msg_id, client_id, create_time) values(?, ?, ?)",
  232.             msg.ID,
  233.             msg.ClientID,
  234.             time.Now().UnixNano(),
  235.         )
  236.         if e != nil {
  237.             return e
  238.         }
  240.         rows, e = r.RowsAffected()
  241.         if e != nil {
  242.             return e
  243.         }
  244.         if rows != 1 {
  245.             return gOneRowsErr
  246.         }
  248.         return nil
  249.     })
  251.     if err != nil {
  252.         logger.Logf("tDeliveryWorker.afterDeliverySuccess, failed, id=%v, msg=%s/%s, err=%s", me.info.ClientID, msg.GlobalID, msg.SubID, err.Error())
  253.     } else {
  254.         logger.Logf("tDeliveryWorker.afterDeliverySuccess, done, id=%v, msg=%s/%s", me.info.ClientID, msg.GlobalID, msg.SubID)
  255.     }
  256. }
  259. // afterDeliveryFailed: if failed, do nothing but just log it
  260. func (me *tDeliveryWorker) afterDeliveryFailed(msg *models.QueuedMsg) {
  261.     logger.Logf("tDeliveryWorker.afterDeliveryFailed, id=%v, msg=%s/%s", me.info.ClientID, msg.GlobalID, msg.SubID)
  262. }
  264. // afterDeliveryTimeout: if timeout, then reset status and retry
  265. func (me *tDeliveryWorker) afterDeliveryTimeout(msg *models.QueuedMsg) {
  266.     err := database.DB(func(db *sqlx.DB) error {
  267.         r,e := db.Exec(
  268.             "update delivery_queue set status_flag=0 where id=? and status_flag=1 and update_time=?",
  269.             msg.ID,
  270.             msg.UpdateTime,
  271.         )
  272.         if e != nil {
  273.             return e
  274.         }
  276.         rows,e := r.RowsAffected()
  277.         if e != nil {
  278.             return e
  279.         }
  281.         if rows != 1 {
  282.             return gOneRowsErr
  283.         }
  285.         return nil
  286.     })
  288.     if err != nil {
  289.         logger.Logf("tDeliveryWorker.afterDeliveryTimeout, failed, id=%v, msg=%s/%s, err=%s", me.info.ClientID, msg.GlobalID, msg.SubID, err.Error())
  290.     } else {
  291.         logger.Logf("tDeliveryWorker.afterDeliveryTimeout, done, id=%v, msg=%s/%s", me.info.ClientID, msg.GlobalID, msg.SubID)
  292.     }
  293. }
  295. var gEmptyRowsErr = errors.New("empty rows")
  296. var gOneRowsErr = errors.New("expecting one row affected")
  297. var gDeliveryTimeoutNanos = int64(10 * (time.Second / time.Nanosecond))

database.go

  • 数据库表相应的细节调整
    • delivery_queue: 去掉failed_count, 增加update_time时间戳
    • success_queue: 去掉sub_id, 改为client_id, 并增加create_time时间戳
    • failed_queue: 因为不允许失败, 因此删除失败投递表 ```go package database

import “github.com/jmoiron/sqlx” import _ “github.com/mattn/go-sqlite3”

type DBFunc func(db sqlx.DB) error type TXFunc func(db sqlx.DB, tx *sqlx.Tx) error

func init() { // must prepare tables err := DB(initDB) if err != nil { panic(err) } }

func initDB(db *sqlx.DB) error { // 订阅者/消费者: subscriber _, e := db.Exec(create table if not exists subscriber( id integer primary key autoincrement, client_id varchar(50) unique not null, topic varchar(100) not null, notify_url varchar(500) not null, expire_time integer )) if e != nil { return e }

  1. // 事务消息: tx_msg
  2. _, e = db.Exec(`create table if not exists tx_msg (
  3. id integer primary key autoincrement,
  4. global_id string varchar(50) not null,
  5. sub_id string varchar(50) unique not null,
  6. sender_id varchar(50) not null,
  7. create_time integer not null,
  8. topic varchar(100) not null,
  9. content nvarchar(2048) not null

)`) if e != nil { return e }

  1. // 投递队列: delivery_queue
  2. _, e = db.Exec(`create table if not exists delivery_queue (
  3. id integer primary key autoincrement,
  5. client_id varchar(50) not null,
  6. notify_url varchar(500) not null,
  8. msg_id integer not null,
  9. global_id string varchar(50) not null,
  10. sub_id string varchar(50) unique not null,
  11. sender_id varchar(50) not null,
  12. create_time integer not null,
  13. topic varchar(100) not null,
  14. content nvarchar(2048) not null,
  16. status_flag integer not null,
  17. update_time integer not null

)`) if e != nil { return e }

  1. // 成功投递队列: success_queue
  2. _, e = db.Exec(`create table if not exists success_queue (
  3. id integer primary key autoincrement,
  4. msg_id integer not null,
  5. client_id varchar(50) not null,
  6. create_time integer not null

)`) if e != nil { return e }

// // 投递失败队列: failedqueue // , e = db.Exec(create table if not exists failed_queue ( // id integer primary key autoincrement, // msg_id integer not null, // client_id varchar(50) not null, // create_time integer not null //)) // if e != nil { // return e // }

  1. return nil

}

func open() (*sqlx.DB, error) { return sqlx.Open(“sqlite3”, “./mqs.db”) }

func DB(action DBFunc) error { db,err := open() if err != nil { return err } defer func() { _ = db.Close() }()

  1. return action(db)

}

func TX(action TXFunc) error { return DB(func(db *sqlx.DB) error { tx, err := db.Beginx() if err != nil { return err }

  1.     err = action(db, tx)
  2.     if err == nil {
  3.         return tx.Commit()
  4.     } else {
  5.         return tx.Rollback()
  6.     }
  7. })

} ```

(未完待续)