消息隊列，聊聊發送消息的四種姿勢！
微服務開發中經常會使用消息隊列進行跨服務通信。在一個典型場景中，服務A執行一個業務邏輯，需要保存數據庫，然后通知服務B執行相應的業務邏輯。在這種場景下，我們需要考慮如何發送消息。
1. 基礎版
首先，我們可能會考慮將數據庫操作和消息發送放在同一個事務中，以下是偽代碼示例：
@Transactional
public void saveWithMessage(BusinessDO businessDO){
String id = IdUtils.nextId();
businessDO.setId(id);
xxxRepository.save(businessDO);
BusinessMessage businessMessage = new BusinessMessage();
businessMessage.setKey(id);
SendResult send = rocketMQTemplate.syncSend("test-topic", sendMessage);
}在這段代碼里通過@Transactional注解將數據庫的操作以及發送消息放到一個事務中，如果數據庫的保存或者消息發送失敗，則回滾事務。
乍一看似乎沒什么問題，但稍微推敲一下就會發現此方式有如下兩個缺陷：
1.1 數據不一致
首先最容易想到的是，這種消息發送方式無法保證數據的最終一致性。
這里先讓我來解釋一下基于消息隊列，生產者發送消息到消費者消費消息的過程：
- 生產者發送消息
- MQ收到消息并將數據持久化，在存儲中新增一條記錄
- 返回ACK給生產者
- MQ 推送消息給對應的消費者，等待消費者返回ACK
- 如果消費者在指定時間內成功返回ACK，那么MQ則認為消費成功，執行第6步刪除消息，如果MQ在指定時間內沒有收到ACK，則認為消息消費失敗，會重新推送消息，重復執行第4、5、6步操作。
- 刪除消息。
圖片
好，現在回到上面發送消息的場景，假設數據庫處理成功，消息消費成功，但是MQ由于某些原因處理超時，導致ACK確認失敗，此時整個事務回滾，結果出現數據不一致問題。
這種數據不一致的問題在RPC調用的場景下也經常出現，其根本的原因在于：遠程調用，結果最終可能為成功、失敗、超時；而對于超時的情況，處理方最終的結果可能是成功，也可能是失敗，調用方是無法知曉的。
1.2 事務未提交
其次，使用以上方式還存在另一個問題，即消費者在處理消息時可能讀不到剛剛保存的數據，即消費者消費速度快于事務提交的速度。
舉例：
假設服務B需要通過消息中的數據ID獲取服務A數據庫保存的數據。這種情況下，數據庫操作與消息發送在同一事務中。可能出現服務B在處理消息時，服務A事務還未提交，導致服務B獲取的數據是空數據，無法執行相應業務邏輯。
1.3 適用場景
盡管這種發送方式存在上述兩個問題，但在某些場景下仍然適用。例如，消費者在處理時不依賴生產者的數據，且對數據一致性要求不高，這種情況下消息發送和數據庫保存可以不在同一個事務中。
2. 進階版
為解決事務未提交問題，我們可以確保事務提交后再發送消息。在SpringBoot項目中，有兩種常見解決方案。
2.1 事務同步器
基于事務同步器的方法：
@Transactional
public void saveWithMessage(BusinessDO businessDO){
xxxRepository.save(businessDO);
TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization() {
@Override
public void afterCommit() {
BusinessMessage businessMessage = new BusinessMessage();
businessMessage.setXXX();
rocketMQTemplate.syncSend("test-topic", sendMessage);
}
});
}TransactionSynchronizationManager.registerSynchronization 是 Spring 框架中用于注冊事務同步的方法。通過這個方法，你可以在事務提交、回滾或完成時執行一些額外的邏輯。
在上述代碼中，使用了afterCommit方法，在事務成功提交后執行發送消息操作，確保在數據庫操作成功且事務穩定的情況下發送消息。
2.2 消息監聽器
另一種方法是基于ApplicationEventPublisher，在保存數據庫操作后發布一個事件，并通過@TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)注解監聽事件。這樣可以確保消息在數據庫事務提交之后再發送。
@Transactional
public void saveWithMessage(BusinessDO businessDO){
xxxRepository.save(businessDO);
eventPublisher.publishEvent(new UserCreatedEvent(registerUser));
}
@TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
public void handleUserRegisteredEvent(UserCreatedEvent userCreatedEvent) {
rocketMQTemplate.syncSend("test-topic", sendMessage);
}這里需要說明一下：在默認情況下Spring的事件監聽機制并不是異步的（上次群友弄錯了），而是同步的將代碼進行解耦，@TransactionalEventListener也是通過同步的方式，但是加入了回調的方式來解決，這樣就能夠控制事務進行Commited、Rollback時才進行事件的處理，來達到事務同步的目的。
2.3 適用范圍
通過以上方式，相較于基礎版，可以確保消息在事務提交后發送，解決了消費者讀取空數據的問題。但仍然無法保證數據的一致性，適用于對數據一致性要求不高的場景。
3. 本地消息表+補償重試
如果需要保證最終一致性而非強一致性，可以采用本地消息表+補償重試的方式來發送消息。
這種方式的執行原理如下：
- 在執行業務操作的同時，在本地消息表中插入一條狀態為待發送的記錄，業務數據的記錄與消息記錄必須在同一個事務中完成，這是此方案的核心原則。由于消息表與業務表在同一個庫中，事務可以通過數據庫來保證。
- 事務提交后發送消息，如果消息發送成功，則將消息狀態標記為發送成功或刪除消息
- 在生產者服務中會創建一個定時任務，定時從消息表中檢索待發送的消息重新發送。
- 對于消費者消費失敗，則依賴MQ本身的重試機制來完成，保證數據的最終一致性。
圖片
核心代碼如下：
@Transactional
public void saveWithMessage(BusinessDO businessDO){
TransactionMessage transactionMessage = new TransactionMessage();
transactionMessage.setStaus(MessageStatus.WAITING_SEND);
transactionMessage.setMessageKey(businessDO.getId());
...
xxxRepository.save(businessDO);
messageRepository.save(TransactionMessage);
TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization() {
@Override
public void afterCommit() {
messageService.sendMessage(transactionMessage,businessDO);
}
});
}
public void sendMessage(TransactionMessage transactionMessage,BusinessDO businessDO){
BusinessMessage businessMessage = new BusinessMessage();
businessMessage.setXXX();
try{
rocketMQTemplate.syncSend("test-topic", sendMessage);
transactionMessage.setStatus(MessageStatus.SUCCESS);
messageRepository.update(transactionMessage);
}catch (Exception e){
// 執行失敗的業務邏輯
}
}3.1 問題
雖然這種方式能夠保證消息在事務提交后發送，且能夠保證最終一致性，但仍然存在一些缺陷：
首先，需要額外的消息表，增加了系統復雜度。（針對此問題，我們又可以將該功能單獨提取出來，做成一個消息服務來統一處理，考慮篇幅問題，這里暫不展開。）
其次，通過定時任務輪詢消息表，對于處理成功但ACK超時的數據會重新發送消息，這對下游系統產生了強烈的冪等性保障要求，消費者的處理邏輯必須做好冪等控制。關于冪等的處理方案，我在[[Dailymart17：并發與冪等的實現方案]]一文中有詳細的說明，歡迎翻閱。
4. 基于事務消息發送
目前，RocketMQ是主流MQ中唯一一個支持事務消息的，如果你們項目恰好使用的是RocketMQ，可以采用事務消息來發送。
有關RocketMQ事務消息的詳細信息，可以參考我之前的文章[SpringCloud基于RocketMQ實現分布式事務，這里不再贅述。同時，由于在Dailymart項目中使用的是RocketMQ，也可以參考Dailymart的代碼實現。
小結
本文詳細介紹了微服務開發中常用的4種消息發送方式。對于那些對數據一致性要求不高的場景，可以選擇使用進階版的消息發送方式。而對于需要保證最終一致性的情況，推薦采用事務消息和本地消息表的方式進行消息發送。
