微服務(wù)開(kāi)發(fā)中經(jīng)常會(huì)使用消息隊(duì)列進(jìn)行跨服務(wù)通信。在一個(gè)典型場(chǎng)景中，服務(wù)A執(zhí)行一個(gè)業(yè)務(wù)邏輯，需要保存數(shù)據(jù)庫(kù)，然后通知服務(wù)B執(zhí)行相應(yīng)的業(yè)務(wù)邏輯。在這種場(chǎng)景下，我們需要考慮如何發(fā)送消息。

圖片

1. 基礎(chǔ)版

首先，我們可能會(huì)考慮將數(shù)據(jù)庫(kù)操作和消息發(fā)送放在同一個(gè)事務(wù)中，以下是偽代碼示例：

@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);
}

在這段代碼里通過(guò)@Transactional注解將數(shù)據(jù)庫(kù)的操作以及發(fā)送消息放到一個(gè)事務(wù)中，如果數(shù)據(jù)庫(kù)的保存或者消息發(fā)送失敗，則回滾事務(wù)。

乍一看似乎沒(méi)什么問(wèn)題，但稍微推敲一下就會(huì)發(fā)現(xiàn)此方式有如下兩個(gè)缺陷：

1.1 數(shù)據(jù)不一致

首先最容易想到的是，這種消息發(fā)送方式無(wú)法保證數(shù)據(jù)的最終一致性。

這里先讓我來(lái)解釋一下基于消息隊(duì)列，生產(chǎn)者發(fā)送消息到消費(fèi)者消費(fèi)消息的過(guò)程：

1. 生產(chǎn)者發(fā)送消息
2. MQ收到消息并將數(shù)據(jù)持久化，在存儲(chǔ)中新增一條記錄
3. 返回ACK給生產(chǎn)者
4. MQ 推送消息給對(duì)應(yīng)的消費(fèi)者，等待消費(fèi)者返回ACK
5. 如果消費(fèi)者在指定時(shí)間內(nèi)成功返回ACK，那么MQ則認(rèn)為消費(fèi)成功，執(zhí)行第6步刪除消息，如果MQ在指定時(shí)間內(nèi)沒(méi)有收到ACK，則認(rèn)為消息消費(fèi)失敗，會(huì)重新推送消息，重復(fù)執(zhí)行第4、5、6步操作。
6. 刪除消息。

圖片

好，現(xiàn)在回到上面發(fā)送消息的場(chǎng)景，假設(shè)數(shù)據(jù)庫(kù)處理成功，消息消費(fèi)成功，但是MQ由于某些原因處理超時(shí)，導(dǎo)致ACK確認(rèn)失敗，此時(shí)整個(gè)事務(wù)回滾，結(jié)果出現(xiàn)數(shù)據(jù)不一致問(wèn)題。

這種數(shù)據(jù)不一致的問(wèn)題在RPC調(diào)用的場(chǎng)景下也經(jīng)常出現(xiàn)，其根本的原因在于：遠(yuǎn)程調(diào)用，結(jié)果最終可能為成功、失敗、超時(shí)；而對(duì)于超時(shí)的情況，處理方最終的結(jié)果可能是成功，也可能是失敗，調(diào)用方是無(wú)法知曉的。

1.2 事務(wù)未提交

其次，使用以上方式還存在另一個(gè)問(wèn)題，即消費(fèi)者在處理消息時(shí)可能讀不到剛剛保存的數(shù)據(jù)，即消費(fèi)者消費(fèi)速度快于事務(wù)提交的速度。

舉例：

假設(shè)服務(wù)B需要通過(guò)消息中的數(shù)據(jù)ID獲取服務(wù)A數(shù)據(jù)庫(kù)保存的數(shù)據(jù)。這種情況下，數(shù)據(jù)庫(kù)操作與消息發(fā)送在同一事務(wù)中。可能出現(xiàn)服務(wù)B在處理消息時(shí)，服務(wù)A事務(wù)還未提交，導(dǎo)致服務(wù)B獲取的數(shù)據(jù)是空數(shù)據(jù)，無(wú)法執(zhí)行相應(yīng)業(yè)務(wù)邏輯。

1.3 適用場(chǎng)景

盡管這種發(fā)送方式存在上述兩個(gè)問(wèn)題，但在某些場(chǎng)景下仍然適用。例如，消費(fèi)者在處理時(shí)不依賴生產(chǎn)者的數(shù)據(jù)，且對(duì)數(shù)據(jù)一致性要求不高，這種情況下消息發(fā)送和數(shù)據(jù)庫(kù)保存可以不在同一個(gè)事務(wù)中。

2. 進(jìn)階版

為解決事務(wù)未提交問(wèn)題，我們可以確保事務(wù)提交后再發(fā)送消息。在SpringBoot項(xiàng)目中，有兩種常見(jiàn)解決方案。

2.1 事務(wù)同步器

基于事務(wù)同步器的方法：

@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 框架中用于注冊(cè)事務(wù)同步的方法。通過(guò)這個(gè)方法，你可以在事務(wù)提交、回滾或完成時(shí)執(zhí)行一些額外的邏輯。

在上述代碼中，使用了afterCommit方法，在事務(wù)成功提交后執(zhí)行發(fā)送消息操作，確保在數(shù)據(jù)庫(kù)操作成功且事務(wù)穩(wěn)定的情況下發(fā)送消息。

2.2 消息監(jiān)聽(tīng)器

另一種方法是基于ApplicationEventPublisher，在保存數(shù)據(jù)庫(kù)操作后發(fā)布一個(gè)事件，并通過(guò)@TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)注解監(jiān)聽(tīng)事件。這樣可以確保消息在數(shù)據(jù)庫(kù)事務(wù)提交之后再發(fā)送。

