架構設計 | 非同步處理流程,多種實現模式詳解

来源:https://www.cnblogs.com/cicada-smile/archive/2020/06/04/13046699.html
-Advertisement-
Play Games

本文源碼:GitHub·點這裡 || GitEE·點這裡 一、非同步處理 1、非同步概念 非同步處理不用阻塞當前線程來等待處理完成,而是允許後續操作,直至其它線程將處理完成,並回調通知此線程。 必須強調一個基礎邏輯,非同步是一種設計理念,非同步操作不等於多線程,MQ中間件,或者消息廣播,這些是可以實現非同步處理 ...


本文源碼:GitHub·點這裡 || GitEE·點這裡

一、非同步處理

1、非同步概念

非同步處理不用阻塞當前線程來等待處理完成,而是允許後續操作,直至其它線程將處理完成,並回調通知此線程。

必須強調一個基礎邏輯,非同步是一種設計理念,非同步操作不等於多線程,MQ中間件,或者消息廣播,這些是可以實現非同步處理的方式。

同步處理和非同步處理相對,需要實時處理並響應,一旦超過時間會結束會話,在該過程中調用方一直在等待響應方處理完成並返回。同步類似電話溝通,需要實時對話,非同步則類似簡訊交流,發送消息之後無需保持等待狀態。

2、非同步處理優點

雖然非同步處理不能實時響應,但是處理複雜業務場景,多數情況都會使用非同步處理。

  • 非同步可以解耦業務間的流程關聯,降低耦合度;
  • 降低介面響應時間,例如用戶註冊,非同步生成相關信息表;
  • 非同步可以提高系統性能,提升吞吐量;
  • 流量削峰即把請求先承接下來,然後在非同步處理;
  • 非同步用在不同服務間,可以隔離服務,避免雪崩;

非同步處理的實現方式有很多種,常見多線程,消息中間件,發佈訂閱的廣播模式,其根據邏輯在於先把請求承接下來,放入容器中,在從容器中把請求取出,統一調度處理。

註意:一定要監控任務是否產生積壓過度情況,任務如果積壓到雪崩之勢的地步,你會感覺每一片雪花都想勇闖天涯。

3、非同步處理模式

非同步流程處理的實現有好多方式,但是實際開發中常用的就那麼幾種,例如:

  • 基於介面非同步響應,常用在第三方對接流程;
  • 基於消息生產和消費模式,解耦複雜流程;
  • 基於發佈和訂閱的廣播模式,常見系統通知

非同步適用的業務場景,對數據強一致性的要求不高,非同步處理的數據更多時候追求的是最終一致性。

二、介面響應非同步

1、流程描述

基於介面非同步響應的方式,有一個本地業務服務,第三方介面服務,流程如下:

  • 本地服務發起請求,調用第三方服務介面;
  • 請求包含業務參數,和成功或失敗的回調地址;
  • 第三方服務實時響應流水號,作為該調用的標識;
  • 之後第三方服務處理請求,得到最終處理結果;
  • 如果處理成功,回調本地服務的成功通知介面;
  • 如果處理失敗,回調本地服務的失敗通知介面;
  • 整個流程基於部分非同步和部分實時的模式,完整處理;

註意:如果本地服務多次請求第三方服務,需要根據流水號判斷該請求的狀態,業務的狀態設計也是極其複雜,要根據流水號和狀態追溯整個流程的執行進度,避免錯亂。

2、流程實現案例

模擬基礎介面

@RestController
public class ReqAsyncWeb {
    private static final Logger LOGGER = LoggerFactory.getLogger(ReqAsyncWeb.class);
    @Resource
    private ReqAsyncService reqAsyncService ;
    // 本地交易介面
    @GetMapping("/tradeBegin")
    public String tradeBegin (){
        String sign = reqAsyncService.tradeBegin("TradeClient");
        return sign ;
    }
    // 交易成功通知介面
    @GetMapping("/tradeSucNotify")
    public String tradeSucNotify (@RequestParam("param") String param){
        LOGGER.info("tradeSucNotify param={"+ param +"}");
        return "success" ;
    }
    // 交易失敗通知介面
    @GetMapping("/tradeFailNotify")
    public String tradeFailNotify (@RequestParam("param") String param){
        LOGGER.info("tradeFailNotify param={"+ param +"}");
        return "success" ;
    }
    // 第三方交易介面
    @GetMapping("/respTrade")
    public String respTrade (@RequestParam("param") String param){
        LOGGER.info("respTrade param={"+ param +"}");
        reqAsyncService.respTrade(param);
        return "NO20200520" ;
    }
}

