印度登月最關鍵一步！月船三號今晚進入環月軌道
8月5日消息，據印度官方消息，月船三號將于北京時間今晚21時30分左右開始近月制動進入環月軌道。這是該探測器能夠成功的最關鍵步驟之一，如果成功將開始圍
微服務開發中經常會使用消息隊列進行跨服務通信。在一個典型場景中，服務A執行一個業務邏輯，需要保存數據庫，然后通知服務B執行相應的業務邏輯。在這種場景下，我們需要考慮如何發送消息。
圖片
首先，我們可能會考慮將數據庫操作和消息發送放在同一個事務中，以下是偽代碼示例：
@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注解將數據庫的操作以及發送消息放到一個事務中，如果數據庫的保存或者消息發送失敗，則回滾事務。
乍一看似乎沒什么問題，但稍微推敲一下就會發現此方式有如下兩個缺陷：
首先最容易想到的是，這種消息發送方式無法保證數據的最終一致性。
這里先讓我來解釋一下基于消息隊列，生產者發送消息到消費者消費消息的過程：
圖片
好，現在回到上面發送消息的場景，假設數據庫處理成功，消息消費成功，但是MQ由于某些原因處理超時，導致ACK確認失敗，此時整個事務回滾，結果出現數據不一致問題。
這種數據不一致的問題在RPC調用的場景下也經常出現，其根本的原因在于：遠程調用，結果最終可能為成功、失敗、超時；而對于超時的情況，處理方最終的結果可能是成功，也可能是失敗，調用方是無法知曉的。
其次，使用以上方式還存在另一個問題，即消費者在處理消息時可能讀不到剛剛保存的數據，即消費者消費速度快于事務提交的速度。
舉例：
假設服務B需要通過消息中的數據ID獲取服務A數據庫保存的數據。這種情況下，數據庫操作與消息發送在同一事務中。可能出現服務B在處理消息時，服務A事務還未提交，導致服務B獲取的數據是空數據，無法執行相應業務邏輯。
盡管這種發送方式存在上述兩個問題，但在某些場景下仍然適用。例如，消費者在處理時不依賴生產者的數據，且對數據一致性要求不高，這種情況下消息發送和數據庫保存可以不在同一個事務中。
為解決事務未提交問題，我們可以確保事務提交后再發送消息。在SpringBoot項目中，有兩種常見解決方案。
基于事務同步器的方法：
@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方法，在事務成功提交后執行發送消息操作，確保在數據庫操作成功且事務穩定的情況下發送消息。
另一種方法是基于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時才進行事件的處理，來達到事務同步的目的。
通過以上方式，相較于基礎版，可以確保消息在事務提交后發送，解決了消費者讀取空數據的問題。但仍然無法保證數據的一致性，適用于對數據一致性要求不高的場景。
如果需要保證最終一致性而非強一致性，可以采用本地消息表+補償重試的方式來發送消息。
這種方式的執行原理如下：
圖片
核心代碼如下：
@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){ // 執行失敗的業務邏輯 } }雖然這種方式能夠保證消息在事務提交后發送，且能夠保證最終一致性，但仍然存在一些缺陷：
首先，需要額外的消息表，增加了系統復雜度。（針對此問題，我們又可以將該功能單獨提取出來，做成一個消息服務來統一處理，考慮篇幅問題，這里暫不展開。）
其次，通過定時任務輪詢消息表，對于處理成功但ACK超時的數據會重新發送消息，這對下游系統產生了強烈的冪等性保障要求，消費者的處理邏輯必須做好冪等控制。關于冪等的處理方案，我在[[Dailymart17：并發與冪等的實現方案]]一文中有詳細的說明，歡迎翻閱。
目前，RocketMQ是主流MQ中唯一一個支持事務消息的，如果你們項目恰好使用的是RocketMQ，可以采用事務消息來發送。
有關RocketMQ事務消息的詳細信息，可以參考我之前的文章[SpringCloud基于RocketMQ實現分布式事務，這里不再贅述。同時，由于在Dailymart項目中使用的是RocketMQ，也可以參考Dailymart的代碼實現。
本文詳細介紹了微服務開發中常用的4種消息發送方式。對于那些對數據一致性要求不高的場景，可以選擇使用進階版的消息發送方式。而對于需要保證最終一致性的情況，推薦采用事務消息和本地消息表的方式進行消息發送。
在文中涉及的相關代碼，你可以在我星球 DDD&微服務 系列專欄中找到相應的實現，歡迎加入我們的討論。
DailyMart是一個基于 DDD 和Spring Cloud Alibaba的微服務商城系統，采用SpringBoot3.x以及JDK17。旨在為開發者提供集成式的學習體驗，并將其無縫地應用于實際項目中。該專欄包含領域驅動設計（DDD）、Spring Cloud Alibaba企業級開發實踐、設計模式實際應用場景解析、分庫分表戰術及實用技巧等內容。如果你對這個系列感興趣，可在本公眾號回復關鍵詞 DDD 獲取完整文檔以及相關源碼。
本文鏈接：http://www.www897cc.com/showinfo-26-48351-0.html消息隊列，聊聊發送消息的四種姿勢~
聲明：本網頁內容旨在傳播知識，若有侵權等問題請及時與本網聯系，我們將在第一時間刪除處理。郵件：2376512515@qq.com
下一篇： 神技能！一招教你免費搞定PDF轉word
Copyright ? 2016-2023 天津谷騏科技有限公司 版權所有 sitemap.xml
違法及侵權請聯系：2376512515@qq.com 津ICP備18001702號
津公網安備 12010102000574號