本文源碼：GitHub·點(diǎn)這里 || GitEE·點(diǎn)這里

一、異步處理

1、異步概念

異步處理不用阻塞當(dāng)前線程來(lái)等待處理完成，而是允許后續(xù)操作，直至其它線程將處理完成，并回調(diào)通知此線程。

必須強(qiáng)調(diào)一個(gè)基礎(chǔ)邏輯，異步是一種設(shè)計(jì)理念，異步操作不等于多線程，MQ中間件，或者消息廣播，這些是可以實(shí)現(xiàn)異步處理的方式。

同步處理和異步處理相對(duì)，需要實(shí)時(shí)處理并響應(yīng)，一旦超過(guò)時(shí)間會(huì)結(jié)束會(huì)話，在該過(guò)程中調(diào)用方一直在等待響應(yīng)方處理完成并返回。同步類(lèi)似電話溝通，需要實(shí)時(shí)對(duì)話，異步則類(lèi)似短信交流，發(fā)送消息之后無(wú)需保持等待狀態(tài)。

2、異步處理優(yōu)點(diǎn)

雖然異步處理不能實(shí)時(shí)響應(yīng)，但是處理復(fù)雜業(yè)務(wù)場(chǎng)景，多數(shù)情況都會(huì)使用異步處理。

• 異步可以解耦業(yè)務(wù)間的流程關(guān)聯(lián)，降低耦合度；
• 降低接口響應(yīng)時(shí)間，例如用戶注冊(cè)，異步生成相關(guān)信息表；
• 異步可以提高系統(tǒng)性能，提升吞吐量；
• 流量削峰即把請(qǐng)求先承接下來(lái)，然后在異步處理；
• 異步用在不同服務(wù)間，可以隔離服務(wù)，避免雪崩；

異步處理的實(shí)現(xiàn)方式有很多種，常見(jiàn)多線程，消息中間件，發(fā)布訂閱的廣播模式，其根據(jù)邏輯在于先把請(qǐng)求承接下來(lái)，放入容器中，在從容器中把請(qǐng)求取出，統(tǒng)一調(diào)度處理。

注意：一定要監(jiān)控任務(wù)是否產(chǎn)生積壓過(guò)度情況，任務(wù)如果積壓到雪崩之勢(shì)的地步，你會(huì)感覺(jué)每一片雪花都想勇闖天涯。

3、異步處理模式

異步流程處理的實(shí)現(xiàn)有好多方式，但是實(shí)際開(kāi)發(fā)中常用的就那么幾種，例如：

• 基于接口異步響應(yīng)，常用在第三方對(duì)接流程；
• 基于消息生產(chǎn)和消費(fèi)模式，解耦復(fù)雜流程；
• 基于發(fā)布和訂閱的廣播模式，常見(jiàn)系統(tǒng)通知

異步適用的業(yè)務(wù)場(chǎng)景，對(duì)數(shù)據(jù)強(qiáng)一致性的要求不高，異步處理的數(shù)據(jù)更多時(shí)候追求的是最終一致性。

二、接口響應(yīng)異步

1、流程描述

基于接口異步響應(yīng)的方式，有一個(gè)本地業(yè)務(wù)服務(wù)，第三方接口服務(wù)，流程如下：

• 本地服務(wù)發(fā)起請(qǐng)求，調(diào)用第三方服務(wù)接口；
• 請(qǐng)求包含業(yè)務(wù)參數(shù)，和成功或失敗的回調(diào)地址；
• 第三方服務(wù)實(shí)時(shí)響應(yīng)流水號(hào)，作為該調(diào)用的標(biāo)識(shí)；
• 之后第三方服務(wù)處理請(qǐng)求，得到最終處理結(jié)果；
• 如果處理成功，回調(diào)本地服務(wù)的成功通知接口；
• 如果處理失敗，回調(diào)本地服務(wù)的失敗通知接口；
• 整個(gè)流程基于部分異步和部分實(shí)時(shí)的模式，完整處理；

注意：如果本地服務(wù)多次請(qǐng)求第三方服務(wù)，需要根據(jù)流水號(hào)判斷該請(qǐng)求的狀態(tài)，業(yè)務(wù)的狀態(tài)設(shè)計(jì)也是極其復(fù)雜，要根據(jù)流水號(hào)和狀態(tài)追溯整個(gè)流程的執(zhí)行進(jìn)度，避免錯(cuò)亂。

2、流程實(shí)現(xiàn)案例

模擬基礎(chǔ)接口

