噓!非同步事件這樣用真的好麽?

来源:https://www.cnblogs.com/coderjava/archive/2020/06/29/13210405.html

為了方便大家理解我把之前方案的圖片複製過來了,如下: 上圖的方案存在一個問題,就是我們今天文章要聊的內容。 這個問題就是當 MQ Consumer 收到消息後,就直接發佈 Event 了,如果是同步的,沒有問題。如果某個 EventListener 中處理失敗了,那麼這條消息將不會 ACK。 如果是 ...


為了方便大家理解我把之前方案的圖片複製過來了,如下:

img

上圖的方案存在一個問題,就是我們今天文章要聊的內容。

這個問題就是當 MQ Consumer 收到消息後,就直接發佈 Event 了,如果是同步的,沒有問題。如果某個 EventListener 中處理失敗了,那麼這條消息將不會 ACK。

如果是非同步發佈 Event 的場景,發佈完消息馬上就 ACK 了。就算某個 EventListener 中處理失敗了,MQ 也感知不到,不會進行消息的重新投遞,這就是存在的問題。

img

解決方案

方案一

既然消息已經 ACK 了,那就不利用 MQ 的重試功能了,使用方自己重試是不是也可以呢?

可肯定是可以的,內部處理是否成功肯定是可以知道的,如果處理失敗了可以預設重試,或者有一定策略的重試。實在不行還可以落庫,保存記錄。

這樣的問題在於太煩了呀,每個使用的地方都要去做這件事情,而且對於未來接手你代碼的程式小哥哥來說,這很有可能讓小哥哥頭髮慢慢脫落啊。。。。

脫落不要緊,關鍵他還不知道要做這個處理,說不定哪天就背鍋了,慘兮兮。。。。

方案二

要保證消息和業務處理的一致性,就不能立馬進行 ACK 操作。而是要等業務處理完成後再決定是否要 ACK。

如果有處理失敗的就不應該 ACK,這樣就能復用 MQ 的重試機制了。

分析下來,這就是一個典型的非同步轉同步的場景。像 Dubbo 中也有這個場景,所以我們可以借鑒 Dubbo 中的實現思路。

創建一個 DefaultFuture 用於同步等待獲取任務執行結果。然後在 MQ 消費的地方使用 DefaultFuture。

 @Service
 @RocketMQMessageListener(topic = "${rocketmq.topic.data_change}", consumerGroup = "${rocketmq.group.data_change_consumer}")
 public class DataChangeConsume implements RocketMQListener<DataChangeRequest> {
     @Autowired
     private ApplicationContext applicationContext;
     @Autowired
     private CustomApplicationContextAware customApplicationContextAware;
     @Override
     public void onMessage(DataChangeRequest dataChangeRequest) {
         log.info("received message {} , Thread {}", dataChangeRequest, Thread.currentThread().getName());
         DataChangeEvent event = new DataChangeEvent(this);
         event.setChangeType(dataChangeRequest.getChangeType());
         event.setTable(dataChangeRequest.getTable());
         event.setMessageId(dataChangeRequest.getMessageId());
         DefaultFuture defaultFuture = DefaultFuture.newFuture(dataChangeRequest, customApplicationContextAware.getTaskCount(), 6000 * 10);
         applicationContext.publishEvent(event);
         Boolean result = defaultFuture.get();
         log.info("MessageId {} 處理結果 {}", dataChangeRequest.getMessageId(), result);
         if (!result) {
             throw new RuntimeException("處理失敗,不進行消息ACK,等待下次重試");
         }
     }
 }

 

newFuture() 會傳入事件參數,超時時間,任務數量幾個參數。任務數量是用於判斷所有 EventListener 是否全部執行完成。

defaultFuture.get(); 這不就會阻塞,等待所有任務執行完成才會返回結果,如果所有業務都處理成功了,那麼會返回 true,流程結束,消息自動 ACK。

如果返回了 false 證明有處理失敗的或者超時的,就不需要 ACK 了,拋出異常等待重試。

 public Boolean get() {
     if (isDone()) {
         return true;
     }
     long start = System.currentTimeMillis();
     lock.lock();
     try {
         while (!isDone()) {
             done.await(timeout, TimeUnit.MILLISECONDS);
             // 有失敗的任務反饋
             if (!isSuccessDone()) {
                 return false;
             }
             // 全部執行成功
             if (isDone()) {
                 return true;
             }
             // 超時
             if (System.currentTimeMillis() - start > timeout) {
                 return false;
             }
         }
     } catch (InterruptedException e) {
         throw new RuntimeException(e);
     } finally {
         lock.unlock();
     }
     return true;
 }

 

