缘起

最近阅读<> (刘金亮, 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 8)
  13. - 创建虚拟的库存服务
  14.    - 启动时, 注册到MQ
  15.    - 接收到订单创建的消息时, 扣减库存
  16.    - 扣库成功时, 经MQ通知订单服务扣库成功
  17.    - 扣库失败时, 经MQ通知订单服务扣库失败
  19. <a name="KVAqs"></a>
  20. # 设计
  22. - IStockService: 模拟的库存服务接口
  23. - tStockService: 虚拟库存服务, 实现IStockService接口
  24. - NotifySaleOrderCreated: 用于监听订单创建消息的http回调处理器
  26. <a name="qkOFr"></a>
  27. # 单元测试
  28. order_test.go
  30. 1. 初始化10个产品库存
  31. 1. 订单服务, 创建订单1, 尝试扣减1个库存, 预期成功
  32. 1. 订单服务, 创建订单2, 尝试扣减10个库存, 预期失败
  33. 1. 校验订单1的最终状态为出库成功
  34. 1. 校验订单2的最终状态为出库失败
  35. ```go
  36. package saga
  38. import (
  39.     "github.com/jmoiron/sqlx"
  40.     "learning/gooop/saga/mqs/cmd"
  41.     "learning/gooop/saga/mqs/database"
  42.     "learning/gooop/saga/mqs/logger"
  43.     "learning/gooop/saga/order"
  44.     "learning/gooop/saga/stock"
  45.     "sync"
  46.     "testing"
  47.     "time"
  48. )
  50. var gRunOnce sync.Once
  51. func fnBootMQS() {
  52.     gRunOnce.Do(func() {
  53.         // boot mqs
  54.         go cmd.BootMQS()
  56.         // wait for mqs up
  57.         time.Sleep(1 * time.Second)
  58.     })
  59. }
  61. func fnAssertTrue (t *testing.T, b bool, msg string) {
  62.     if !b {
  63.         t.Fatal(msg)
  64.     }
  65. }
  67. func Test_SagaSaleOrder(t *testing.T) {
  68.     // prepare mqs
  69.     fnClearDB(t)
  70.     fnBootMQS()
  72.     // 1 create prod stock
  73.     prodID := "test-prod-1"
  74.     err := stock.MockStockService.AddStock(prodID, 10)
  75.     if err != nil {
  76.         t.Fatal(err)
  77.     }
  79.     // create order 1
  80.     o1 := &order.SaleOrder{
  81.         OrderID: "test-order-1",
  82.         ProductID: prodID,
  83.         CustomerID: "test-customer-1",
  84.         Quantity: 1,
  85.         Price: 100,
  86.         Amount: 100,
  87.         CreateTime: time.Now().UnixNano(),
  88.         StatusFlag: order.StatusNotDelivered,
  89.     }
  90.     err = order.MockSaleOrderService.Create(o1)
  91.     if err != nil {
  92.         t.Fatal(err)
  93.     }
  95.     // create order 2
  96.     time.Sleep(10*time.Millisecond)
  97.     o2 := &order.SaleOrder{
  98.         OrderID: "test-order-2",
  99.         ProductID: prodID,
  100.         CustomerID: "test-customer-2",
  101.         Quantity: 10,
  102.         Price: 100,
  103.         Amount: 1000,
  104.         CreateTime: time.Now().UnixNano(),
  105.         StatusFlag: order.StatusNotDelivered,
  106.     }
  107.     err = order.MockSaleOrderService.Create(o2)
  108.     if err != nil {
  109.         t.Fatal(err)
  110.     }
  112.     time.Sleep(1 * time.Second)
  113.     logger.Logf("============================================")
  114.     log := "tSaleOrderService.beginSubscribeMQ, done"
  115.     fnAssertTrue(t, logger.Count(log)==1, "expecting log: " + log)
  117.     log = "tSaleOrderService.publishMQ, done, order=test-order-1"
  118.     fnAssertTrue(t, logger.Count(log)==1, "expecting log: " + log)
  120.     log = "tSaleOrderService.publishMQ, done, order=test-order-2"
  121.     fnAssertTrue(t, logger.Count(log)==1, "expecting log: " + log)
  123.     log = "stock.NotifySaleOrderCreated, order=test-order-1"
  124.     fnAssertTrue(t, logger.Count(log)==1, "expecting log: " + log)
  126.     log = "stock.NotifySaleOrderCreated, order=test-order-2"
  127.     fnAssertTrue(t, logger.Count(log)==1, "expecting log: " + log)
  129.     o1 = order.MockSaleOrderService.Get(o1.OrderID)
  130.     fnAssertTrue(t, o1.StatusFlag == order.StatusStockOutboundDone, "expecting o1 done")
  132.     o2 = order.MockSaleOrderService.Get(o2.OrderID)
  133.     fnAssertTrue(t, o2.StatusFlag == order.StatusStockOutboundFailed, "expecting o2 failed")
  135.     logger.Logf("test passed")
  136. }
  139. func fnClearDB(t *testing.T) {
  140.     fnDBExec(t, "delete from subscriber")
  141.     fnDBExec(t, "delete from tx_msg")
  142.     fnDBExec(t, "delete from delivery_queue")
  143.     fnDBExec(t, "delete from success_queue")
  144. }
  147. func fnDBExec(t *testing.T, sql string, args... interface{}) int {
  148.     rows := []int64{ 0 }
  149.     err := database.DB(func(db *sqlx.DB) error {
  150.         r,e := db.Exec(sql, args...)
  151.         if e != nil {
  152.             return e
  153.         }
  155.         rows[0], e = r.RowsAffected()
  156.         if e != nil {
  157.             return e
  158.         }
  160.         return nil
  161.     })
  163.     if err != nil {
  164.         t.Fatal(err)
  165.     }
  166.     return int(rows[0])
  167. }

测试输出

  1. $ go test -v order_test.go 
  2. === RUN   Test_SagaSaleOrder
  3. 23:55:54.292132442 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] POST   /notify/sale-order.stock.outbound --> learning/gooop/saga/order.NotifyStockOutbound (4 handlers)
  15. [GIN-debug] POST   /notify/sale-order.created --> learning/gooop/saga/stock.NotifySaleOrderCreated (4 handlers)
  16. [GIN-debug] Listening and serving HTTP on :3333
  17. 23:55:54.292287032 tDeliveryService.beginCleanExpiredWorkers
  18. 23:55:54.292345845 tDeliveryService.beginCreatingWorkers
  19. 23:55:54.356542981 handlers.Subscribe, msg=&{sale-order-service sale-order.stock.outbound http://localhost:3333/notify/sale-order.stock.outbound 1616086554355593476}
  20. 23:55:54.356524325 handlers.Subscribe, msg=&{stock-service sale-order.created http://localhost:3333/notify/sale-order.created 1616086554355598830}
  21. 23:55:54.365256441 handlers.Subscribe, event=subscriber.registered, msg=&{sale-order-service sale-order.stock.outbound http://localhost:3333/notify/sale-order.stock.outbound 1616086554355593476}
  22. 23:55:54.365271105 eventbus.Pub, event=subscriber.registered, handler=gDeliveryService.handleSubscriberRegistered
  23. [GIN] 2021/03/18 - 23:55:54 | 200 |    8.865173ms |             ::1 | POST     "/subscribe"
  24. [GIN] 2021/03/18 - 23:55:54 | 200 |    8.882138ms |             ::1 | POST     "/subscribe"
  25. 23:55:54.365488163 tSaleOrderService.beginSubscribeMQ, done
  26. 23:55:54.365861542 database.DB, err=empty rows
  27. 23:55:54.366239244 tDeliveryWorker.afterInitialLoad, clientID=sale-order-service, rows=0
  28. 23:55:54.373588493 handlers.Subscribe, event=subscriber.registered, msg=&{stock-service sale-order.created http://localhost:3333/notify/sale-order.created 1616086554355598830}
  29. 23:55:54.373605972 eventbus.Pub, event=subscriber.registered, handler=gDeliveryService.handleSubscriberRegistered
  30. [GIN] 2021/03/18 - 23:55:54 | 200 |   17.189632ms |             ::1 | POST     "/subscribe"
  31. [GIN] 2021/03/18 - 23:55:54 | 200 |   17.205549ms |             ::1 | POST     "/subscribe"
  32. 23:55:54.373843032 tStockService.beginSubscribeMQ, done
  33. 23:55:54.3743926 database.DB, err=empty rows
  34. 23:55:54.374499757 tDeliveryWorker.afterInitialLoad, clientID=stock-service, rows=0
  35. 23:55:55.292336699 tStockService.AddStock, done, prodId=test-prod-1, stock=0, delta=0, after=10
  36. 23:55:55.323746568 handlers.Publish, msg=test-order-1/test-order-1/sale-order.created, msgId=112
  37. [GIN] 2021/03/18 - 23:55:55 | 200 |   31.112478ms |             ::1 | POST     "/publish"
  38. [GIN] 2021/03/18 - 23:55:55 | 200 |   31.125855ms |             ::1 | POST     "/publish"
  39. 23:55:55.323811205 handlers.Publish, pubLiveMsg 112
  40. 23:55:55.323910377 tSaleOrderService.publishMQ, done, order=test-order-1/&{test-order-1 test-customer-1 test-prod-1 1 100 100 1616082955292352151 0}
  41. 23:55:55.324227736 handlers.Publish, pubLiveMsg, msgId=112, rows=1
  42. 23:55:55.324273573 handlers.Publish, event=msg.published, clientID=stock-service, msg=test-order-1/test-order-1/http://localhost:3333/notify/sale-order.created
  43. 23:55:55.32428051 eventbus.Pub, event=msg.published, handler=tLiveMsgSource.sale-order-service
  44. 23:55:55.324285512 eventbus.Pub, event=msg.published, handler=tLiveMsgSource.stock-service
  45. 23:55:55.324292286 tLiveMsgSource.handleMsgPublished, clientID=stock-service, msg=test-order-1/test-order-1/sale-order.created
  46. 23:55:55.324346678 tDeliveryWorker.beginPollAndDeliver, msg from live=&{98 stock-service http://localhost:3333/notify/sale-order.created 112 test-order-1 test-order-1 sale-order-service 1616082955292352151 sale-order.created {"OrderID":"test-order-1","CustomerID":"test-customer-1","ProductID":"test-prod-1","Quantity":1,"Price":100,"Amount":100,"CreateTime":1616082955292352151,"StatusFlag":0} 0 0}
  47. 23:55:55.33925766 handlers.Publish, msg=test-order-2/test-order-2/sale-order.created, msgId=113
  48. [GIN] 2021/03/18 - 23:55:55 | 200 |   15.264561ms |             ::1 | POST     "/publish"
  49. [GIN] 2021/03/18 - 23:55:55 | 200 |   15.280884ms |             ::1 | POST     "/publish"
  50. 23:55:55.339353768 handlers.Publish, pubLiveMsg 113
  51. 23:55:55.339446893 tSaleOrderService.publishMQ, done, order=test-order-2/&{test-order-2 test-customer-2 test-prod-1 10 100 1000 1616082955302734821 0}
  52. 23:55:55.339909493 handlers.Publish, pubLiveMsg, msgId=113, rows=1
  53. 23:55:55.339919874 handlers.Publish, event=msg.published, clientID=stock-service, msg=test-order-2/test-order-2/http://localhost:3333/notify/sale-order.created
  54. 23:55:55.339925049 eventbus.Pub, event=msg.published, handler=tLiveMsgSource.sale-order-service
  55. 23:55:55.339929964 eventbus.Pub, event=msg.published, handler=tLiveMsgSource.stock-service
  56. 23:55:55.339935935 tLiveMsgSource.handleMsgPublished, clientID=stock-service, msg=test-order-2/test-order-2/sale-order.created
  57. 23:55:55.350117186 tDeliveryWorker.deliver, begin, id=stock-service, msg=test-order-1/test-order-1
  58. 23:55:55.35041833 stock.NotifySaleOrderCreated, order=test-order-1/&{test-order-1 test-customer-1 test-prod-1 1 100 100 1616082955292352151 0}
  59. 23:55:55.350429178 tStockService.AddStock, done, prodId=test-prod-1, stock=10, delta=-1, after=9
  60. [GIN] 2021/03/18 - 23:55:55 | 200 |      88.872µs |             ::1 | POST     "/notify/sale-order.created"
  61. [GIN] 2021/03/18 - 23:55:55 | 200 |     133.617µs |             ::1 | POST     "/notify/sale-order.created"
  62. 23:55:55.350592351 tDeliveryWorker.deliver, OK, id=stock-service, msg=test-order-1/test-order-1
  63. 23:55:55.367336707 tDeliveryWorker.afterDeliverySuccess, done, id=stock-service, msg=test-order-1/test-order-1
  64. 23:55:55.36738322 tDeliveryWorker.beginPollAndDeliver, msg from live=&{99 stock-service http://localhost:3333/notify/sale-order.created 113 test-order-2 test-order-2 sale-order-service 1616082955302734821 sale-order.created {"OrderID":"test-order-2","CustomerID":"test-customer-2","ProductID":"test-prod-1","Quantity":10,"Price":100,"Amount":1000,"CreateTime":1616082955302734821,"StatusFlag":0} 0 0}
  65. 23:55:55.367530495 database.DB, err=empty rows
  66. 23:55:55.374978535 tDeliveryWorker.deliver, begin, id=stock-service, msg=test-order-2/test-order-2
  67. 23:55:55.375201115 stock.NotifySaleOrderCreated, order=test-order-2/&{test-order-2 test-customer-2 test-prod-1 10 100 1000 1616082955302734821 0}
  68. 23:55:55.375211216 tStockService.AddStock, failed, prodId=test-prod-1, stock=9, delta=-10
  69. 23:55:55.375219558 tStockService.HandleSaleOrderCreated, err=insufficient stock, order=&{test-order-2 test-customer-2 test-prod-1 10 100 1000 1616082955302734821 0}
  70. [GIN] 2021/03/18 - 23:55:55 | 200 |      102.52µs |             ::1 | POST     "/notify/sale-order.created"
  71. [GIN] 2021/03/18 - 23:55:55 | 200 |     116.933µs |             ::1 | POST     "/notify/sale-order.created"
  72. 23:55:55.375354895 tDeliveryWorker.deliver, OK, id=stock-service, msg=test-order-2/test-order-2
  73. 23:55:55.389901711 tDeliveryWorker.afterDeliverySuccess, done, id=stock-service, msg=test-order-2/test-order-2
  74. 23:55:55.38993077 tDeliveryWorker.beginPollAndDeliver, msg from db=&{99 stock-service http://localhost:3333/notify/sale-order.created 113 test-order-2 test-order-2 sale-order-service 1616082955302734821 sale-order.created {"OrderID":"test-order-2","CustomerID":"test-customer-2","ProductID":"test-prod-1","Quantity":10,"Price":100,"Amount":1000,"CreateTime":1616082955302734821,"StatusFlag":0} 1 1616082955367401386}
  75. 23:55:55.420121681 handlers.Publish, msg=test-order-1/test-order-1.outbound/sale-order.stock.outbound, msgId=114
  76. [GIN] 2021/03/18 - 23:55:55 | 200 |   69.507171ms |             ::1 | POST     "/publish"
  77. [GIN] 2021/03/18 - 23:55:55 | 200 |   69.520805ms |             ::1 | POST     "/publish"
  78. 23:55:55.420220719 handlers.Publish, pubLiveMsg 114
  79. 23:55:55.420321792 tStockService.publishMQ, done, msg=&{test-order-1 test-order-1.outbound stock-service 1616082955350432496 sale-order.stock.outbound 1}
  80. 23:55:55.42071623 handlers.Publish, pubLiveMsg, msgId=114, rows=1
  81. 23:55:55.420731889 handlers.Publish, event=msg.published, clientID=sale-order-service, msg=test-order-1/test-order-1.outbound/http://localhost:3333/notify/sale-order.stock.outbound
  82. 23:55:55.420741935 eventbus.Pub, event=msg.published, handler=tLiveMsgSource.sale-order-service
  83. 23:55:55.420746401 eventbus.Pub, event=msg.published, handler=tLiveMsgSource.stock-service
  84. 23:55:55.420755367 tLiveMsgSource.handleMsgPublished, clientID=sale-order-service, msg=test-order-1/test-order-1.outbound/sale-order.stock.outbound
  85. 23:55:55.42079505 tDeliveryWorker.beginPollAndDeliver, msg from live=&{100 sale-order-service http://localhost:3333/notify/sale-order.stock.outbound 114 test-order-1 test-order-1.outbound stock-service 1616082955350432496 sale-order.stock.outbound 1 0 0}
  86. 23:55:55.435844021 handlers.Publish, msg=test-order-2/test-order-2.outbound/sale-order.stock.outbound, msgId=115
  87. [GIN] 2021/03/18 - 23:55:55 | 200 |   15.407267ms |             ::1 | POST     "/publish"
  88. [GIN] 2021/03/18 - 23:55:55 | 200 |   15.420327ms |             ::1 | POST     "/publish"
  89. 23:55:55.4359058 handlers.Publish, pubLiveMsg 115
  90. 23:55:55.436026025 tStockService.publishMQ, done, msg=&{test-order-2 test-order-2.outbound stock-service 1616082955375214295 sale-order.stock.outbound 0}
  91. 23:55:55.436398324 handlers.Publish, pubLiveMsg, msgId=115, rows=1
  92. 23:55:55.436409937 handlers.Publish, event=msg.published, clientID=sale-order-service, msg=test-order-2/test-order-2.outbound/http://localhost:3333/notify/sale-order.stock.outbound
  93. 23:55:55.43642793 eventbus.Pub, event=msg.published, handler=tLiveMsgSource.sale-order-service
  94. 23:55:55.436433697 eventbus.Pub, event=msg.published, handler=tLiveMsgSource.stock-service
  95. 23:55:55.43644379 tLiveMsgSource.handleMsgPublished, clientID=sale-order-service, msg=test-order-2/test-order-2.outbound/sale-order.stock.outbound
  96. 23:55:55.446599314 tDeliveryWorker.deliver, begin, id=sale-order-service, msg=test-order-1/test-order-1.outbound
  97. 23:55:55.446809726 order.NotifyStockOutbound, orderID=test-order-1, succeeded=true
  98. [GIN] 2021/03/18 - 23:55:55 | 200 |      61.898µs |             ::1 | POST     "/notify/sale-order.stock.outbound"
  99. [GIN] 2021/03/18 - 23:55:55 | 200 |      81.911µs |             ::1 | POST     "/notify/sale-order.stock.outbound"
  100. 23:55:55.446951354 tDeliveryWorker.deliver, OK, id=sale-order-service, msg=test-order-1/test-order-1.outbound
  101. 23:55:55.462584405 tDeliveryWorker.afterDeliverySuccess, done, id=sale-order-service, msg=test-order-1/test-order-1.outbound
  102. 23:55:55.462615131 tDeliveryWorker.beginPollAndDeliver, msg from live=&{101 sale-order-service http://localhost:3333/notify/sale-order.stock.outbound 115 test-order-2 test-order-2.outbound stock-service 1616082955375214295 sale-order.stock.outbound 0 0 0}
  103. 23:55:55.469999185 tDeliveryWorker.deliver, begin, id=sale-order-service, msg=test-order-2/test-order-2.outbound
  104. 23:55:55.470163043 order.NotifyStockOutbound, orderID=test-order-2, succeeded=false
  105. [GIN] 2021/03/18 - 23:55:55 | 200 |       85.14µs |             ::1 | POST     "/notify/sale-order.stock.outbound"
  106. [GIN] 2021/03/18 - 23:55:55 | 200 |     105.638µs |             ::1 | POST     "/notify/sale-order.stock.outbound"
  107. 23:55:55.470369408 tDeliveryWorker.deliver, OK, id=sale-order-service, msg=test-order-2/test-order-2.outbound
  108. 23:55:55.486229145 tDeliveryWorker.afterDeliverySuccess, done, id=sale-order-service, msg=test-order-2/test-order-2.outbound
  109. 23:55:56.302885199 ============================================
  110. 23:55:56.303470422 test passed
  111. --- PASS: Test_SagaSaleOrder (2.05s)
  112. PASS
  113. ok      command-line-arguments  2.057s

IStockService.go

模拟的库存服务接口

  1. package stock;
  3. import "learning/gooop/saga/order"
  5. type IStockService interface {
  6.     GetStock(prodId string) int
  7.     AddStock(prodId string, delta int) error
  8.     HandleSaleOrderCreated(it *order.SaleOrder) error
  9. }

tStockService.go

虚拟库存服务, 实现IStockService接口

  1. package stock
  3. import (
  4.     "bytes"
  5.     "encoding/json"
  6.     "errors"
  7.     "io/ioutil"
  8.     "learning/gooop/saga/mqs/logger"
  9.     "learning/gooop/saga/mqs/models"
  10.     "learning/gooop/saga/order"
  11.     "net/http"
  12.     "sync"
  13.     "time"
  14. )
  16. type tStockService struct {
  17.     rwmutex *sync.RWMutex
  18.     stock map[string]int
  19.     bMQReady bool
  20.     publishQueue chan *models.TxMsg
  21. }
  23. func newStockService() IStockService {
  24.     it := new(tStockService)
  25.     it.init()
  26.     return it
  27. }
  30. func (me *tStockService) init() {
  31.     me.rwmutex = new(sync.RWMutex)
  32.     me.stock = make(map[string]int)
  33.     me.bMQReady = false
  34.     me.publishQueue = make(chan *models.TxMsg, gMQMaxQueuedMsg)
  36.     go func() {
  37.         time.Sleep(100*time.Millisecond)
  38.         go me.beginSubscribeMQ()
  39.         go me.beginPublishMQ()
  40.     }()
  41. }
  43. func (me *tStockService) GetStock(prodId string) int {
  44.     me.rwmutex.RLock()
  45.     defer me.rwmutex.RUnlock()
  47.     it,ok := me.stock[prodId]
  48.     if ok {
  49.         return it
  50.     } else {
  51.         return 0
  52.     }
  53. }
  55. func (me *tStockService) AddStock(prodId string, delta int) error {
  56.     me.rwmutex.RLock()
  57.     defer me.rwmutex.RUnlock()
  59.     it,ok := me.stock[prodId]
  60.     if ok {
  61.         n := it + delta
  62.         if n < 0 {
  63.             logger.Logf("tStockService.AddStock, failed, prodId=%s, stock=%d, delta=%d", prodId, it, delta)
  64.             return gInsufficientStockError
  65.         } else {
  66.             logger.Logf("tStockService.AddStock, done, prodId=%s, stock=%d, delta=%d, after=%d", prodId, it, delta, n)
  67.             me.stock[prodId] = n
  68.         }
  69.     } else {
  70.         if delta < 0 {
  71.             logger.Logf("tStockService.AddStock, failed, prodId=%s, stock=0, delta=%d", prodId, delta)
  72.             return gInsufficientStockError
  73.         } else {
  74.             logger.Logf("tStockService.AddStock, done, prodId=%s, stock=0, delta=%d, after=%d", prodId, it, delta)
  75.             me.stock[prodId] = delta
  76.         }
  77.     }
  79.     return nil
  80. }
  83. func (me *tStockService) beginSubscribeMQ() {
  84.     expireDuration := int64(1 * time.Hour)
  85.     subscribeDuration := 20 * time.Minute
  86.     pauseDuration := 3*time.Second
  87.     lastSubscribeTime := int64(0)
  89.     for {
  90.         now := time.Now().UnixNano()
  91.         if now - lastSubscribeTime >= int64(subscribeDuration) {
  92.             expireTime := now + expireDuration
  93.             err := fnSubscribeMQ(expireTime)
  95.             if err != nil {
  96.                 me.bMQReady = false
  97.                 logger.Logf("tStockService.beginSubscribeMQ, failed, err=%v", err)
  99.             } else {
  100.                 lastSubscribeTime = now
  101.                 me.bMQReady = true
  102.                 logger.Logf("tStockService.beginSubscribeMQ, done")
  103.             }
  104.         }
  105.         time.Sleep(pauseDuration)
  106.     }
  107. }
  109. func fnSubscribeMQ(expireTime int64) error {
  110.     msg := &models.SubscribeMsg{
  111.         ClientID: gMQClientID,
  112.         Topic: gMQSubscribeTopic,
  113.         NotifyUrl: gMQServerURL + PathOfNotifySaleOrderCreated,
  114.         ExpireTime: expireTime,
  115.     }
  116.     url := gMQServerURL + "/subscribe"
  117.     return fnPost(msg, url)
  118. }
  121. func fnPost(msg interface{}, url string) error {
  122.     body,_ := json.Marshal(msg)
  123.     rsp, err := http.Post(url, "application/json;charset=utf-8", bytes.NewReader(body))
  124.     if err != nil {
  125.         return err
  126.     }
  128.     defer rsp.Body.Close()
  129.     j, err := ioutil.ReadAll(rsp.Body)
  130.     if err != nil {
  131.         return err
  132.     }
  133.     ok := &models.OkMsg{}
  134.     err = json.Unmarshal(j, ok)
  135.     if err != nil {
  136.         return err
  137.     }
  139.     if !ok.OK {
  140.         return gMQReplyFalse
  141.     }
  143.     return nil
  144. }
  148. func (me *tStockService) beginPublishMQ() {
  149.     for {
  150.         select {
  151.         case msg := <- me.publishQueue :
  152.             me.publishMQ(msg)
  153.             break
  154.         }
  155.     }
  156. }
  158. func (me *tStockService) publishMQ(msg *models.TxMsg) {
  159.     url := gMQServerURL + "/publish"
  160.     for i := 0;i < gMQMaxPublishRetry;i++ {
  161.         err := fnPost(msg, url)
  162.         if err != nil {
  163.             logger.Logf("tStockService.publishMQ, failed, err=%v, msg=%v", err, msg)
  164.             time.Sleep(gMQPublishInterval)
  166.         } else {
  167.             logger.Logf("tStockService.publishMQ, done, msg=%v", msg)
  168.             return
  169.         }
  170.     }
  172.     // publish failed
  173.     logger.Logf("tStockService.publishMQ, failed max retries, msg=%v", msg)
  174. }
  177. func (me *tStockService) HandleSaleOrderCreated(it *order.SaleOrder) error {
  178.     msg := &models.TxMsg{}
  179.     msg.GlobalID = it.OrderID
  180.     msg.SubID = it.OrderID + ".outbound"
  181.     msg.SenderID = gMQClientID
  182.     msg.Topic = gMQPublishTopic
  184.     err := me.AddStock(it.ProductID, -it.Quantity)
  185.     msg.CreateTime = time.Now().UnixNano()
  187.     if err != nil {
  188.         logger.Logf("tStockService.HandleSaleOrderCreated, err=%s, order=%v", err.Error(), it)
  189.         msg.Content = "0"
  190.     } else {
  191.         msg.Content = "1"
  192.     }
  194.     if len(me.publishQueue) >= gMQMaxQueuedMsg {
  195.         logger.Logf("tStockService.HandleSaleOrderCreated, err=%s, order=%v", gMQBlocked.Error(), it)
  196.         return gMQBlocked
  198.     } else {
  199.         me.publishQueue <- msg
  200.         return err
  201.     }
  202. }
  205. var gInsufficientStockError = errors.New("insufficient stock")
  207. var gMQReplyFalse = errors.New("mq reply false")
  208. var gMQBlocked = errors.New("mq blocked")
  209. var gMQMaxPublishRetry = 10
  210. var gMQPublishInterval = 1*time.Second
  211. var gMQSubscribeTopic = "sale-order.created"
  212. var gMQPublishTopic = "sale-order.stock.outbound"
  213. var gMQClientID = "stock-service"
  214. var gMQServerURL = "http://localhost:3333"
  215. var gMQMaxQueuedMsg = 1024
  217. var MockStockService = newStockService()

NotifySaleOrderCreated.go

用于监听订单创建消息的http回调处理器

package stock

import (
    "encoding/json"
    "github.com/gin-gonic/gin"
    "io/ioutil"
    "learning/gooop/saga/mqs/logger"
    "learning/gooop/saga/mqs/models"
    "learning/gooop/saga/order"
    "net/http"
)

func NotifySaleOrderCreated(c *gin.Context) {
    body := c.Request.Body
    defer body.Close()

    j, e := ioutil.ReadAll(body)
    if e != nil {
        logger.Logf("stock.NotifySaleOrderCreated, failed ioutil.ReadAll")
        c.JSON(http.StatusBadRequest, gin.H { "ok": false, "error": e.Error()})
        return
    }

    msg := &models.TxMsg{}
    e = json.Unmarshal(j, msg)
    if e != nil {
        logger.Logf("stock.NotifySaleOrderCreated, failed json.Unmarshal msg")
        c.JSON(http.StatusBadRequest, gin.H { "ok": false, "error": e.Error()})
        return
    }

    order := &order.SaleOrder{}
    e = json.Unmarshal([]byte(msg.Content), order)
    if e != nil {
        logger.Logf("stock.NotifySaleOrderCreated, failed json.Unmarshal order")
        c.JSON(http.StatusBadRequest, gin.H { "ok": false, "error": e.Error()})
        return
    }
    logger.Logf("stock.NotifySaleOrderCreated, order=%s/%v", order.OrderID, order)

    // notify stock service
    _ = MockStockService.HandleSaleOrderCreated(order)
    c.JSON(http.StatusOK, gin.H{ "ok": true })
}

var PathOfNotifySaleOrderCreated = "/notify/sale-order.created"

(未完待续)