缘起
最近阅读<<Go微服务实战>> (刘金亮, 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（原子性、一致性、持久性）特征。
摘自 <<Go微服务实战>> 刘金亮, 2021.1
目标
- 为实现saga模式的分布式事务, 先撸一个pub/sub事务消息队列服务
- 事务消息队列服务的功能性要求
- 消息不会丢失: 消息的持久化
- 消息的唯一性: 要求每个消息有全局ID和子事务ID
- 确保投递成功: 投递队列持久化, 投递状态持久化, 失败重试
子目标(Day 7)
- MQS已基本可用, 现在实现一个模拟的订单微服务, 并与MQ联动
- 长事务: 订单创建后, 联动库存服务, 扣减库存
- 补偿动作
- 如果扣库成功, 更新订单状态为已出库(实际系统中, 可能还涉及物流发货等复杂流程)
- 否则(库存不足), 更新订单状态为出库失败(实际系统中, 可能还涉及退款和通知客户等复杂流程)
- 流程
- 创建订单后, 向MQ发布[销售订单.创建]消息
- 订阅MQ的[销售订单.出库.成功], [销售订单.出库.失败]消息
- 接收到MQ的出库消息后, 更新订单状态
设计
- ISaleOrderService: 订单服务接口
- SaleOrder: 销售订单实体
- tSaleOrderService: 模拟订单服务, 实现ISaleOrderService接口
- NotifyStockOutbound: 接收库存服务的扣库结果消息
ISaleOrderService.go
订单服务接口
package order
// ISaleOrderService to manage sale order creation and modification
type ISaleOrderService interface {
// get order info
Get(orderID string) *SaleOrder
// create new order
Create(it *SaleOrder) error
// update order status
Update(orderID string, oldStatusFlag int32, newStatusFlag int32) (error, *SaleOrder)
}
SaleOrder.go
销售订单实体
package order
type SaleOrder struct {
OrderID string
CustomerID string
ProductID string
Quantity int
Price float64
Amount float64
CreateTime int64
StatusFlag int32
}
const StatusNotDelivered int32 = 0
const StatusStockOutboundDone int32 = 1
const StatusStockOutboundFailed int32 = 2
const StatusMQServiceFailed int32 = 3
tSaleOrderService.go
模拟订单服务, 实现ISaleOrderService接口
package order
import (
"bytes"
"encoding/json"
"errors"
"io/ioutil"
"learning/gooop/saga/mqs/logger"
"learning/gooop/saga/mqs/models"
"net/http"
"sync"
"sync/atomic"
"time"
)
type tSaleOrderService struct {
rwmutex *sync.RWMutex
orders map[string]*SaleOrder
bMQReady bool
publishQueue chan *SaleOrder
}
func newSaleOrderService() ISaleOrderService {
it := new(tSaleOrderService)
it.init()
return it
}
func (me *tSaleOrderService) init() {
me.rwmutex = new(sync.RWMutex)
me.orders = make(map[string]*SaleOrder)
me.bMQReady = false
me.publishQueue = make(chan *SaleOrder, gMQMaxQueuedMsg)
go me.beginSubscribeMQ()
go me.beginPublishMQ()
}
func (me *tSaleOrderService) beginSubscribeMQ() {
expireDuration := int64(1 * time.Hour)
subscribeDuration := 20 * time.Minute
pauseDuration := 3*time.Second
lastSubscribeTime := int64(0)
for {
now := time.Now().UnixNano()
if now - lastSubscribeTime >= int64(subscribeDuration) {
expireTime := now + expireDuration
err := fnSubscribeMQ(expireTime)
if err != nil {
me.bMQReady = false
logger.Logf("tSaleOrderService.beginSubscribeMQ, failed, err=%v", err)
} else {
lastSubscribeTime = now
me.bMQReady = true
logger.Logf("tSaleOrderService.beginSubscribeMQ, done")
}
}
time.Sleep(pauseDuration)
}
}
func fnSubscribeMQ(expireTime int64) error {
msg := &models.SubscribeMsg{
ClientID: gMQClientID,
Topic: gMQSubscribeTopic,
NotifyUrl: gMQServerURL + PathOfNotifyStockOutbound,
ExpireTime: expireTime,
}
url := gMQServerURL + "/subscribe"
return fnPost(msg, url)
}
func fnPost(msg interface{}, url string) error {
body,_ := json.Marshal(msg)
rsp, err := http.Post(url, "application/json;charset=utf-8", bytes.NewReader(body))
if err != nil {
return err
}
defer rsp.Body.Close()
j, err := ioutil.ReadAll(rsp.Body)
if err != nil {
return err
}
ok := &models.OkMsg{}
err = json.Unmarshal(j, ok)
if err != nil {
return err
}
if !ok.OK {
return gMQReplyFalse
}
return nil
}
func (me *tSaleOrderService) beginPublishMQ() {
for {
select {
case msg := <- me.publishQueue :
me.publishMQ(msg)
break
}
}
}
func (me *tSaleOrderService) Get(orderID string) *SaleOrder {
me.rwmutex.RLock()
defer me.rwmutex.RUnlock()
it,ok := me.orders[orderID]
if ok {
return it
} else {
return nil
}
}
func (me *tSaleOrderService) Create(it *SaleOrder) error {
me.rwmutex.Lock()
defer me.rwmutex.Unlock()
if len(me.publishQueue) >= gMQMaxQueuedMsg {
return gMQNotAvailableError
}
me.orders[it.OrderID] = it
me.publishQueue <- it
return nil
}
func (me *tSaleOrderService) publishMQ(it *SaleOrder) {
url := gMQServerURL + "/publish"
j,_ := json.Marshal(it)
msg := &models.TxMsg{
GlobalID: it.OrderID,
SubID: it.OrderID,
SenderID: gMQClientID,
Topic: gMQPublishTopic,
CreateTime: it.CreateTime,
Content: string(j),
}
for i := 0;i < gMQMaxPublishRetry;i++ {
err := fnPost(msg, url)
if err != nil {
logger.Logf("tSaleOrderService.publishMQ, failed, err=%v, order=%v", err, it)
time.Sleep(gMQPublishInterval)
} else {
logger.Logf("tSaleOrderService.publishMQ, done, order=%v", it)
return
}
}
// publish failed
logger.Logf("tSaleOrderService.publishMQ, failed max retries, order=%v", it)
_,_ = me.Update(it.OrderID, StatusNotDelivered, StatusMQServiceFailed)
}
func (me *tSaleOrderService) Update(orderID string, oldStatusFlag int32, newStatusFlag int32) (error, *SaleOrder) {
me.rwmutex.RLock()
defer me.rwmutex.RUnlock()
it, ok := me.orders[orderID]
if !ok {
return gNotFoundError, nil
}
if !atomic.CompareAndSwapInt32(&it.StatusFlag, oldStatusFlag, newStatusFlag) {
return gStatusChangedError, it
}
it.StatusFlag = newStatusFlag
return nil, it
}
var gMQReplyFalse = errors.New("mq reply false")
var gMQNotAvailableError = errors.New("mq not ready")
var gNotFoundError = errors.New("order not found")
var gStatusChangedError = errors.New("status changed")
var gMQMaxPublishRetry = 3
var gMQPublishInterval = 1*time.Second
var gMQSubscribeTopic = "sale-order.stock.outbound"
var gMQPublishTopic = "sale-order.created"
var gMQClientID = "sale-order-service"
var gMQServerURL = "http://localhost:333"
var gMQMaxQueuedMsg = 1024
var SaleOrderService = newSaleOrderService()
NotifyStockOutbound.go
接收库存服务的扣库结果消息
package order
import (
"encoding/json"
"github.com/gin-gonic/gin"
"io/ioutil"
"learning/gooop/saga/mqs/logger"
"learning/gooop/saga/mqs/models"
"net/http"
)
func NotifyStockOutbound(c *gin.Context) {
body := c.Request.Body
defer body.Close()
j, e := ioutil.ReadAll(body)
if e != nil {
logger.Logf("order.NotifyStockOutbound, 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("order.NotifyStockOutbound, failed json.Unmarshal")
c.JSON(http.StatusBadRequest, gin.H { "ok": false, "error": e.Error()})
return
}
orderID := msg.GlobalID
succeeded := msg.Content == "1"
logger.Logf("order.NotifyStockOutbound, orderID=%v, succeeded=%s", orderID, succeeded)
var newStatusFlag int32
if succeeded {
newStatusFlag = StatusStockOutboundDone
} else {
newStatusFlag = StatusStockOutboundFailed
}
err, order := SaleOrderService.Update(orderID, StatusNotDelivered, newStatusFlag)
if err != nil {
logger.Logf("order.NotifyStockOutbound, failed SaleOrderService.Update, err=%v, order=%v", err, order)
}
c.JSON(http.StatusOK, gin.H{ "ok": true })
}
var PathOfNotifyStockOutbound = "/notify/sale-order.stock.outbound"
(未完待续)
有疑问加站长微信联系（非本文作者）
