缘起

最近阅读<> (刘金亮, 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.    - 确保投递成功: 投递队列持久化, 投递状态持久化, 失败重试
  9. <a name="naUyZ"></a>
  10. # 子目标(Day 2)
  12. - 消息的设计
  13. - 消息的持久化
  14. - 投递队列的持久化
  15. - 投递状态的持久化
  17. <a name="3CP5G"></a>
  18. # 设计
  20. - TxMsg: 事务消息模型
  21. - database: 添加事务消息表, 消息投递表, 成功投递表, 失败投递表
  22. - publish: 消息发布api
  23. - routers: 添加/publish路由
  25. <a name="YILum"></a>
  26. # TxMsg.go
  27. 事务消息模型
  28. ```go
  29. package models
  31. // 事务消息体
  32. type TxMsg struct {
  33.     // 全局事务ID
  34.     GlobalID string
  36.     // 子事务ID
  37.     SubID string
  39.     // 发送者ID
  40.     SenderID string
  42.     // 时间戳, 使用time.Now().UnixNano()
  43.     CreateTime int64
  45.     // 主题, 即消息类型
  46.     Topic string
  48.     // 消息内容, 一般是json
  49.     Content string
  50. }

database.go

添加事务消息表, 消息投递表, 成功投递表, 失败投递表

  1. package database
  3. import "github.com/jmoiron/sqlx"
  4. import _ "github.com/mattn/go-sqlite3"
  6. type DBFunc func(db *sqlx.DB) error
  7. type TXFunc func(db *sqlx.DB, tx *sqlx.Tx) error
  9. func init() {
  10.     // must prepare tables
  11.     err := DB(initDB)
  12.     if err != nil {
  13.         panic(err)
  14.     }
  15. }
  17. func initDB(db *sqlx.DB) error {
  18.     // 订阅者/消费者: subscriber
  19.     _, e := db.Exec(`create table if not exists subscriber(
  20.     id integer primary key autoincrement,
  21.     client_id varchar(50) unique not null,
  22.     topic varchar(100) not null,
  23.     notify_url varchar(500) not null,
  24.     expire_time integer
  25. )`)
  26.     if e != nil {
  27.         return e
  28.     }
  30.     // 事务消息: tx_msg
  31.     _, e = db.Exec(`create table if not exists tx_msg (
  32.     id integer primary key autoincrement,
  33.     global_id string varchar(50) not null,
  34.     sub_id string varchar(50) unique not null,
  35.     sender_id varchar(50) not null,
  36.     create_time integer not null,
  37.     topic varchar(100) not null,
  38.     content nvarchar(2048) not null
  39. )`)
  40.     if e != nil {
  41.         return e
  42.     }
  44.     // 投递队列: delivery_queue
  45.     _, e = db.Exec(`create table if not exists delivery_queue (
  46.     id integer primary key autoincrement,
  47.     msg_id integer not null,
  48.     sub_id integer not null,
  49.     status_flag integer not null,
  50.     failed_count integer not null
  51. )`)
  52.     if e != nil {
  53.         return e
  54.     }
  56.     // 成功投递队列: success_queue
  57.     _, e = db.Exec(`create table if not exists success_queue (
  58.     id integer primary key autoincrement,
  59.     msg_id integer not null,
  60.     sub_id integer not null
  61. )`)
  62.     if e != nil {
  63.         return e
  64.     }
  66.     // 投递失败队列: failed_queue
  67.     _, e = db.Exec(`create table if not exists failed_queue (
  68.     id integer primary key autoincrement,
  69.     msg_id integer not null,
  70.     sub_id integer not null
  71. )`)
  72.     if e != nil {
  73.         return e
  74.     }
  76.     return nil
  77. }
  79. func open() (*sqlx.DB, error) {
  80.     return sqlx.Open("sqlite3", "./mqs.db")
  81. }
  83. func DB(action DBFunc) error {
  84.     db,err := open()
  85.     if err != nil {
  86.         return err
  87.     }
  88.     defer func() { _ = db.Close() }()
  90.     return action(db)
  91. }
  93. func TX(action TXFunc) error {
  94.     return DB(func(db *sqlx.DB) error {
  95.         tx, err := db.Beginx()
  96.         if err != nil {
  97.             return err
  98.         }
  100.         err = action(db, tx)
  101.         if err == nil {
  102.             return tx.Commit()
  103.         } else {
  104.             return tx.Rollback()
  105.         }
  106.     })
  107. }

publish.go

消息发布api

  1. package handlers
  3. import (
  4.     "github.com/gin-gonic/gin"
  5.     "github.com/gin-gonic/gin/binding"
  6.     "github.com/jmoiron/sqlx"
  7.     "learning/gooop/saga/mqs/database"
  8.     "learning/gooop/saga/mqs/logger"
  9.     "learning/gooop/saga/mqs/models"
  10.     "net/http"
  11.     "time"
  12. )
  14. func Publish(c *gin.Context) {
  15.     // parse request
  16.     msg := &models.TxMsg{}
  17.     if err := c.ShouldBindBodyWith(&msg, binding.JSON); err != nil {
  18.         c.AbortWithStatusJSON(
  19.             http.StatusInternalServerError,
  20.             gin.H{ "ok": false, "error": err.Error()})
  21.         return
  22.     }
  24.     // fixme: validate msg
  26.     // save to db
  27.     ids := []int64{0}
  28.     err := database.TX(func(db *sqlx.DB, tx *sqlx.Tx) error {
  29.         id, e := saveTxMsg(db, tx, msg)
  30.         if e != nil {
  31.             return e
  32.         }
  34.         ids[0] = id
  35.         return nil
  36.     })
  38.     if ids[0] > 0 {
  39.         newMsgId := ids[0]
  40.         logger.Logf("publish new msg: %d", newMsgId)
  42.         // todo: 新增消息, 开始投送
  43.     }
  45.     // reply
  46.     if err != nil {
  47.         c.JSON(http.StatusInternalServerError, gin.H{ "ok": false, "error": err.Error()})
  48.     } else {
  49.         c.JSON(http.StatusOK, gin.H { "ok": true })
  50.     }
  51. }
  54. func saveTxMsg(db *sqlx.DB, tx *sqlx.Tx, msg *models.TxMsg) (int64,error) {
  55.     // insert tx_msg
  56.     r, e := db.Exec(
  57.         `replace into tx_msg(global_id, sub_id, sender_id, create_time, topic, content) values(?,?,?,?,?,?)`,
  58.         msg.GlobalID, msg.SubID, msg.SenderID, msg.CreateTime, msg.Topic, msg.Content,
  59.     )
  60.     if e != nil {
  61.         return 0,e
  62.     }
  64.     // get last insert id
  65.     id,e := r.LastInsertId()
  66.     if e != nil {
  67.         return 0,e
  68.     }
  70.     if id > 0 {
  71.         // copy to delivery queue
  72.         now := time.Now().UnixNano()
  73.         r, e = db.Exec(`
  74. insert into delivery_queue(msg_id, sub_id, status_flag, failed_count) 
  75. select
  76.     ?, s.id, 0, 0 
  77. from
  78.     sub_info s
  79. where
  80.     s.expire_time>?
  81. `, id, now)
  82.         if e != nil {
  83.             return 0,e
  84.         }
  85.     }
  87.     return id, nil
  88. }

routers.go

添加/publish路由

  1. package routers
  3. import (
  4.     "github.com/gin-gonic/gin"
  5.     "learning/gooop/saga/mqs/handlers"
  6. )
  8. func RegisterRouters() *gin.Engine {
  9.     r := gin.Default()
  10.     r.Use(gin.Logger())
  12.     r.GET("/ping", handlers.Ping)
  13.     r.POST("/subscribe", handlers.Subscribe)
  14.     r.POST("/publish", handlers.Publish)
  16.     return r
  17. }

(未完待续)