分散式事務(4)---RocketMQ實現分散式事務專案
阿新 • • 發佈：2019-07-17
RocketMQ實現分散式事務
有關RocketMQ實現分散式事務前面寫了一篇部落格
1、RocketMQ實現分散式事務原理
下面就這個專案做個整體簡單介紹，並在文字最下方附上專案Github地址。
一、專案概述
1、技術架構
專案總體技術選型
SpringCloud(Finchley.RELEASE) + SpringBoot2.0.4 + Maven3.5.4 + RocketMQ4.3 +MySQL + lombok(外掛)有關SpringCloud主要用到以下四個組建
Eureka Server +config-server(配置中心)+ Eureka Client + Feign(服務間呼叫)
配置中心是用MySQL儲存資料。
2、專案整體結構
config-service # 配置中心
eureka # 註冊中心
service-order #訂單微服務
service-produce #商品微服務各服務的啟動順序就安裝上面的順序啟動。
大致流程
啟動後，配置中心、訂單微服務、商品微服務都會將資訊註冊到註冊中心。
如果訪問：localhost:7001(註冊中心地址）,以上服務都出現說明啟動成功。
3、分散式服務流程
使用者在訂單微服務下單後，會去回撥商品微服務去減庫存。這個過程需要事務的一致性。
4、測試流程
頁面輸入：
http://localhost:9001/api/v1/order/save?userId=1&productId=1&total=4
訂單微服務執行情況（訂單服務事務執行成功）
商品微服務執行情況（商品服務事務執行成功）
當然你也可以通過修改引數來模擬分散式事務出現的各種情況。
二、MQ中生產者核心程式碼
這裡展示下，生產者傳送訊息核心程式碼。
@Slf4j @Component public class TransactionProducer { /** * 需要自定義事務監聽器 用於 事務的二次確認 和 事務回查 */ private TransactionListener transactionListener ; /** * 這裡的生產者和之前的不一樣 */ private TransactionMQProducer producer = null; /** * 官方建議自定義執行緒 給執行緒取自定義名稱 發現問題更好排查 */ private 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; } }); public TransactionProducer(@Autowired Jms jms, @Autowired ProduceOrderService produceOrderService) { transactionListener = new TransactionListenerImpl(produceOrderService); // 初始化 事務生產者 producer = new TransactionMQProducer(jms.getOrderTopic()); // 新增伺服器地址 producer.setNamesrvAddr(jms.getNameServer()); // 新增事務監聽器 producer.setTransactionListener(transactionListener); // 新增自定義執行緒池 producer.setExecutorService(executorService); start(); } public TransactionMQProducer getProducer() { return this.producer; } /** * 物件在使用之前必須要呼叫一次，只能初始化一次 */ public void start() { try { this.producer.start(); } catch (MQClientException e) { e.printStackTrace(); } } /** * 一般在應用上下文，使用上下文監聽器，進行關閉 */ public void shutdown() { this.producer.shutdown(); } } /** * @author xub * @Description: 自定義事務監聽器 * @date 2019/7/15 下午12:20 */ @Slf4j class TransactionListenerImpl implements TransactionListener { @Autowired private ProduceOrderService produceOrderService ; public TransactionListenerImpl( ProduceOrderService produceOrderService) { this.produceOrderService = produceOrderService; } @Override public LocalTransactionState executeLocalTransaction(Message msg, Object arg) { log.info("=========本地事務開始執行============="); String message = new String(msg.getBody()); JSONObject jsonObject = JSONObject.parseObject(message); Integer productId = jsonObject.getInteger("productId"); Integer total = jsonObject.getInteger("total"); int userId = Integer.parseInt(arg.toString()); //模擬執行本地事務begin======= /** * 本地事務執行會有三種可能 * 1、commit 成功 * 2、Rollback 失敗 * 3、網路等原因服務宕機收不到返回結果 */ log.info("本地事務執行引數,使用者id={},商品ID={},銷售庫存={}",userId,productId,total); int result = produceOrderService.save(userId, productId, total); //模擬執行本地事務end======== //TODO 實際開發下面不需要我們手動返回，而是根據本地事務執行結果自動返回 //1、二次確認訊息，然後消費者可以消費 if (result == 0) { return LocalTransactionState.COMMIT_MESSAGE; } //2、回滾訊息，Broker端會刪除半訊息 if (result == 1) { return LocalTransactionState.ROLLBACK_MESSAGE; } //3、Broker端會進行回查訊息 if (result == 2) { return LocalTransactionState.UNKNOW; } return LocalTransactionState.COMMIT_MESSAGE; } /** * 只有上面介面返回 LocalTransactionState.UNKNOW 才會呼叫查介面被呼叫 * * @param msg 訊息 * @return */ @Override public LocalTransactionState checkLocalTransaction(MessageExt msg) { log.info("==========回查介面========="); String key = msg.getKeys(); //TODO 1、必須根據key先去檢查本地事務訊息是否完成。 /** * 因為有種情況就是：上面本地事務執行成功了，但是return LocalTransactionState.COMMIT_MESSAG的時候 * 服務掛了，那麼最終 Brock還未收到訊息的二次確定，還是個半訊息 ，所以當重新啟動的時候還是回撥這個回撥介面。 * 如果不先查詢上面本地事務的執行情況 直接在執行本地事務，那麼就相當於成功執行了兩次本地事務了。 */ // TODO 2、這裡返回要麼commit 要麼rollback。沒有必要在返回 UNKNOW return LocalTransactionState.COMMIT_MESSAGE; } }
三、MQ消費端核心程式碼
這裡展示下，消費端消費訊息核心程式碼。消費端和普通消費一樣。
@Slf4j
@Component
public class OrderConsumer {
private DefaultMQPushConsumer consumer;
private String consumerGroup = "produce_consumer_group";
public OrderConsumer(@Autowired Jms jms,@Autowired ProduceService produceService) throws MQClientException {
//設定消費組
consumer = new DefaultMQPushConsumer(consumerGroup);
// 新增伺服器地址
consumer.setNamesrvAddr(jms.getNameServer());
// 新增訂閱號
consumer.subscribe(jms.getOrderTopic(), "*");
// 監聽訊息
consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
MessageExt msg = msgs.get(0);
String message = new String(msgs.get(0).getBody());
JSONObject jsonObject = JSONObject.parseObject(message);
Integer productId = jsonObject.getInteger("productId");
Integer total = jsonObject.getInteger("total");
String key = msg.getKeys();
log.info("消費端消費訊息，商品ID={},銷售數量={}",productId,total);
try {
produceService.updateStore(productId, total, key);
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
} catch (Exception e) {
log.info("消費失敗，進行重試，重試到一定次數 那麼將該條記錄記錄到資料庫中，進行如果處理");
e.printStackTrace();
return ConsumeConcurrentlyStatus.RECONSUME_LATER;
}
});
consumer.start();
System.out.println("consumer start ...");
}
}至於完整的專案地址見GitHub。
GitHubd地址 https://github.com/yudiandemingzi/spring-cloud-rocketmq-transaction.git
晚安!
只要自己變優秀了，其他的事情才會跟著好起來（上將8）