@RestController public class ReqAsyncWeb {private static final Logger LOGGER = LoggerFactory.getLogger(ReqAsyncWeb.class);@Resourceprivate 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" ;@Overridepublic String tradeBegin(String param) {String orderNo = HttpUtil.get(serverUrl+"/respTrade?param="+param);if (StringUtils.isEmpty(orderNo)){return "Trade..Fail...";}return orderNo ;}@Overridepublic 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();} }

三、生產(chǎn)消費(fèi)異步

1、流程描述

這里基于Kafka中間件，演示流程消息生成，消息處理的異步解耦流程，基本步驟：

• 消息生成之后，寫(xiě)入Kafka隊(duì)列 ；
• 消息處理方獲取消息后，進(jìn)行流程處理；
• 消息在中間件提供的隊(duì)列中持久化存儲(chǔ) ；
• 消息發(fā)起方如果掛掉，不影響消息處理 ；
• 消費(fèi)方如果掛掉，不影響消息生成；

基于這種消息中間件模式，完成業(yè)務(wù)解耦，提高系統(tǒng)吞吐量，是架構(gòu)中常用的方式。

2、流程實(shí)現(xiàn)案例

消息發(fā)送

@Service public class KafkaAsyncServiceImpl implements KafkaAsyncService {@Resourceprivate KafkaTemplate<String, String> kafkaTemplate;@Overridepublic void sendMsg(String msg) {// 這里Topic如果不存在，會(huì)自動(dòng)創(chuàng)建kafkaTemplate.send("kafka-topic", msg);} }

消息消費(fèi)

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

注意：這里就算有多個(gè)消息消費(fèi)方，也只會(huì)在一個(gè)消費(fèi)方處理消息，這就是該模式的特點(diǎn)。

四、發(fā)布訂閱異步

1、流程描述

這里基于Redis中間件，說(shuō)明消息廣播模式流程，基本步驟：

• 提供一個(gè)消息傳遞頻道channel；
• 多個(gè)訂閱頻道的客戶端client；
• 消息通過(guò)PUBLISH命令發(fā)送給頻道channel ；
• 客戶端就會(huì)收到頻道中傳遞的消息 ；

之所以稱(chēng)為廣播模式，該模式更注重通知下發(fā)，流程交互性不強(qiáng)。實(shí)際開(kāi)發(fā)場(chǎng)景：運(yùn)維總控系統(tǒng)，更新了某類(lèi)服務(wù)配置，通知消息發(fā)送之后，相關(guān)業(yè)務(wù)線上的服務(wù)在拉取最新配置，更新到服務(wù)中。

2、流程實(shí)現(xiàn)案例

發(fā)送通知消息

@Service public class RedisAsyncServiceImpl implements RedisAsyncService {@Resourceprivate StringRedisTemplate stringRedisTemplate ;@Overridepublic void sendMsg(String topic, String msg) {stringRedisTemplate.convertAndSend(topic,msg);} }

客戶端接收

@Service public class ReceiverServiceImpl implements ReceiverService {private static final Logger LOGGER = LoggerFactory.getLogger("ReceiverMsg");@Overridepublic void receiverMsg(String msg) {LOGGER.info("Receiver01 收到消息：msg-{}",msg);} }

配置廣播模式

@Configuration public class SubMsgConfig {@BeanRedisMessageListenerContainer container(RedisConnectionFactory factory,MessageListenerAdapter msgListenerAdapter,MessageListenerAdapter msgListenerAdapter02){RedisMessageListenerContainer container = new RedisMessageListenerContainer();container.setConnectionFactory(factory);//注冊(cè)多個(gè)監(jiān)聽(tīng),訂閱一個(gè)主題，實(shí)現(xiàn)消息廣播container.addMessageListener(msgListenerAdapter, new PatternTopic("topic:msg"));container.addMessageListener(msgListenerAdapter02, new PatternTopic("topic:msg"));return container;}@BeanMessageListenerAdapter msgListenerAdapter(ReceiverService receiverService){return new MessageListenerAdapter(receiverService, "receiverMsg");}@BeanMessageListenerAdapter msgListenerAdapter02(ReceiverService02 receiverService02){return new MessageListenerAdapter(receiverService02, "receiverMsg");}@BeanReceiverService receiverService(){return new ReceiverServiceImpl();}@BeanReceiverService02 receiverService02(){return new ReceiverServiceImpl02();} }

這里配置了多個(gè)訂閱的客戶端。

五、任務(wù)積壓監(jiān)控

生成一個(gè)消息，就因?yàn)橛幸粋€(gè)處理該消息的任務(wù)要執(zhí)行，這就導(dǎo)致任務(wù)可能出現(xiàn)積壓的情況，常見(jiàn)原因大致有如下幾個(gè)：

• 任務(wù)產(chǎn)生的服務(wù)過(guò)多，任務(wù)處理的服務(wù)過(guò)少，不均衡；
• 任務(wù)處理時(shí)間太長(zhǎng)，也導(dǎo)致生產(chǎn)過(guò)剩；
• 中間件本身容量偏小，需要擴(kuò)容或集群化管理；

如果任務(wù)積壓過(guò)多，可能要對(duì)任務(wù)生成進(jìn)行流量控制，或者提升任務(wù)的處理能力，從而避免雪崩情況。

六、源代碼地址

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

推薦閱讀：《架構(gòu)設(shè)計(jì)系列》，蘿卜青菜，各有所需

序號(hào)標(biāo)題

01	架構(gòu)設(shè)計(jì)：單服務(wù).集群.分布式，基本區(qū)別和聯(lián)系
02	架構(gòu)設(shè)計(jì)：分布式業(yè)務(wù)系統(tǒng)中，全局ID生成策略
03	架構(gòu)設(shè)計(jì)：分布式系統(tǒng)調(diào)度，Zookeeper集群化管理
04	架構(gòu)設(shè)計(jì)：接口冪等性原則，防重復(fù)提交Token管理
05	架構(gòu)設(shè)計(jì)：緩存管理模式，監(jiān)控和內(nèi)存回收策略

總結(jié)

以上是生活随笔為你收集整理的架构设计 | 异步处理流程，多种实现模式详解的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

如果覺(jué)得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。