架构设计 | 异步处理流程，多种实现模式详解

Posted 2021-04-10 猿天地

转自：知了一笑  作者：知了一笑

一、异步处理

1、异步概念

异步处理不用阻塞当前线程来等待处理完成，而是允许后续操作，直至其它线程将处理完成，并回调通知此线程。

必须强调一个基础逻辑，异步是一种设计理念，异步操作不等于多线程，MQ中间件，或者消息广播，这些是可以实现异步处理的方式。

2、异步处理优点

虽然异步处理不能实时响应，但是处理复杂业务场景，多数情况都会使用异步处理。

• 异步可以解耦业务间的流程关联，降低耦合度；

• 降低接口响应时间，例如用户注册，异步生成相关信息表；

• 异步可以提高系统性能，提升吞吐量；

• 流量削峰即把请求先承接下来，然后在异步处理；

• 异步用在不同服务间，可以隔离服务，避免雪崩；

异步处理的实现方式有很多种，常见多线程，消息中间件，发布订阅的广播模式，其根据逻辑在于先把请求承接下来，放入容器中，在从容器中把请求取出，统一调度处理。

注意：一定要监控任务是否产生积压过度情况，任务如果积压到雪崩之势的地步，你会感觉每一片雪花都想勇闯天涯。

3、异步处理模式

异步流程处理的实现有好多方式，但是实际开发中常用的就那么几种，例如：

• 基于接口异步响应，常用在第三方对接流程；

• 基于消息生产和消费模式，解耦复杂流程；

• 基于发布和订阅的广播模式，常见系统通知

异步适用的业务场景，对数据强一致性的要求不高，异步处理的数据更多时候追求的是最终一致性。

二、接口响应异步

1、流程描述

基于接口异步响应的方式，有一个本地业务服务，第三方接口服务，流程如下：

• 本地服务发起请求，调用第三方服务接口；

• 第三方服务实时响应流水号，作为该调用的标识；

• 之后第三方服务处理请求，得到最终处理结果；

• 如果处理成功，回调本地服务的成功通知接口；

• 如果处理失败，回调本地服务的失败通知接口；

• 整个流程基于部分异步和部分实时的模式，完整处理；

注意：如果本地服务多次请求第三方服务，需要根据流水号判断该请求的状态，业务的状态设计也是极其复杂，要根据流水号和状态追溯整个流程的执行进度，避免错乱。

2、流程实现案例

模拟基础接口

@RestControllerpublic 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" ; }}

模拟第三方处理

@Servicepublic 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、流程实现案例

消息发送

@Servicepublic class KafkaAsyncServiceImpl implements KafkaAsyncService {
 @Resource private KafkaTemplate<String, String> kafkaTemplate;
 @Override public void sendMsg(String msg) { // 这里Topic如果不存在，会自动创建 kafkaTemplate.send("kafka-topic", msg); }}

消息消费

@Componentpublic 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、流程实现案例

发送通知消息

@Servicepublic class RedisAsyncServiceImpl implements RedisAsyncService {
 @Resource private StringRedisTemplate stringRedisTemplate ;
 @Override public void sendMsg(String topic, String msg) { stringRedisTemplate.convertAndSend(topic,msg); }}

客户端接收

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

配置广播模式

@Configurationpublic 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

相关推荐

• 
  

• 
  

• 
  

后台回复 学习资料 领取学习视频

如有收获，点个在看，诚挚感谢