事務消息

RocketMQ提供了事務消息，通過事務消息就能達到分布式事務的最終一致。

事務消息交互流程:

兩個概念:

半事務消息：暫不能投遞的消息，發送方已經成功地將消息發送到了RocketMQ服務端，但是服務端未收到生產者對該消息的二次確認，此時該消息被標記成“暫不能投遞”狀態，處于該種狀態下的消息即半事務消息。

消息回查：由于網絡閃斷、生產者應用重啟等原因，導致某條事務消息的二次確認丟失，RocketMQ服務端通過掃描發現某條消息長期處于“半事務消息”時，需要主動向消息生產者詢問該消息的最終狀態（Commit 或是 Rollback），該詢問過程即消息回查。

事務消息發送步驟：

1. 發送方將半事務消息發送至RocketMQ服務端。

2. RocketMQ服務端將消息持久化之后，向發送方返回Ack確認消息已經發送成功，此時消息為半事務消息。

3. 發送方開始執行本地事務邏輯。

4. 發送方根據本地事務執行結果向服務端提交二次確認（Commit 或是 Rollback），服務端收到Commit 狀態則將半事務消息標記為可投遞，訂閱方最終將收到該消息；服務端收到 Rollback 狀態則刪除半事務消息，訂閱方將不會接受該消息。

事務消息回查步驟：

1. 在斷網或者是應用重啟的特殊情況下，上述步驟4提交的二次確認最終未到達服務端，經過固定時間后服務端將對該消息發起消息回查。

2. 發送方收到消息回查后，需要檢查對應消息的本地事務執行的最終結果。

3. 發送方根據檢查得到的本地事務的最終狀態再次提交二次確認，服務端仍按照步驟4對半事務消息進行操作。

package com.leon.domain;import lombok.Data;import javax.persistence.Entity; import javax.persistence.Id; import java.util.Date;//消息事物狀態記錄 @Entity(name = "shop_txlog") @Data public class TxLog {@Idprivate String txId;private Date date; } package com.leon.service.impl;import com.leon.dao.OrderDao; import com.leon.dao.TxLogDao; import com.leon.domain.Order; import com.leon.domain.TxLog; import com.leon.service.OrderService; import org.apache.rocketmq.spring.core.RocketMQTemplate; import org.aspectj.weaver.ast.Or; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.messaging.support.MessageBuilder; import org.springframework.stereotype.Service; import org.springframework.transaction.annotation.Transactional;import java.util.Date; import java.util.UUID;@Service public class OrderServiceImpl4 {@Autowiredprivate OrderDao orderDao;@Autowiredprivate TxLogDao txLogDao;@Autowiredprivate RocketMQTemplate rocketMQTemplate;public void createOrderBefore(Order order) {String txId = UUID.randomUUID().toString();//發送半事務消息rocketMQTemplate.sendMessageInTransaction("tx_producer_group","tx_topic",MessageBuilder.withPayload(order).setHeader("txId", txId).build(),order);}@Transactionalpublic void createOrder(String txId, Order order) {//保存訂單orderDao.save(order);TxLog txLog = new TxLog();txLog.setTxId(txId);txLog.setDate(new Date());//記錄事物日志txLogDao.save(txLog);}} package com.leon.service.impl;import com.leon.dao.OrderDao; import com.leon.dao.TxLogDao; import com.leon.domain.Order; import com.leon.domain.TxLog; import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener; import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener; import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState; import org.apache.rocketmq.spring.core.RocketMQTemplate; import org.aspectj.weaver.ast.Or; import org.omg.CORBA.ORB; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.messaging.Message; import org.springframework.messaging.support.MessageBuilder; import org.springframework.stereotype.Service;@Service @RocketMQTransactionListener(txProducerGroup = "tx_producer_group") public class OrderServiceImpl4Listener implements RocketMQLocalTransactionListener {@Autowiredprivate OrderServiceImpl4 orderServiceImpl4;@Autowiredprivate TxLogDao txLogDao;//執行本地事物@Overridepublic RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object arg) {String txId = (String) msg.getHeaders().get("txId");try {//本地事物Order order = (Order) arg;orderServiceImpl4.createOrder(txId,order);return RocketMQLocalTransactionState.COMMIT;} catch (Exception e) {return RocketMQLocalTransactionState.ROLLBACK;}}//消息回查@Overridepublic RocketMQLocalTransactionState checkLocalTransaction(Message msg) {String txId = (String) msg.getHeaders().get("txId");TxLog txLog = txLogDao.findById(txId).get();if (txLog != null){//本地事物(訂單)成功了return RocketMQLocalTransactionState.COMMIT;}else {return RocketMQLocalTransactionState.ROLLBACK;}} }

?

總結

以上是生活随笔為你收集整理的消息类型-事务消息的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。