缘起

最近阅读<<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"

(未完待续)

本文来自:Segmentfault

感谢作者:ioly

查看原文:手撸golang GO与微服务 Saga模式之7

• 编辑
• 预览