golang rocketmq包管数据一致性（以电商订单为例）

数据一致性的原因

1.所有情况
a):当地订单信息创建失败，库存扣减成功（一致）
b):当地订单信息新建失败，但是库存扣减成功（不一致）
c):当地订单信息新建成功，但是库存扣减失败（不一致）
d):当地订单信息新建失败，库存扣减失败（一致）
当地数据库直接通过数据库事务就能完成，但是分布式体系跨微服务就很复杂
2.大概会出现的标题

a):库存扣减成功，当地执行失败-调用库存的归还接口（磁盘满了导致服务宕机、网络出标题调用失败-重试到成功-网络标题比力复杂：抖动、拥塞导致过了一段时间之后才会被接收到？-会不会导致重复归还-幂等性标题）
b):非当地执行失败，而是当地代码非常（挂掉），不知道当地执行情况

a):扣减库存发送出去，当网络壅闭，重试三次后哀求回滚
b):扣减库存发送出去，在库存服务返回之前当地事务宕机
TCC解决库存扣减的标题

confirm或者cancel的时候都大概会宕机，但是下一次启动的时候可以读取日志继续没有完成的逻辑，实现比力复杂，java有seta库，golang有go-seta但不支持grpc

基于可靠消息终极一致性方案

基于可靠消息本质上解决了一个标题：可靠消息（消耗者应该确保这个消息能被精确消耗，即库存服务代码要可靠而且肯定要执行，代码有bug或者宕机都能解决）
积分服务没有标题，但是库存服务大概存在库存不足，得答应失败（TCC比力合理一些，而且TCC并发没有可靠消息并发高）

改进执行库存后执行当地事务，关注库存扣减成功后出现的那种非常，扣减失败直接不执行

如何通过基于可靠消息实现库存的一致性？
1.订单信息准备一个归还库存的half消息
2.订单服务通过grpc调用库存服务
3.假如失败直接rollback不发送，什么事变都没有发生；假如成功执行当地事务
4.当地事务执行成功，通过rollback不发送1准备的消息；当地事务假如失败，通过commit发送1准备的消息通过调用reback逻辑归还库存
5.通过commit发送1准备的消息通过调用reback逻辑归还库存
到此正常的逻辑已经结束，第六七步是宕机或者步调堕落的回查
6.假如非常步调挂掉，就回查当地逻辑
7.假如没有查询到当地信息就通过commit发送1准备的消息通过调用reback逻辑归还库存

潜在标题：下单但不付出-设置超时归还订单
改进：通过库存服务监听延时消息确保超时归还

发送端逻辑
构造OrderListener

1. type OrderListener struct{}
3. // When send transactional prepare(half) message succeed, this method will be invoked to execute local transaction.
4. func (o *OrderListener) ExecuteLocalTransaction(addr *primitive.Message) primitive.LocalTransactionState {
5.         fmt.Println("开始执行本地逻辑")
6.         time.Sleep(time.Second * 3)
7.         fmt.Println("执行本地逻辑失败")
8.         //本地执行逻辑无缘无故失败 代码异常 宕机
9.         return primitive.UnknowState
10. }
12. // When no response to prepare(half) message. broker will send check message to check the transaction status, and this
13. // method will be invoked to get local transaction status.
14. func (o *OrderListener) CheckLocalTransaction(msg *primitive.MessageExt) primitive.LocalTransactionState {
15.         fmt.Println("rocketmq的消息回查")
16.         time.Sleep(time.Second * 15)
17.         return primitive.CommitMessageState
18. }

复制代码

创建订单API

1. func (*OrderServer) CreateOrder(ctx context.Context, req *proto.OrderRequest) (*proto.OrderInfoResponse, error) {
2.         orderListener := OrderListener{}
3.         p, err := rocketmq.NewTransactionProducer(
4.                 &orderListener,
5.                 producer.WithNameServer([]string{"IP:PORT"}),
6.         )
7.         if err != nil {
8.                 // panic("生成producer失败")
9.                 zap.S().Errorf("生成producer失败:%s", err.Error())
10.                 return nil, err
11.         }
12.         if err = p.Start(); err != nil {
13.                 // panic("启动producer失败")
14.                 zap.S().Errorf("启动producer失败:%s", err.Error())
15.                 return nil, err
16.         }
18.         order := model.OrderInfo{
19.                 OrderSn:      GenerateOrderSn(req.UserId),
20.                 Address:      req.Address,
21.                 SignerName:   req.Name,
22.                 SingerMobile: req.Mobile,
23.                 Post:         req.Post,
24.                 User:         req.UserId,
25.         }
26.         //应该在消息中具体指名一个具体的商品的扣减情况
27.         jsonString, _ := json.Marshal(order)
29.         _, err = p.SendMessageInTransaction(context.Background(),
30.                 primitive.NewMessage("order_reback", jsonString))
31.         if err != nil {
32.                 fmt.Printf("发送失败:%s\n", err)
33.                 return nil, status.Error(codes.Internal, "发送消息失败")
34.         }
35.         // 4.本地事务失败，commit
36.         if orderListener.Code != codes.OK {
37.                 return nil, status.Error(orderListener.Code, orderListener.Detail)
38.         }
40.         return &proto.OrderInfoResponse{Id: orderListener.ID, OrderSn: order.OrderSn, Total: orderListener.OrderAmount}, nil
41. }

