背景 在一個微服務架構的項目中,一個業務操作可能涉及到多個服務,這些服務往往是獨立部署,構成一個個獨立的系統。這種分散式的系統架構往往面臨著分散式事務的問題。為了保證系統數據的一致性,我們需要確保這些服務中的操作要麼全部成功,要麼全部失敗。通過使用RocketMQ實現分散式事務,我們可以協調這些服務 ...
背景
在一個微服務架構的項目中,一個業務操作可能涉及到多個服務,這些服務往往是獨立部署,構成一個個獨立的系統。這種分散式的系統架構往往面臨著分散式事務的問題。為了保證系統數據的一致性,我們需要確保這些服務中的操作要麼全部成功,要麼全部失敗。通過使用RocketMQ實現分散式事務,我們可以協調這些服務的操作,保證數據的一致性。
功能原理
RocketMQ的分散式事務消息功能,在普通消息基礎上,支持二階段的提交。將二階段提交和本地事務綁定,實現全局提交結果的一致性。
整個事務消息的詳細交互流程如下圖所示:
1、生產者將消息發送至RocketMQ服務端。
2、RocketMQ服務端將消息持久化成功之後,向生產者返回Ack確認消息已經發送成功,此時消息被標記為"暫不能投遞",這種狀態下的消息即為半事務消息。
3、生產者開始執行本地事務邏輯。
4、生產者根據本地事務執行結果向服務端提交二次確認結果(Commit或是Rollback),服務端收到確認結果後處理邏輯如下:
-
二次確認結果為Commit:服務端將半事務消息標記為可投遞,並投遞給消費者。
-
二次確認結果為Rollback:服務端將回滾事務,不會將半事務消息投遞給消費者。
5、在斷網或者是生產者應用重啟的特殊情況下,若服務端未收到生產者提交的二次確認結果,或服務端收到的二次確認結果為Unknown未知狀態,經過固定時間後,服務端將對消息生產者集群中任一生產者實例發起消息回查。
6、生產者收到消息回查後,需要檢查對應消息的本地事務執行的最終結果。
7、生產者根據檢查到的本地事務的最終狀態再次提交二次確認,服務端仍按照步驟4對半事務消息進行處理。
註意問題
消息類型
事務消息僅支持在MessageType為Transaction的主題使用,即事務消息只能發送至類型為事務消息的主題中。
消息消費
RocketMQ事務消息保證生產者本地事務和下游消息發送事務的一致性,但不保證消息消費結果和上游事務的一致性。因此需要下游業務自行保證消息正確處理,建議消費端做好消費重試。
中間狀態
RocketMQ事務消息一致性為最終一致性,即在消息提交到下游消費端處理完成之前,下游和上游事務之間的狀態會不一致。因此,事務消息僅適合能接受非同步執行的場景。
事務超時
RocketMQ事務消息的生命周期存在超時機制,即半事務消息被生產者發送服務端後,如果在指定時間內服務端無法確認提交或者回滾狀態,則消息預設會被回滾。
示例代碼
以下為RocketMQ 4.x版本事務消息示例代碼,
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import java.util.concurrent.*;
public class RocketMqTransactionDemo {
public static void main(String[] args) throws Exception {
// 創建事務消息生產者
TransactionMQProducer producer = new TransactionMQProducer("transaction_producer");
producer.setNamesrvAddr("127.0.0.1:9876");
// 設置事務監聽器
TransactionListener transactionListener = new MyTransactionListener();
producer.setTransactionListener(transactionListener);
// 設置事務回查的線程池,可以不必設置,如果不設置也會預設生成一個
ExecutorService executorService = new ThreadPoolExecutor(2, 5, 100, TimeUnit.SECONDS, new ArrayBlockingQueue <Runnable> (2000), new ThreadFactory() {
@Override
public Thread newThread(Runnable r) {
Thread thread = new Thread(r);
thread.setName("client-transaction-msg-check-thread");
return thread;
}
});
producer.setExecutorService(executorService);
// 啟動生產者
producer.start();
// 發送事務消息
Message message = new Message("transaction_topic", "test_tag", "test_key", "Hello RocketMQ".getBytes());
producer.sendMessageInTransaction(message, null);
// 關閉生產者
producer.shutdown();
}
}
/**
* 事務監聽器
*/
class MyTransactionListener implements TransactionListener {
@Override
public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
// 執行本地事務操作
System.out.println("執行本地事務操作,消息內容:" + new String(msg.getBody()));
return LocalTransactionState.COMMIT_MESSAGE; // 提交事務,允許消費者消費該消息
// return LocalTransactionState.ROLLBACK_MESSAGE;// 回滾事務,消息將被丟棄不允許消費。
// return LocalTransactionState.UNKNOW;// 暫時無法判斷狀態,等待固定時間以後Broker端根據回查規則向生產者進行消息回查。
}
@Override
public LocalTransactionState checkLocalTransaction(MessageExt msg) {
// 檢查本地事務狀態
System.out.println("檢查本地事務狀態,消息內容:" + new String(msg.getBody()));
return LocalTransactionState.COMMIT_MESSAGE;
}
}
代碼解釋:
1、事務消息的生產者使用TransactionMQProducer
創建。
2、MyTransactionListener
作為事務監聽器,實現了介面TransactionListener
,該介面有兩個方法,分別是:
-
executeLocalTransaction
:
半事務消息發送成功後,執行本地事務的方法,具體執行完本地事務後,可以在該方法中返回以下三種狀態:
LocalTransactionState.COMMIT_MESSAGE: 提交事務,允許消費者消費該消息。
LocalTransactionState.ROLLBACK_MESSAGE: 回滾事務,消息將被丟棄不允許消費。
LocalTransactionState.UNKNOW: 暫時無法判斷狀態,等待固定時間以後RocketMQ服務端根據回查規則向生產者進行消息回查。 -
checkLocalTransaction
:
二次確認消息沒有收到,RocketMQ服務端回查生產者端事務結果的方法。回查規則:本地事務執行完成後,若RocketMQ服務端收到的本地事務返回狀態為LocalTransactionState.UNKNOW,或生產者應用退出導致本地事務未提交任何狀態。則RocketMQ服務端會向消息生產者發起事務回查,第一次回查後仍未獲取到事務狀態,則之後每隔一段時間會再次回查。
本文來自博客園,作者:Y00,轉載請註明原文鏈接:https://www.cnblogs.com/ayic/p/18067431
聊聊技術,聊聊人生。歡迎關註我的公眾號!^_^
![](https://images.cnblogs.com/cnblogs_com/ayic/2228020/o_221015170948_qr_code_search.png)