saga事务问题

请问一下这里已经扣减库存成功了，为什么还会有一直重复调用扣减库存的情况呢？

感觉上是有多个线程一直在调用，saga事务状态一直是submitted状态

如图：这里已经执行成功扣减库存的sell方法，并进行了421和422的库存扣减，创建了stockselldetail信息

这里仍然会有其他线程的调用记录：

YLDALONG 2023-06-27

源自：订单服务重构、wire进行ioc控制 1-10 调试新建订单接口(下)

收起

3回答

bobby 回答被采纳获得+3积分 2023-06-29 00:48:18

库存扣减成功需要一个返回告诉这个调用逻辑成功了否则dtm会反复重试

收起回答

YLDALONG 提问者 #1

扣减库存的事务commit后不是这样返回吗？怎样返回告诉这个调用逻辑成功呢？

2023-06-29 09:43:51
bobby 回复提问者 YLDALONG #2

只要这里返回的err不是nil就代表成功，你在return的时候打印一下看看是否返回的err是nil，同时记得这个接口不能超时，dtm对每个请求是有超时的，所以你不要打断点或者执行时间过长

2023-06-29 22:33:55
YLDALONG 提问者回复 bobby #3

排查出是grpc调用设置的超时时间太短了，谢谢老师的提醒！

2023-06-30 10:20:59

提问者 YLDALONG 2023-06-27 17:42:20

调用扣减库存的逻辑：

r := gin.Default()
r.GET("start", func(c *gin.Context) {
orderSn := shortuuid.New()
req := &proto.SellInfo{
GoodsInfo: []*proto.GoodsInvInfo{
{
GoodsId: 421,
Num: 2,
},
{
GoodsId: 422,
Num: 2,
},
},
OrderSn: orderSn,
}
dmtServer := "192.168.56.1:36790"
qsBusi := "discovery://192.168.159.142:8500/mxshop-inventory-srv"
//qsBusi := "192.168.56.1:8019"
fmt.Println(orderSn)
saga := dtmgrpc.NewSagaGrpc(dmtServer, orderSn).
Add(qsBusi+"/Inventory/Sell", qsBusi+"/Inventory/Reback", req)
err := saga.Submit()
if err != nil {
c.JSON(500, gin.H{"message": err.Error()})
}
c.JSON(200, gin.H{"message": "ok"})
})

r.Run(":8089")

收起回答

提问者 YLDALONG #1

扣减库存的逻辑:
func (is *inventoryService) Sell(ctx context.Context, ordersn string, details []do.GoodsDetail) error {
log.Infof("订单%s扣减库存", ordersn)
//解决了空悬挂的问题
//先查询刚才插入的记录是否存在，如果存在则说明已经cancel就不能执行了

rs := redsync.New(is.pool)
//实际上批量扣减库存的时候，我们经常会先按照商品的id排序，然后从小大小逐个扣减库存，这样可以减少锁的竞争
//如果无序的话那么就有可能订单a 扣减 1,3,4 订单B 扣减 3,2,1
var detail = do.GoodsDetailList(details)
sort.Sort(detail)

txn := is.data.Begin()
defer func() {
if err := recover(); err != nil {
_ = txn.Rollback()
log.Error("事务进行中出现异常，回滚")
return
}
}()

sellDetail := do.StockSellDetailDO{
OrderSn: ordersn,
Status: 1,
Detail: detail,
}

for _, goodsInfo := range detail {
mutex := rs.NewMutex(inventoryLockPrefix + ordersn)
if err := mutex.Lock(); err != nil {
log.Errorf("订单%s获取锁失败", ordersn)
}
defer mutex.Unlock()

inv, err := is.data.Inventorys().Get(ctx, uint64(goodsInfo.Goods))
if err != nil {
log.Errorf("订单%s获取库存失败", ordersn)
return err
}

//判断库存是否充足
if inv.Stocks < goodsInfo.Num {
txn.Rollback() //回滚
log.Errorf("商品%d库存%d不足, 现有库存: %d", goodsInfo.Goods, goodsInfo.Num, inv.Stocks)
return errors.WithCode(code.ErrInvNotEnough, "库存不足")
}
inv.Stocks -= goodsInfo.Num

err = is.data.Inventorys().Reduce(ctx, txn, uint64(goodsInfo.Goods), int(goodsInfo.Num))
if err != nil {
txn.Rollback() //回滚
log.Errorf("订单%s扣减库存失败", ordersn)
return err
}

}

err := is.data.Inventorys().CreateStockSellDetail(ctx, txn, &sellDetail)
if err != nil {
txn.Rollback() //回滚
log.Errorf("订单%s创建扣减库存记录失败", ordersn)
return err
}

txn.Commit()
return nil
}

2023-06-27 17:44:10

提问者 YLDALONG 2023-06-27 17:41:48

扣减库存的逻辑:

func (is *inventoryService) Sell(ctx context.Context, ordersn string, details []do.GoodsDetail) error {
log.Infof("订单%s扣减库存", ordersn)
//解决了空悬挂的问题
//先查询刚才插入的记录是否存在，如果存在则说明已经cancel就不能执行了

rs := redsync.New(is.pool)
//实际上批量扣减库存的时候，我们经常会先按照商品的id排序，然后从小大小逐个扣减库存，这样可以减少锁的竞争
//如果无序的话那么就有可能订单a 扣减 1,3,4 订单B 扣减 3,2,1
var detail = do.GoodsDetailList(details)
sort.Sort(detail)

txn := is.data.Begin()
defer func() {
if err := recover(); err != nil {
_ = txn.Rollback()
log.Error("事务进行中出现异常，回滚")
return
}
}()

sellDetail := do.StockSellDetailDO{
OrderSn: ordersn,
Status: 1,
Detail: detail,
}

for _, goodsInfo := range detail {
mutex := rs.NewMutex(inventoryLockPrefix + ordersn)
if err := mutex.Lock(); err != nil {
log.Errorf("订单%s获取锁失败", ordersn)
}
defer mutex.Unlock()

inv, err := is.data.Inventorys().Get(ctx, uint64(goodsInfo.Goods))
if err != nil {
log.Errorf("订单%s获取库存失败", ordersn)
return err
}

//判断库存是否充足
if inv.Stocks < goodsInfo.Num {
txn.Rollback() //回滚
log.Errorf("商品%d库存%d不足, 现有库存: %d", goodsInfo.Goods, goodsInfo.Num, inv.Stocks)
return errors.WithCode(code.ErrInvNotEnough, "库存不足")
}
inv.Stocks -= goodsInfo.Num

err = is.data.Inventorys().Reduce(ctx, txn, uint64(goodsInfo.Goods), int(goodsInfo.Num))
if err != nil {
txn.Rollback() //回滚
log.Errorf("订单%s扣减库存失败", ordersn)
return err
}

}

err := is.data.Inventorys().CreateStockSellDetail(ctx, txn, &sellDetail)
if err != nil {
txn.Rollback() //回滚
log.Errorf("订单%s创建扣减库存记录失败", ordersn)
return err
}

txn.Commit()
return nil
}