复制代码

修改OrderListener通过指针的情势回传值

1. type OrderListener struct {
2.         Code        codes.Code
3.         Detail      string
4.         ID          int32
5.         OrderAmount float32
6. }

复制代码

ExecuteLocalTransaction执行当地事务逻辑

1. // When send transactional prepare(half) message succeed, this method will be invoked to execute local transaction.
2. func (o *OrderListener) ExecuteLocalTransaction(msg *primitive.Message) primitive.LocalTransactionState {
4.         var orderInfo model.OrderInfo
5.         _ = json.Unmarshal(msg.Body, &orderInfo)
7.         var goodsIds []int32
8.         var shopCarts []model.ShoppingCart
9.         goodsNumsMap := make(map[int32]int32)
10.         if result := global.DB.Where(&model.ShoppingCart{User: orderInfo.User, Checked: true}).Find(&shopCarts); result.RowsAffected == 0 {
11.                 o.Code = codes.InvalidArgument
12.                 o.Detail = "没有选中结算的商品"
13.                 //没扣减回滚，没必要发送
14.                 return primitive.RollbackMessageState
15.         }
16.         for _, shopCart := range shopCarts {
17.                 goodsIds = append(goodsIds, shopCart.Goods)
18.                 goodsNumsMap[shopCart.Goods] = shopCart.Nums
19.         }
20.         //跨服务调用
21.         //商品微服务
22.         goods, err := global.GoodsSrvClient.BatchGetGoods(context.Background(), &proto.BatchGoodsIdInfo{
23.                 Id: goodsIds,
24.         })
25.         if err != nil {
26.                 o.Code = codes.Internal
27.                 o.Detail = "批量查询商品信息失败"
28.                 //没扣减回滚，没必要发送
29.                 return primitive.RollbackMessageState
30.         }
31.         var orderAmount float32
32.         var orderGoods []*model.OrderGoods
33.         var goodsInvInfo []*proto.GoodsInvInfo
34.         for _, good := range goods.Data {
35.                 orderAmount += good.ShopPrice * float32(goodsNumsMap[good.Id])
36.                 orderGoods = append(orderGoods, &model.OrderGoods{
37.                         Goods:      good.Id,
38.                         GoodsName:  good.Name,
39.                         GoodsImage: good.GoodsFrontImage,
40.                         GoodsPrice: good.ShopPrice,
41.                         Nums:       goodsNumsMap[good.Id],
42.                 })
43.                 goodsInvInfo = append(goodsInvInfo, &proto.GoodsInvInfo{
44.                         GoodsId: good.Id,
45.                         Num:     goodsNumsMap[good.Id],
46.                 })
47.         }
49.         //跨服务调用库存微服务进行库存扣减
50.         if _, err = global.InventorySrvClient.Sell(context.Background(), &proto.SellInfo{OrderSn: orderInfo.OrderSn, GoodsInfo: goodsInvInfo}); err != nil {
51.                 o.Code = codes.ResourceExhausted
52.                 o.Detail = "扣减库存失败"
53.                 return primitive.RollbackMessageState
54.         }
55.         //生成订单表
56.         //20241020xxxx
57.         tx := global.DB.Begin()
58.         orderInfo.OrderMount = orderAmount
59.         if result := tx.Save(&orderInfo); result.RowsAffected == 0 {
60.                 tx.Rollback()
61.                 o.Code = codes.ResourceExhausted
62.                 o.Detail = "创建订单失败"
63.                 return primitive.CommitMessageState
64.         }
66.         for _, orderGood := range orderGoods {
67.                 orderGood.Order = orderInfo.ID
68.         }
70.         o.OrderAmount = orderAmount
71.         o.ID = orderInfo.ID
72.         //批量插入orderGoods
73.         if result := tx.CreateInBatches(orderGoods, 100); result.RowsAffected == 0 {
74.                 tx.Rollback()
75.                 o.Code = codes.Internal
76.                 o.Detail = "批量插入商品失败"
77.                 return primitive.CommitMessageState
78.         }
80.         if result := tx.Where(&model.ShoppingCart{User: orderInfo.User, Checked: true}).Delete(&model.ShoppingCart{}); result.RowsAffected == 0 {
81.                 tx.Rollback()
82.                 o.Code = codes.Internal
83.                 o.Detail = "删除购物车记录失败"
84.                 return primitive.CommitMessageState
85.         }
86.         //发送延时消息
87.         p, err := rocketmq.NewProducer(producer.WithNameServer([]string{"IP:PORT"}))
88.         if err != nil {
89.                 panic("生成producer失败")
90.         }
91.         if err = p.Start(); err != nil {
92.                 panic("启动producer失败")
93.         }
94.         msgs := primitive.NewMessage("order_timeout", msg.Body)
95.         msgs.WithDelayTimeLevel(16) //延时30mins
96.         _, err = p.SendSync(context.Background(), msgs)
97.         if err != nil {
98.                 zap.S().Errorf("发送延时消息失败:%v\n", err.Error())
99.                 tx.Rollback()
100.                 o.Code = codes.Internal
101.                 o.Detail = "发送延时消息失败"
102.                 return primitive.CommitMessageState
103.         }
104.         if err = p.Shutdown(); err != nil {
105.                 panic("关闭producer失败")
106.         }
107.         //提交事务
108.         tx.Commit()
109.         o.Code = codes.OK
110.         //本地执行逻辑无缘无故失败 代码异常 宕机
111.         return primitive.RollbackMessageState
112. }

