SpringBoot分布式事務RocketMQ

原文鏈接:https://blog.csdn.net/Timeguys/article/details/107949660

一、使用：

一、引入依賴：

1.  
   <dependency>
2.  
   <groupId>org.apache.rocketmq</groupId>
3.  
   <artifactId>rocketmq-spring-boot-starter</artifactId>
4.  
   <version>2.0.3</version>
5.  
   </dependency>

二、舉例：生產者創建訂單---->生產者發送消息----->MQ服務接受消息----->消費者監聽消息並減庫存

【生產者】：

application.yml

1.  
   rocketmq:
2.  
   name-server: 192.168.85.128:9876 # rocketMQ地址
3.  
   producer:
4.  
   group: producer-group-test # 生產者的組名需要和消費者監聽consumerGroup一致

業務代碼：

1.  
   @Service
2.  
   public class OrderServiceImpl extends ServiceImpl<OrderMapper, TbOrder> implements OrderService {
3.  
    
4.  
   @Resource
5.  
   private RocketMQTemplate rocketMQTemplate;
6.  
    
7.  
   @Override
8.  
   public void create() {
9.  
   //創建訂單--> 發送消息 --> 消息發送成功后調用本地事務提交 -->
10.  
    TbOrder order = new TbOrder();
11.  
    order.setCount( 10);
12.  
    order.setMoney(BigDecimal.valueOf( 10));
13.  
    order.setProductId( 1L);
14.  
    order.setStatus( 1);
15.  
    order.setUserId( 1L);
16.  
    sendMsg(order);
17.  
    }
18.  
    @Override
19.  
    public void sendMsg(TbOrder order){
20.  
    /**
21.  
    * String txProducerGroup, 生產者分組
22.  
    * String destination, topic
23.  
    * Message<?> message, 消息
24.  
    * Object arg 消息參數
25.  
    */
26.  
    Message<String> build = MessageBuilder.withPayload(JSONObject.toJSONString(order)).build();
27.  
    rocketMQTemplate.sendMessageInTransaction( "tx-producer-group","txmsg-topic",build , null);
28.  
    }
29.  
    }

創建  ProducerTxmsgListener 並實現 RocketMQLocalTransactionListener：

1.  
   @Component
2.  
   // txProducerGroup 的值和發送事務消息指定的 txProducerGroup 相同
3.  
   @RocketMQTransactionListener(txProducerGroup = "txmsg-producer-group")
4.  
   public class ProducerTxmsgListener implements RocketMQLocalTransactionListener {
5.  
   @Resource
6.  
   private OrderService orderService;
7.  
    
8.  
   /**
9.  
   * @Description: 執行本地事務提交
10.  
    */
11.  
    @Override
12.  
    @Transactional
13.  
    public RocketMQLocalTransactionState executeLocalTransaction(Message message, Object o) {
14.  
    try {
15.  
    TbOrder tbOrder = JSONObject.parseObject(message.getPayload().toString(), TbOrder.class);
16.  
    System.out.println(tbOrder);
17.  
    orderService.save(tbOrder);
18.  
    return RocketMQLocalTransactionState.COMMIT; //變更消息狀態為：可消費
19.  
    } catch (Exception e){
20.  
    return RocketMQLocalTransactionState.ROLLBACK; //本地事務執行異常，將消息遺棄
21.  
    }
22.  
    }
23.  
     
24.  
    /**
25.  
    * @Description: 檢查本地事務是否執行成功
26.  
    */
27.  
    @Override
28.  
    public RocketMQLocalTransactionState checkLocalTransaction(Message message) {
29.  
     
30.  
    TbOrder tbOrder = JSONObject.parseObject(message.getPayload().toString(), TbOrder.class);
31.  
    TbOrder order = orderService.getById(tbOrder.getId());
32.  
    // 不為 null 則表示執行成功
33.  
    if (order != null){
34.  
    return RocketMQLocalTransactionState.COMMIT; //變更消息狀態為：可消費
35.  
    }
36.  
    // 執行本地事務發生問題或還沒執行完成, UNKNOWN 表示會繼續回查
37.  
    return RocketMQLocalTransactionState.UNKNOWN;
38.  
    }
39.  
    }

【消費者】：

application.yml

1.  
   rocketmq:
2.  
   name-server: 127.0.0.1:9876 # rocketMQ地址
3.  
   producer:
4.  
   group: producer-test-group # 生產者的組名需要和消費者監聽consumerGroup一致

創建MyListener 並實現 RocketMQListener 接口：

1.  
   // topic 對應生產者發消息是的topic
2.  
   @RocketMQMessageListener(topic = "test-topic" , consumerGroup = "consumer-group")
3.  
   public class MyListener implements RocketMQListener<String> {
4.  
    
5.  
   @Override
6.  
   public void onMessage(String message) {
7.  
   //執行 減庫存業務 如果發生異常，則消息會隔段時間再次消費
8.  
   System.out.println(message);
9.  
   }
10.  
    }

 

原理圖：