一、异步处理

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

推荐阅读：《架构设计系列》，萝卜青菜，各有所需

什么是回调？今天傻傻地截了张图问了下，然后被陈大牛回答道“就一个回调…”。此时千万个草泥马飞奔而过

哈哈，看着源码，享受着这种回调在代码上的作用，真是美哉。不妨总结总结。

一、什么是回调

回调，回调。要先有调用，才有调用者和被调用者之间的回调。所以在百度百科中是这样的：

软件模块之间总是存在着一定的接口，从调用方式上，可以把他们分为三类：同步调用、回调和异步调用。

回调是一种特殊的调用，至于三种方式也有点不同。

1、同步回调，即阻塞，单向。

2、回调，即双向(类似自行车的两个齿轮)。

3、异步调用，即通过异步消息进行通知。

二、CS中的异步回调(java案例)

比如这里模拟个场景：客户端发送msg给服务端，服务端处理后(5秒)，回调给客户端，告知处理成功。代码如下：

回调接口类：

/**

* @author Jeff Lee

* @since 2015-10-21 21:34:21

* 回调模式-回调接口类

*/

public interface CSCallBack {

public void process(String status);

}

模拟客户端：

/**

* @author Jeff Lee

* @since 2015-10-21 21:25:14

* 回调模式-模拟客户端类

*/

public class Client implements CSCallBack {

private Server server;

public Client(Server server) {

this.server = server;

}

public void sendMsg(final String msg){

System.out.println("客户端：发送的消息为：" + msg);

new Thread(new Runnable() {

@Override

public void run() {

server.getClientMsg(Client.this,msg);

}

}).start();

System.out.println("客户端：异步发送成功");

}

@Override

public void process(String status) {

System.out.println("客户端：服务端回调状态为：" + status);

}

}

模拟服务端:

/**

* @author Jeff Lee

* @since 2015-10-21 21:24:15

* 回调模式-模拟服务端类

*/

public class Server {

public void getClientMsg(CSCallBack csCallBack , String msg) {

System.out.println("服务端：服务端接收到客户端发送的消息为:" + msg);

// 模拟服务端需要对数据处理

try {

Thread.sleep(5 * 1000);

} catch (InterruptedException e) {

e.printStackTrace();

}

System.out.println("服务端:数据处理成功，返回成功状态 200");

String status = "200";

csCallBack.process(status);

}

}

测试类：

/**

* @author Jeff Lee

* @since 2015-10-21 21:24:15

* 回调模式-测试类

*/

public class CallBackTest {

public static void main(String[] args) {

Server server = new Server();

Client client = new Client(server);

client.sendMsg("Server,Hello~");

}

}

运行下测试类 — 打印结果如下：

客户端：发送的消息为：Server,Hello~

客户端：异步发送成功

服务端：服务端接收到客户端发送的消息为:Server,Hello~

(这里模拟服务端对数据处理时间，等待5秒)

服务端：数据处理成功，返回成功状态 200

客户端：服务端回调状态为：200

一步一步分析下代码，核心总结如下

1、接口作为方法参数，其实际传入引用指向的是实现类

2、Client的sendMsg方法中，参数为final，因为要被内部类一个新的线程可以使用。这里就体现了异步。

3、调用server的getClientMsg()，参数传入了Client本身(对应第一点)。

还有值得一提的是

— 开源代码都在我的gitHub上哦~

三、回调的应用场景

回调目前运用在什么场景比较多呢？从操作系统到开发者调用：

1、Windows平台的消息机制

2、异步调用微信接口，根据微信返回状态对出业务逻辑响应。

3、Servlet中的Filter(过滤器)是基于回调函数，需容器支持。

补充：其中 Filter(过滤器)和Interceptor(拦截器)的区别，拦截器基于是Java的反射机制，和容器无关。但与回调机制有异曲同工之妙。

总之，这设计让底层代码调用高层定义(实现层)的子程序，增强了程序的灵活性。

四、模式对比

上面讲了Filter和Intercepter有着异曲同工之妙。其实接口回调机制和一种

观察者模式：

GOF说道 — “定义对象的一种一对多的依赖关系，当一个对象的状态发送改变的时候，所有对他依赖的对象都被通知到并更新。”它是一种模式，是通过接口回调的方法实现的，即它是一种回调的体现。

接口回调：

与观察者模式的区别是，它是种原理，而非具体实现。

五、心得

总结四步走：

机制，即是原理。

模式，即是体现。

记住具体场景，常见模式。

然后深入理解原理。

来源：51CTO