复制代码

消息回查CheckLocalTransaction

1. // When no response to prepare(half) message. broker will send check message to check the transaction status, and this
2. // method will be invoked to get local transaction status.
3. func (o *OrderListener) CheckLocalTransaction(msg *primitive.MessageExt) primitive.LocalTransactionState {
4.         var orderInfo model.OrderInfo
5.         _ = json.Unmarshal(msg.Body, &orderInfo)
6.         //怎么检查之前的逻辑是否完成
7.         if result := global.DB.Where(model.OrderInfo{OrderSn: orderInfo.OrderSn}).First(&orderInfo); result.RowsAffected == 0 {
8.                 //这里并不能说明内存扣减了
9.                 return primitive.CommitMessageState
10.         }
12.         return primitive.RollbackMessageState
13. }

复制代码

消耗端逻辑
归还表

1. type DeliveryHistory struct {
2.         Goods   int32  `gorm:"type:int;index"`
3.         Num     int32  `gorm:"type:int"`
4.         OrderSn string `gorm:"type:varchar(200)"`
5.         Status  string `gorm:"type:varchar(200)"` //1.代表已扣减.2.代表已归还，3.失败
6. }

复制代码

表设计比力贫苦，十件商品创建十条信息，1.数据量大2.查询贫苦
改进表为StockSellDetail

1. type StockSellDetail struct {
2.         OrderSn string          `gorm:"type:varchar(200);index:idx_order_sn,unique"`
3.         Status  int             `gorm:"type:varchar(200)"` //1.表示已扣减，2.表示已归还
4.         Detail  GoodsDetailList `gorm:"type:varchar(200)"` //list类型，包括id和num
5. }

复制代码

GoodsDetailList定义

1. type GoodsDetail struct {
2.         Goods int32
3.         Num   int32
4. }
5. type GoodsDetailList []GoodsDetail
7. func (g *GoodsDetailList) Scan(value interface{}) error {
8.         return json.Unmarshal(value.([]byte), &g)
9. }
10. func (g GoodsDetailList) Value() (driver.Value, error) {
11.         return json.Marshal(g)
12. }

复制代码

通过gorm新建数据表
归还函数

1. func AutoReback(ctx context.Context, msgs ...*primitive.MessageExt) (consumer.ConsumeResult, error) {
2.         type OrderInfo struct {
3.                 OrderSn string
4.         }
5.         for i := range msgs {
6.                 var orderInfo OrderInfo
7.                 err := json.Unmarshal(msgs[i].Body, &orderInfo)
8.                 if err != nil {
9.                         zap.S().Errorf("解析json失败:%v\n", msgs[i].Body)
10.                         //代码问题 下一次重新发consumer.ConsumeRetryLater
11.                         //数据问题 忽略
12.                         return consumer.ConsumeSuccess, nil
13.                 }
14.                 //去将inv的库存加回去，将selldetail的status设置为2，要在事务中进行
15.                 tx := global.DB.Begin()
16.                 var sellDetail model.StockSellDetail
17.                 if result := tx.Where(&model.StockSellDetail{OrderSn: orderInfo.OrderSn, Status: 1}).First(&sellDetail); result.RowsAffected == 0 {
18.                         return consumer.ConsumeSuccess, nil
19.                 }
20.                 //逐个归还
21.                 for _, orderGood := range sellDetail.Detail {
22.                         //先查后改并发容易出问题，直接stocks=stocks+2
23.                         if result := tx.Where(&model.Inventory{Goods: orderGood.Goods}).Update("stocks", gorm.Expr("sotcks+?", orderGood.Num)); result.RowsAffected == 0 {
24.                                 tx.Rollback()
25.                                 //检查代码
26.                                 return consumer.ConsumeRetryLater, nil
27.                         }
28.                 }
29.                 sellDetail.Status = 2
30.                 if result := tx.Where(&model.StockSellDetail{OrderSn: orderInfo.OrderSn}).Update("status", 2); result.RowsAffected == 0 {
31.                         tx.Rollback()
32.                         return consumer.ConsumeRetryLater, nil
33.                 }
34.                 tx.Commit()
35.         }
36.         return consumer.ConsumeSuccess, nil
37. }