isDone() 會判斷反饋結果了的任務數量跟總數量是否一致,如果一直就說明全部執行完成了。

 public boolean isDone() {
     return feedbackResultCount.get() == taskCount;
 }

 

那麼任務執行完了怎麼反饋呢? 不可能讓每個使用的方法去關心,所以我們定義了一個切麵來做這件事情。

 @Aspect
 @Component
 public class EventListenerAspect {
     @Around(value = "@annotation(eventListener)")
     public Object aroundAdvice(ProceedingJoinPoint joinpoint, EventListener eventListener) throws Throwable {
       DataChangeEvent event = null;
       boolean executeResult = true;
        try {
          event = (DataChangeEvent)joinpoint.getArgs()[0];
          Object result = joinpoint.proceed();
          return result;
       } catch (Exception e) {
          executeResult = false;
           throw e;
       } finally {
          DefaultFuture.received(event.getMessageId(), executeResult);
       }
     }
 }

 

通過 DefaultFuture.received() 反饋執行結果。

 
public static void received(String id, boolean result) {
     DefaultFuture future = FUTURES.get(id);
     if (future != null) {
         // 累加失敗任務數量
         if (!result) {
             future.feedbackFailResultCount.incrementAndGet();
         }
         // 累加執行完成任務數量
         future.feedbackResultCount.incrementAndGet();
         if (future.isDone()) {
             FUTURES.remove(id);
             future.doReceived();
         }
     }
 }
 private void doReceived() {
     lock.lock();
     try {
         if (done != null) {
             // 喚醒阻塞的線程
             done.signal();
         }
     } finally {
         lock.unlock();
     }
 }

 

下麵我們來總結整個流程:

  • 收到 MQ 消息,組裝成 DefaultFuture,通過 get 方法獲取執行結果,未執行完的時候此方法阻塞。

  • 通過切麵切入加了 EventListener 的方法,判斷是否有異常來判斷任務的執行結果。

  • 通過 DefaultFuture.received() 反饋結果。

  • 反饋時計算是否全部完成,全部完成則喚醒阻塞的線程。DefaultFuture.get()就能獲取到結果。

  • 是否要進行 ACK 操作。

需要註意的是每個 EventListener 內部消費的邏輯都要做冪等控制。

 


您的分享是我們最大的動力!

更多相關文章
  • 13 約定 A common problem that arises when wrapping C libraries is that of maintaining reliability and checking for errors. The fact of the matter is tha ...
  • 構建生命周期 Maven的生命周期(lifecycle)可以理解為由Maven的各種plugin按照一定的順序執行來完成java項目清理、編譯、打包、測試、佈署等整個項目的流程的一個過程。 Maven內置了各種插件,如果再pom中沒有顯示配置就會調用預設的內置插件,如果pom中配置了就會調用配置的插 ...
  • 介面 恰當的原則是優先使用類而不是介面。從類開始,如果使用介面的必要性變得很明確,那麼就重構。介面是一個偉大的工具,但它們容易被濫用。 介面中可添加靜態方法與預設方法 一個類實現一個介面的同時必須實現該介面的所有方法(可以不用實現預設方法即關鍵詞為為 default的方法) extends 只能用於 ...
  • 小白是一名.net程式員,之前小白介紹了過了自己的博客系統http://www.ttblog.site/,用.net寫厭了,所以想學下java嘗嘗鮮,於是小白準備用spring boot來實現一個博客內容管理系統。 因為管理系統要有自己的數據源,但是又要從博客系統獲取博客內容,所以第一反應是要弄一個 ...
  • 初次看到原文我是有一些震撼的,原來作為開發人員,閑暇時間還算可以做這麼多有趣程式的開發。閱讀時暫且拋棄你所使用的語言的限制,你是否也能夠在“無聊”之時找到一個開發者的樂趣。 閱讀以下內容時重點關註項目的創意性,並結合自己的獨特經歷進行拓展,你一定也能夠找到編程的樂趣所在。很多項目都可以通過不同的技術 ...
  • 前言 本文的文字及圖片來源於網路,僅供學習、交流使用,不具有任何商業用途,版權歸原作者所有,如有問題請及時聯繫我們以作處理。 知識點 • 企業資產介紹 • 財務分析方法 • 企業資產數據爬取 • 企業資產數據展示 企業資產介紹 企業的資產包括流動資產、固定資產、無形資產、股東權益等等,本次給大家介紹 ...
  • 主要梳理一下SpringBoot2.x的依賴關係和依賴的版本管理,依賴版本管理是開發和管理一個SpringBoot項目的前提。 SpringBoot其實是通過starter的形式,對spring-framework進行裝箱,消除了(但是相容和保留)原來的XML配置,目的是更加便捷地集成其他框架,打造 ...
  • 1.迭代器 迭代器為我們提供了統一遍歷容器(List/Map/Set)的方式 1.遍歷List或Set 2.遍歷Map 2.Collections工具類 類java.util.Collections提供對Set、List、Map進行排序、填充、查找元素的輔助方法 1.void sort(List): ...
