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

本文源码:GitHub·点这里 || GitEE·点这里 一、异步处理 1、异步概念 异步处理不用阻塞当前线程来等待处理完成,而是允许后续操作,直至其它线程将处理完成,并回调通知此线程。 必须强调一...

本文源码: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 架构设计:缓存管理模式,监控和内存回收策略
  • 发表于 2020-06-05 08:53
  • 阅读 ( 146 )
  • 分类:网络文章

条评论

请先 登录 后评论
不写代码的码农
小编

篇文章

作家榜 »

  1. 小编 文章
返回顶部
部分文章转自于网络,若有侵权请联系我们删除