复制代码

库存服务订阅topic

1. //监听库存归还topic
2.         c, err := rocketmq.NewPushConsumer(
3.                 consumer.WithNameServer([]string{"IP:PORT"}),
4.                 consumer.WithGroupName("shop"),
5.         )
6.         if err != nil {
7.                 panic("连接失败")
8.         }
9.         if err = c.Subscribe("order_reback", consumer.MessageSelector{}, handler.AutoReback); err != nil {
10.                 fmt.Println("读取消息失败")
11.         }
12.         _ = c.Start()
13.         //不能让主goroutine退出
14.         time.Sleep(time.Hour)
15.         _ = c.Shutdown()

复制代码

超时归还订阅topic

1. //订单超时归还topic
2.         c, err := rocketmq.NewPushConsumer(
3.                 consumer.WithNameServer([]string{"IP:PORT"}),
4.                 consumer.WithGroupName("order"),
5.         )
6.         if err != nil {
7.                 panic("连接失败")
8.         }
9.         if err = c.Subscribe("order_timeout", consumer.MessageSelector{}, handler.OrderTimeout); err != nil {
10.                 fmt.Println("读取消息失败")
11.         }
12.         _ = c.Start()
13.         //不能让主goroutine退出
14.         time.Sleep(time.Hour)
15.         _ = c.Shutdown()

复制代码

超时归还处置惩罚函数

1. func OrderTimeout(ctx context.Context, msgs ...*primitive.MessageExt) (consumer.ConsumeResult, error) {
2.         for i := range msgs {
3.                 var orderInfo model.OrderInfo
4.                 _ = json.Unmarshal(msgs[i].Body, &orderInfo)
5.                 fmt.Printf("获取到消息订单超时消息:%v\n", time.Now())
6.                 //查询订单的支付状态，未支付归还
7.                 var order model.OrderInfo
8.                 if result := global.DB.Model(model.OrderInfo{}).Where(model.OrderInfo{OrderSn: orderInfo.OrderSn}).First(&order); result.RowsAffected == 0 {
9.                         return consumer.ConsumeSuccess, nil
10.                 }
11.                 if order.Status != "TRADE_SUCCESS" {
12.                         //归还库存发送一个消息到order_reback
13.                         tx := global.DB.Begin()
14.                         //修改订单的状态已支付
15.                         order.Status = "TRADE_CLOSE"
16.                         tx.Save(&order)
17.                         p, err := rocketmq.NewProducer(producer.WithNameServer([]string{"IP:PORT"}))
18.                         if err != nil {
19.                                 panic("生成producer失败")
20.                         }
21.                         if err = p.Start(); err != nil {
22.                                 panic("启动producer失败")
23.                         }
24.                         res, err := p.SendSync(context.Background(), primitive.NewMessage("order_reback", msgs[i].Body))
25.                         if err != nil {
26.                                 tx.Rollback()
27.                                 fmt.Println("发送失败:%s\n", err)
28.                                 return consumer.ConsumeRetryLater, nil
29.                         } else {
30.                                 fmt.Println("发送成功:%s\n", res.String())
31.                         }
32.                         tx.Commit()
33.                         if err = p.Shutdown(); err != nil {
34.                                 panic("关闭producer失败")
35.                         }
37.                 }
38.         }
39.         return consumer.ConsumeSuccess, nil
40. }

复制代码

shutdown导致消耗偏移量offset更新失败

client.ClientID()默认返回历程的pid，而clientMap通过调用client.ClientID()为key创建或加载，等候shutdown时会移除，假如一个历程中多次调用且存在shutdown大概会导致消耗偏移量offset更新失败，进而导致rocketmq堕落。
解决方法
(1)生成clientID不一致
(2)不随便shutdown
注意
不要在一个历程中利用多个producer或者不要随意调用shutdown影响其他producer

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。