模擬第三方處理

@Service
public class ReqAsyncServiceImpl implements ReqAsyncService {

    private static final String serverUrl = "http://localhost:8005" ;

    @Override
    public String tradeBegin(String param) {
        String orderNo = HttpUtil.get(serverUrl+"/respTrade?param="+param);
        if (StringUtils.isEmpty(orderNo)){
            return "Trade..Fail...";
        }
        return orderNo ;
    }

    @Override
    public void respTrade(String param) {
        try {
            Thread.sleep(10000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        Thread thread01 = new Thread(
                new RespTask(serverUrl+"/tradeSucNotify?param="+param),"SucNotify");
        Thread thread02 = new Thread(
                new RespTask(serverUrl+"/tradeFailNotify?param="+param),"FailNotify");
        thread01.start();
        thread02.start();
    }
}

三、生產消費非同步

1、流程描述

這裡基於Kafka中間件,演示流程消息生成,消息處理的非同步解耦流程,基本步驟:

  • 消息生成之後,寫入Kafka隊列 ;
  • 消息處理方獲取消息後,進行流程處理;
  • 消息在中間件提供的隊列中持久化存儲 ;
  • 消息發起方如果掛掉,不影響消息處理 ;
  • 消費方如果掛掉,不影響消息生成;

基於這種消息中間件模式,完成業務解耦,提高系統吞吐量,是架構中常用的方式。

2、流程實現案例

消息發送

@Service
public class KafkaAsyncServiceImpl implements KafkaAsyncService {

    @Resource
    private KafkaTemplate<String, String> kafkaTemplate;

    @Override
    public void sendMsg(String msg) {
        // 這裡Topic如果不存在,會自動創建
        kafkaTemplate.send("kafka-topic", msg);
    }
}

消息消費

@Component
public class KafkaConsumer {

    private static Logger LOGGER = LoggerFactory.getLogger(KafkaConsumer.class);

    @KafkaListener(topics = "kafka-topic")
    public void listenMsg (ConsumerRecord<?,String> record) {
        String value = record.value();
        LOGGER.info("KafkaConsumer01 ==>>"+value);
    }
}

註意:這裡就算有多個消息消費方,也只會在一個消費方處理消息,這就是該模式的特點。

四、發佈訂閱非同步

1、流程描述

這裡基於Redis中間件,說明消息廣播模式流程,基本步驟:

  • 提供一個消息傳遞頻道channel;
  • 多個訂閱頻道的客戶端client;
  • 消息通過PUBLISH命令發送給頻道channel ;
  • 客戶端就會收到頻道中傳遞的消息 ;

之所以稱為廣播模式,該模式更註重通知下發,流程交互性不強。實際開發場景:運維總控系統,更新了某類服務配置,通知消息發送之後,相關業務線上的服務在拉取最新配置,更新到服務中。

2、流程實現案例

發送通知消息

@Service
public class RedisAsyncServiceImpl implements RedisAsyncService {

    @Resource
    private StringRedisTemplate stringRedisTemplate ;

    @Override
    public void sendMsg(String topic, String msg) {
        stringRedisTemplate.convertAndSend(topic,msg);
    }
}

客戶端接收

@Service
public class ReceiverServiceImpl implements ReceiverService {

    private static final Logger LOGGER = LoggerFactory.getLogger("ReceiverMsg");

    @Override
    public void receiverMsg(String msg) {
        LOGGER.info("Receiver01 收到消息:msg-{}",msg);
    }
}

配置廣播模式

@Configuration
public class SubMsgConfig {

    @Bean
    RedisMessageListenerContainer container(RedisConnectionFactory factory,
                                            MessageListenerAdapter msgListenerAdapter,
                                            MessageListenerAdapter msgListenerAdapter02){
        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        container.setConnectionFactory(factory);
        //註冊多個監聽,訂閱一個主題,實現消息廣播
        container.addMessageListener(msgListenerAdapter, new PatternTopic("topic:msg"));
        container.addMessageListener(msgListenerAdapter02, new PatternTopic("topic:msg"));
        return container;
    }

    @Bean
    MessageListenerAdapter msgListenerAdapter(ReceiverService receiverService){
        return new MessageListenerAdapter(receiverService, "receiverMsg");
    }
    @Bean
    MessageListenerAdapter msgListenerAdapter02(ReceiverService02 receiverService02){
        return new MessageListenerAdapter(receiverService02, "receiverMsg");
    }

    @Bean
    ReceiverService receiverService(){
        return new ReceiverServiceImpl();
    }
    @Bean
    ReceiverService02 receiverService02(){
        return new ReceiverServiceImpl02();
    }
}

這裡配置了多個訂閱的客戶端。

五、任務積壓監控

生成一個消息,就因為有一個處理該消息的任務要執行,這就導致任務可能出現積壓的情況,常見原因大致有如下幾個:

  • 任務產生的服務過多,任務處理的服務過少,不均衡;
  • 任務處理時間太長,也導致生產過剩;
  • 中間件本身容量偏小,需要擴容或集群化管理;

如果任務積壓過多,可能要對任務生成進行流量控制,或者提升任務的處理能力,從而避免雪崩情況。

六、源代碼地址

GitHub·地址
https://github.com/cicadasmile/data-manage-parent
GitEE·地址
https://gitee.com/cicadasmile/data-manage-parent

推薦閱讀:《架構設計系列》,蘿蔔青菜,各有所需

序號 標題
01 架構設計:單服務.集群.分散式,基本區別和聯繫
02 架構設計:分散式業務系統中,全局ID生成策略
03 架構設計:分散式系統調度,Zookeeper集群化管理
04 架構設計:介面冪等性原則,防重覆提交Token管理
05 架構設計:緩存管理模式,監控和記憶體回收策略

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

-Advertisement-
Play Games
更多相關文章
  • canvas 和 webGL 這兩項圖形技術結合 css3 可以說能完成絕大部分的動畫和需求。但 canvas 和 webGL 畢竟是偏向底層的繪製引擎,某些場景使用起來還是過於繁瑣的,不分場合一律使用錘子解決的行為不值得提倡。svg 在解決排版,圖標,相關動畫還是非常高效的,而且 svg 還是矢量 ...
  • 一、ODS層ODS 全稱是 Operational Data Store,一般對應的是操作性數據存儲,直接面向主題的,也叫數據運營層,通常是最接近數據源中數據的一層,數據源中的數據,經過抽取、洗凈、傳輸,也就是通常說的 ETL 之後的數據存入本層。本層的數據,總體上大多是按照源頭業務系統的分類方式而 ...
  • 一、MPP 架構 1、MPP架構的基礎概念 MPP (Massively Parallel Processing),即大規模並行處理,在資料庫非共用集群中,每個節點都有獨立的磁碟存儲系統和記憶體系統,業務數據根據資料庫模型和應用特點劃分到各個節點上,每台數據節點通過專用網路或者商業通用網路互相連接,彼 ...
  • 背景 有人對Java主流鎖做了下麵全面的梳理。梳理的確實挺好的。但是我看到這張圖,第一個感覺是:記不住。 因為分了太多類,彼此之間沒有什麼聯繫。做PPT可以。如果聊天或者面試,不用紙筆的情況下,就不太好描述了。也不利於對原理和應用的理解。 基於上述的考慮,我就自己系統的梳理一下鎖,希望可以有助於大家 ...
  • 一、使用反射機制來 (1)獲取一個類; (2)獲取類的構造函數 (3)通過構造函數來獲取一個對象 package com.bjpowernode.java_learning; import java.lang.reflect.*; ​ public class D120_1_ConstructerO ...
  • 本教程源碼請訪問:tutorial_demo 一、AOP概述 1.1、概念 AOP:全稱是Aspect Oriented Programming,即:面向切麵編程。 通過預編譯方式和運行期間動態代理實現程式功能的統一維護的一種技術。AOP是OOP的延續,是軟體開發中的一個熱點,也是Spring框架中 ...
  • 本篇主要討論的是不同存儲結構(主要是LSM-tree和B-tree),它們應對的不同場景,所採用的底層存儲結構,以及對應用以提升效率的索引。 所謂資料庫,最基礎的功能,就是保存數據,並且在需要的時候可以方便地檢索到需要的數據。在這個基礎上,演化出了不同的資料庫系統,以及多種索引機制幫助檢索數據。這篇 ...
  • 雙向鏈表的實現 創建3個文件:doubleLinked.h、doubleLinked.c、doubleLinkedTest.c doubleLinked.h #ifndef DOUBLE_LINKED_H_ #define DOUBLE_LINKED_H_ #ifdef lxx_gnuc #defi ...
一周排行
    -Advertisement-
    Play Games
  • 移動開發(一):使用.NET MAUI開發第一個安卓APP 對於工作多年的C#程式員來說,近來想嘗試開發一款安卓APP,考慮了很久最終選擇使用.NET MAUI這個微軟官方的框架來嘗試體驗開發安卓APP,畢竟是使用Visual Studio開發工具,使用起來也比較的順手,結合微軟官方的教程進行了安卓 ...
  • 前言 QuestPDF 是一個開源 .NET 庫,用於生成 PDF 文檔。使用了C# Fluent API方式可簡化開發、減少錯誤並提高工作效率。利用它可以輕鬆生成 PDF 報告、發票、導出文件等。 項目介紹 QuestPDF 是一個革命性的開源 .NET 庫,它徹底改變了我們生成 PDF 文檔的方 ...
  • 項目地址 項目後端地址: https://github.com/ZyPLJ/ZYTteeHole 項目前端頁面地址: ZyPLJ/TreeHoleVue (github.com) https://github.com/ZyPLJ/TreeHoleVue 目前項目測試訪問地址: http://tree ...
  • 話不多說,直接開乾 一.下載 1.官方鏈接下載: https://www.microsoft.com/zh-cn/sql-server/sql-server-downloads 2.在下載目錄中找到下麵這個小的安裝包 SQL2022-SSEI-Dev.exe,運行開始下載SQL server; 二. ...
  • 前言 隨著物聯網(IoT)技術的迅猛發展,MQTT(消息隊列遙測傳輸)協議憑藉其輕量級和高效性,已成為眾多物聯網應用的首選通信標準。 MQTTnet 作為一個高性能的 .NET 開源庫,為 .NET 平臺上的 MQTT 客戶端與伺服器開發提供了強大的支持。 本文將全面介紹 MQTTnet 的核心功能 ...
  • Serilog支持多種接收器用於日誌存儲,增強器用於添加屬性,LogContext管理動態屬性,支持多種輸出格式包括純文本、JSON及ExpressionTemplate。還提供了自定義格式化選項,適用於不同需求。 ...
  • 目錄簡介獲取 HTML 文檔解析 HTML 文檔測試參考文章 簡介 動態內容網站使用 JavaScript 腳本動態檢索和渲染數據,爬取信息時需要模擬瀏覽器行為,否則獲取到的源碼基本是空的。 本文使用的爬取步驟如下: 使用 Selenium 獲取渲染後的 HTML 文檔 使用 HtmlAgility ...
  • 1.前言 什麼是熱更新 游戲或者軟體更新時,無需重新下載客戶端進行安裝,而是在應用程式啟動的情況下,在內部進行資源或者代碼更新 Unity目前常用熱更新解決方案 HybridCLR,Xlua,ILRuntime等 Unity目前常用資源管理解決方案 AssetBundles,Addressable, ...
  • 本文章主要是在C# ASP.NET Core Web API框架實現向手機發送驗證碼簡訊功能。這裡我選擇是一個互億無線簡訊驗證碼平臺,其實像阿裡雲,騰訊雲上面也可以。 首先我們先去 互億無線 https://www.ihuyi.com/api/sms.html 去註冊一個賬號 註冊完成賬號後,它會送 ...
  • 通過以下方式可以高效,並保證數據同步的可靠性 1.API設計 使用RESTful設計,確保API端點明確,並使用適當的HTTP方法(如POST用於創建,PUT用於更新)。 設計清晰的請求和響應模型,以確保客戶端能夠理解預期格式。 2.數據驗證 在伺服器端進行嚴格的數據驗證,確保接收到的數據符合預期格 ...