@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);  
}

這里需要說(shuō)明一下：在默認(rèn)情況下Spring的事件監(jiān)聽(tīng)機(jī)制并不是異步的（上次群友弄錯(cuò)了），而是同步的將代碼進(jìn)行解耦，@TransactionalEventListener也是通過(guò)同步的方式，但是加入了回調(diào)的方式來(lái)解決，這樣就能夠控制事務(wù)進(jìn)行Commited、Rollback時(shí)才進(jìn)行事件的處理，來(lái)達(dá)到事務(wù)同步的目的。

2.3 適用范圍

通過(guò)以上方式，相較于基礎(chǔ)版，可以確保消息在事務(wù)提交后發(fā)送，解決了消費(fèi)者讀取空數(shù)據(jù)的問(wèn)題。但仍然無(wú)法保證數(shù)據(jù)的一致性，適用于對(duì)數(shù)據(jù)一致性要求不高的場(chǎng)景。

3. 本地消息表+補(bǔ)償重試

如果需要保證最終一致性而非強(qiáng)一致性，可以采用本地消息表+補(bǔ)償重試的方式來(lái)發(fā)送消息。

這種方式的執(zhí)行原理如下：

1. 在執(zhí)行業(yè)務(wù)操作的同時(shí)，在本地消息表中插入一條狀態(tài)為待發(fā)送的記錄，業(yè)務(wù)數(shù)據(jù)的記錄與消息記錄必須在同一個(gè)事務(wù)中完成，這是此方案的核心原則。由于消息表與業(yè)務(wù)表在同一個(gè)庫(kù)中，事務(wù)可以通過(guò)數(shù)據(jù)庫(kù)來(lái)保證。
2. 事務(wù)提交后發(fā)送消息，如果消息發(fā)送成功，則將消息狀態(tài)標(biāo)記為發(fā)送成功或刪除消息
3. 在生產(chǎn)者服務(wù)中會(huì)創(chuàng)建一個(gè)定時(shí)任務(wù)，定時(shí)從消息表中檢索待發(fā)送的消息重新發(fā)送。
4. 對(duì)于消費(fèi)者消費(fèi)失敗，則依賴MQ本身的重試機(jī)制來(lái)完成，保證數(shù)據(jù)的最終一致性。

圖片

核心代碼如下：

@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){  
        // 執(zhí)行失敗的業(yè)務(wù)邏輯  
    }  
  
}

3.1 問(wèn)題

雖然這種方式能夠保證消息在事務(wù)提交后發(fā)送，且能夠保證最終一致性，但仍然存在一些缺陷：

首先，需要額外的消息表，增加了系統(tǒng)復(fù)雜度。（針對(duì)此問(wèn)題，我們又可以將該功能單獨(dú)提取出來(lái)，做成一個(gè)消息服務(wù)來(lái)統(tǒng)一處理，考慮篇幅問(wèn)題，這里暫不展開(kāi)。）

其次，通過(guò)定時(shí)任務(wù)輪詢消息表，對(duì)于處理成功但ACK超時(shí)的數(shù)據(jù)會(huì)重新發(fā)送消息，這對(duì)下游系統(tǒng)產(chǎn)生了強(qiáng)烈的冪等性保障要求，消費(fèi)者的處理邏輯必須做好冪等控制。關(guān)于冪等的處理方案，我在[[Dailymart17：并發(fā)與冪等的實(shí)現(xiàn)方案]]一文中有詳細(xì)的說(shuō)明，歡迎翻閱。

4. 基于事務(wù)消息發(fā)送

目前，RocketMQ是主流MQ中唯一一個(gè)支持事務(wù)消息的，如果你們項(xiàng)目恰好使用的是RocketMQ，可以采用事務(wù)消息來(lái)發(fā)送。

有關(guān)RocketMQ事務(wù)消息的詳細(xì)信息，可以參考我之前的文章[SpringCloud基于RocketMQ實(shí)現(xiàn)分布式事務(wù)，這里不再贅述。同時(shí)，由于在Dailymart項(xiàng)目中使用的是RocketMQ，也可以參考Dailymart的代碼實(shí)現(xiàn)。

小結(jié)

本文詳細(xì)介紹了微服務(wù)開(kāi)發(fā)中常用的4種消息發(fā)送方式。對(duì)于那些對(duì)數(shù)據(jù)一致性要求不高的場(chǎng)景，可以選擇使用進(jìn)階版的消息發(fā)送方式。而對(duì)于需要保證最終一致性的情況，推薦采用事務(wù)消息和本地消息表的方式進(jìn)行消息發(fā)送。

在文中涉及的相關(guān)代碼，你可以在我星球 DDD&微服務(wù) 系列專欄中找到相應(yīng)的實(shí)現(xiàn)，歡迎加入我們的討論。

DailyMart是一個(gè)基于 DDD 和Spring Cloud Alibaba的微服務(wù)商城系統(tǒng)，采用SpringBoot3.x以及JDK17。旨在為開(kāi)發(fā)者提供集成式的學(xué)習(xí)體驗(yàn)，并將其無(wú)縫地應(yīng)用于實(shí)際項(xiàng)目中。該專欄包含領(lǐng)域驅(qū)動(dòng)設(shè)計(jì)（DDD）、Spring Cloud Alibaba企業(yè)級(jí)開(kāi)發(fā)實(shí)踐、設(shè)計(jì)模式實(shí)際應(yīng)用場(chǎng)景解析、分庫(kù)分表戰(zhàn)術(shù)及實(shí)用技巧等內(nèi)容。如果你對(duì)這個(gè)系列感興趣，可在本公眾號(hào)回復(fù)關(guān)鍵詞 DDD 獲取完整文檔以及相關(guān)源碼。