缘起

最近阅读<> (刘金亮, 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 6)
  13. - 终于可以进行基本功能测试了
  14.    - 订阅接口测试: /subscribe
  15.    - 发布接口测试: /publish
  16.    - 通知接口测试: /notify
  17. - 重构
  18.    - database.go: 重构DDL语句, 以兼容sqlx的StructScan字段映射
  19.    - 所有Queryx返回的rows, 必须加上defer rows.Close()
  20.    - 添加大量的过程诊断日志
  21.    - tLiveMsgSource: fix handleMsgPublished未接收数据就return的bug
  23. <a name="mLMji"></a>
  24. # 单元测试
  25. mqs_test.go, 依次清空数据库, 测试订阅接口, 发布接口和通知接口, 并诊断过程日志和数据库记录变化.
  26. ```go
  27. package saga
  29. import (
  30.     "bytes"
  31.     "encoding/json"
  32.     "errors"
  33.     "fmt"
  34.     "github.com/jmoiron/sqlx"
  35.     "io/ioutil"
  36.     "learning/gooop/saga/mqs/cmd"
  37.     "learning/gooop/saga/mqs/database"
  38.     "learning/gooop/saga/mqs/logger"
  39.     "learning/gooop/saga/mqs/models"
  40.     "net/http"
  41.     "sync"
  42.     "testing"
  43.     "time"
  44. )
  47. var gRunOnce sync.Once
  48. func fnBootMQS() {
  49.     gRunOnce.Do(func() {
  50.         // boot mqs
  51.         go cmd.BootMQS()
  53.         // wait for mqs up
  54.         time.Sleep(1 * time.Second)
  55.     })
  56. }
  58. func fnAssertTrue (t *testing.T, b bool, msg string) {
  59.     if !b {
  60.         t.Fatal(msg)
  61.     }
  62. }
  64. func Test_MQS(t *testing.T) {
  65.     // prepare mqs
  66.     fnClearDB(t)
  67.     fnBootMQS()
  69.     // subscribe
  70.     fnTestSubscribe(t)
  71.     t.Log("passed fnTestSubscribe")
  73.     // publish and notify
  74.     fnTestPublishAndNotify(t)
  75.     t.Log("passed fnTestPublishAndNotify")
  76. }
  79. func fnTestPublishAndNotify(t *testing.T) {
  80.     t.Log("testing fnTestPublishAndNotify")
  82.     msg := &models.TxMsg{
  83.         GlobalID: "test-global-id",
  84.         SubID: "test-sub-id",
  85.         SenderID: "test-sender-id",
  86.         Topic: "test-topic",
  87.         Content: "test-content",
  88.     }
  89.     fnPost(t, msg, "http://localhost:3333/publish")
  91.     // check log
  92.     fnAssertTrue(t, logger.Count("handlers.Publish, msg=test-global-id")==1, "expecting log: handlers.Publish, msg=test-global-id")
  94.     // check notify
  95.     time.Sleep(100 * time.Millisecond)
  96.     fnAssertTrue(t, logger.Count("tLiveMsgSource.handleMsgPublished, clientID=test-client, msg=test-global-id")==1, "expecting log: tLiveMsgSource.handleMsgPublished")
  97.     fnAssertTrue(t, logger.Count("tDeliveryWorker.afterDeliverySuccess, done, id=test-client, msg=test-global-id")==1, "expecting log: tDeliveryWorker.afterDeliverySuccess")
  98.     fnAssertTrue(t, logger.Count("handlers.Notify, msg=")==1, "expecting log: handlers.Notify, msg=")
  100.     // check success queue
  101.     fnAssertTrue(t, fnDBCount(t, "select count(1) from success_queue where ClientID='test-client'")==1, "expectiang db.success_queue.count == 1")
  102. }
  105. func fnClearDB(t *testing.T) {
  106.     fnDBExec(t, "delete from subscriber")
  107.     fnDBExec(t, "delete from tx_msg")
  108.     fnDBExec(t, "delete from delivery_queue")
  109.     fnDBExec(t, "delete from success_queue")
  110. }
  113. func fnDBCount(t *testing.T, sql string, args... interface{}) int {
  114.     sum := []int{ 0 }
  115.     err := database.DB(func(db *sqlx.DB) error {
  116.         r,e := db.Queryx(sql, args...)
  117.         if e != nil {
  118.             return e
  119.         }
  121.         defer r.Close()
  122.         if !r.Next() {
  123.             return errors.New("empty rows")
  124.         }
  126.         e = r.Scan(&sum[0])
  127.         if e != nil {
  128.             return e
  129.         }
  131.         return nil
  132.     })
  134.     if err != nil {
  135.         t.Fatal(err)
  136.     }
  138.     return sum[0]
  139. }
  142. func fnDBExec(t *testing.T, sql string, args... interface{}) int {
  143.     rows := []int64{ 0 }
  144.     err := database.DB(func(db *sqlx.DB) error {
  145.         r,e := db.Exec(sql, args...)
  146.         if e != nil {
  147.             return e
  148.         }
  150.         rows[0], e = r.RowsAffected()
  151.         if e != nil {
  152.             return e
  153.         }
  155.         return nil
  156.     })
  158.     if err != nil {
  159.         t.Fatal(err)
  160.     }
  161.     return int(rows[0])
  162. }
  164. func fnTestSubscribe(t *testing.T) {
  165.     t.Log("testing fnTestSubscribe")
  167.     // clear subscriber
  168.     fnDBExec(t, "delete from subscriber")
  170.     msg := &models.SubscribeMsg{
  171.         ClientID: "test-client",
  172.         Topic: "test-topic",
  173.         NotifyUrl: "http://localhost:3333/notify",
  174.         ExpireTime: time.Now().UnixNano() + int64(30*time.Second),
  175.     }
  176.     fnPost(t, msg, "http://localhost:3333/subscribe")
  178.     // check log
  179.     fnAssertTrue(t, logger.Count("handlers.Subscribe, event=subscriber.registered") == 1, "expecting event=subscriber.registered")
  181.     // check db
  182.     count := fnDBCount(t, "select count(1) as n from subscriber where ClientID=? and topic=?", msg.ClientID, msg.Topic)
  183.     fnAssertTrue(t, count == 1, "expecting subscriber.count == 1")
  184. }
  187. func fnPost(t *testing.T, msg interface{}, url string) {
  188.     body,_ := json.Marshal(msg)
  189.     rsp, err := http.Post(url, "application/json;charset=utf-8", bytes.NewReader(body))
  190.     if err != nil {
  191.         t.Fatal(err)
  192.     }
  194.     defer rsp.Body.Close()
  195.     j, err := ioutil.ReadAll(rsp.Body)
  196.     if err != nil {
  197.         t.Fatal(err)
  198.     }
  199.     ok := &models.OkMsg{}
  200.     err = json.Unmarshal(j, ok)
  201.     if err != nil {
  202.         t.Fatal(err)
  203.     }
  204.     fnAssertTrue(t, ok.OK, fmt.Sprintf("expecting replying ok from %s", url))
  205. }

测试输出

根据assert失败提示, 逐步排查诊断日志, 主要是sqlx字段映射错误, 和未及时调用rows.Close()错误. 

  1. $ go test -v mqs_test.go 
  2. === RUN   Test_MQS
  3. eventbus.Pub, event=system.boot, handler=gDeliveryService.handleBootEvent
  4. [GIN-debug] [WARNING] Creating an Engine instance with the Logger and Recovery middleware already attached.
  6. [GIN-debug] [WARNING] Running in "debug" mode. Switch to "release" mode in production.
  7.  - using env:   export GIN_MODE=release
  8.  - using code:  gin.SetMode(gin.ReleaseMode)
  10. [GIN-debug] GET    /ping                     --> learning/gooop/saga/mqs/handlers.Ping (4 handlers)
  11. [GIN-debug] POST   /subscribe                --> learning/gooop/saga/mqs/handlers.Subscribe (4 handlers)
  12. [GIN-debug] POST   /publish                  --> learning/gooop/saga/mqs/handlers.Publish (4 handlers)
  13. [GIN-debug] POST   /notify                   --> learning/gooop/saga/mqs/handlers.Notify (4 handlers)
  14. [GIN-debug] Listening and serving HTTP on :3333
  15. tDeliveryService.beginCreatingWorkers
  16. tDeliveryService.beginCleanExpiredWorkers
  17.     mqs_test.go:139: testing fnTestSubscribe
  18. handlers.Subscribe, event=subscriber.registered, msg=&{test-client test-topic http://localhost:3333/notify 1615868155394998875}
  19. eventbus.Pub, event=subscriber.registered, handler=gDeliveryService.handleSubscriberRegistered
  20. [GIN] 2021/03/16 - 12:15:25 | 200 |    8.118873ms |             ::1 | POST     "/subscribe"
  21. [GIN] 2021/03/16 - 12:15:25 | 200 |    8.214968ms |             ::1 | POST     "/subscribe"
  22. tDeliveryWorker.afterInitialLoad, clientID=test-client, rows=0
  23. database.DB, err=empty rows
  24.     mqs_test.go:45: passed fnTestSubscribe
  25.     mqs_test.go:54: testing fnTestPublishAndNotify
  26. handlers.Publish, msg=test-global-id/test-sub-id/test-topic, msgId=15
  27. [GIN] 2021/03/16 - 12:15:25 | 200 |   15.200109ms |             ::1 | POST     "/publish"
  28. [GIN] 2021/03/16 - 12:15:25 | 200 |   15.216578ms |             ::1 | POST     "/publish"
  29. handlers.Publish, pubLiveMsg 15
  30. handlers.Publish, pubLiveMsg, msgId=15, rows=1
  31. handlers.Publish, event=msg.published, clientID=test-client, msg=test-global-id/test-sub-id/http://localhost:3333/notify
  32. eventbus.Pub, event=msg.published, handler=tLiveMsgSource.test-client
  33. tLiveMsgSource.handleMsgPublished, clientID=test-client, msg=test-global-id/test-sub-id/test-topic
  34. handlers.Notify, msg=&{test-global-id test-sub-id  0 test-topic test-content}
  35. [GIN] 2021/03/16 - 12:15:25 | 200 |       48.38µs |             ::1 | POST     "/notify"
  36. [GIN] 2021/03/16 - 12:15:25 | 200 |     110.659µs |             ::1 | POST     "/notify"
  37. tDeliveryWorker.afterDeliverySuccess, done, id=test-client, msg=test-global-id/test-sub-id
  38.     mqs_test.go:49: passed fnTestPublishAndNotify
  39. --- PASS: Test_MQS (1.18s)
  40. PASS
  41. ok      command-line-arguments  1.187s

(未完待续)