一周排行
  • C#6.0新特性 C#7.0新特性 C#8.0新特性 ...
  • out變數 可以直接在方法中使用out申明變數 int.TryParse("123", out var result); 元組 元組的申明 var alphaBetaStart = (alpha: "a", beta: "b"); Console.WriteLine($"{alphaBetaStar ...
  • 在我們的項目中,通常會把數據存儲到關係型資料庫中,比如Oracle,SQL Server,Mysql等,但是關係型資料庫對於併發的支持並不是很強大,這樣就會造成系統的性能不佳,而且存儲的數據多為結構化數據,對於非結構數據(比如文本)和半結構化數據(比如JSon) 就顯得不夠靈活,而非關係型資料庫則很 ...
  • 這幾天終於弄懂了async和await的模式,也搞明白了一直在心裡面積壓著的許多問題,所以寫一篇博客來和大家分享一下。 關於非同步機制我認為只要記住的以下幾點,就可以弄明白了: 1.我認為async和awwait兩個修飾符中最關鍵的是await,async是由於方法中包含await修飾符之後才在方法定 ...
  • 實現WCF的步驟如下: 設計服務協議 實現服務協議 配置服務 托管服務 生成客戶端(這步可有可無) 設計或定義服務協議要麼使用介面,要麼使用類。建議介面,使用介面好處一堆例如修改介面的實現,但是服務協定有無需改變。 設計服務協議,介面上使用 ServiceContractAttribute ,方法上 ...
  • 什麼鬼,我的CPF快寫好了,你居然也要搞跨平臺UI框架?什麼Maui? 之前怎麼不早說要搞跨平臺UI框架呢?看到谷歌搞flutter眼紅了?明年年底發佈?又搞這種追別人屁股的爛事情。 什麼MVU模式?模仿Dart?用C#代碼直接寫UI的模式和我的CPF很像啊。 當初我考慮過XML,Json來描述UI ...
  • 寫在前面 Docker作為開源的應用容器引擎,可以讓我們很輕鬆的構建一個輕量級、易移植的容器,通過Docker方式進行持續交付、測試和部署,都是極為方便的,並且對於我們開發來說,最直觀的優點還是解決了日常開發中的環境配置與部署環境配置上的差異所帶來的種種疑難雜症,從此推脫產品的措辭也少了——“我電腦 ...
  • 一、前言 回顧:認證授權方案之授權初識 從上一節中,我們在對授權系統已經有了初步的認識和使用,可以發現,asp.net core為我們提供的授權策略是一個非常強大豐富且靈活的認證授權方案,能夠滿足大部分的授權場景。 在ConfigureServices中配置服務:將授權服務添加到容器 public ...
  • 項目背景: 工作之餘兼職一家公司(方向是工業4.0)給做IM系統,主要功能包括:文字、 圖片、文件傳輸、遠程協助、視頻語音等等。這些功能都是基於群會話, 比如工廠操作工人遇到問題,請求遠程專家,這個初級專家不能解決問題,會邀請一個高級專家進來解決。開發過程中主要遇到的問題是視頻和語音這一塊,像其他的... ...
  • 基礎概念 Microsoft中間語言(MSIL),也成為通用中間語言(CIL),是一組與平臺無關的指令,由特定於語言的編譯器從源代碼生成。MSIL是獨立於平臺的,因此,他可以在任何公共語言基礎架構支持特定的環境上執行。 通過JIT編譯器將MSIL轉換為特定電腦環境的特定機器代碼。這是在執行MSIL ...