网络规划设计师视频网盘,自动优化网站软件没有了,抖音代运营套餐价格表,深圳被点名前言在很多互联网应用系统中#xff0c;请求处理异步化是提升系统性能一种常用的手段#xff0c;而基于消息系统的异步处理由于具备高可靠性、高吞吐量的特点#xff0c;因而在并发请求量比较高的互联网系统中被广泛应用。与此同时#xff0c;这种方案也带来了调用链路处理…前言在很多互联网应用系统中请求处理异步化是提升系统性能一种常用的手段而基于消息系统的异步处理由于具备高可靠性、高吞吐量的特点因而在并发请求量比较高的互联网系统中被广泛应用。与此同时这种方案也带来了调用链路处理上的问题因为大部分应用请求都会要求同步响应实时处理结果而由于请求的处理过程已经通过消息异步解耦所以整个调用链路就变成了异步链路此时请求链路的发起者如何同步拿到响应结果就需要进行额外的系统设计考虑。为了更清晰地理解这个问题小码哥以最近正在做的共享单车的IOT系统为例给大家来一张图描述下如图所示在上述系统流程中终端设备与服务端之间通过MQTT协议相连而MQTT协议本质上是一种异步消息的连接方式因此业务应用如图中的订单系统发起开锁请求后IOT应用系统会以MQTT协议的方式通过物联网平台此处使用的是AWS IOT服务向设备发起开锁下行消息而这一过程在IOT应用系统完成与物联网平台的交互后同步调用链路就结束了因为物联网平台与锁设备之间通过MQTT消息服务异步解耦了当然物联网平台会通过一系列可靠消息机制来确保开锁消息能够发送到指定设备的监听MQTT队列。而至于锁设备是否能够及时接收到开锁下行MQTT消息则取决于锁设备当时的移动网络情况。锁设备在收到MQTT开锁消息后会通过嵌入式软件系统触发硬件设备完成开锁动作之后就需要通过MQTT上行消息将开锁结果反馈到服务端从而由服务端系统判断是否创建骑行订单并计算费用。这一过程需要物联网平台监听指定锁设备相应的MQTT上行消息队列由于物联网平台需要与上万个设备进行连接因而不可能将每一个锁设备所产生的MQTT上行消息都直接转发给Iot应用系统因此在物联网平台可以将一类的设备MQTT消息转发至特定的业务消息队列中例如开锁上行消息所有设备的MQTT开锁响应上行消息都可以转发到表示开锁响应的Iot业务消息队列如“iot_upstream_lock_response”这样Iot业务系统则不需要再关注底层设备的MQTT消息可以用更利于业务理解的方式去处理开锁响应结果。现在的问题是通过MQTT协议的开锁下行消息、上行消息已经完全处于两条不同的异步网络链路而链路的发起者此时却需要同步等待开锁结果但是实际上同步链路早已在Iot应用系统向物联网平台发送开锁消息后就已经完成所以为了满足调用方的同步请求/响应需要就需要在Iot应用系统的下发开锁消息后进行额外的同步阻塞等待并监听开锁响应的Iot业务消息队列“iot_upstream_lock_response”关于此次开锁请求的上行消息之后再结束掉之前的同步阻塞等待逻辑从而实现向业务调用方返回实时开锁响应结果的同步调用效果。那么在上述流程中如何实施额外的同步阻塞以及如何进行回调消息的监听呢在接下来的内容中就和大家一起探讨具体的实施方案解决方案分析以上问题在使用消息服务进行异步解耦的应用场景中是比较普遍的需求由于异步调用链路非常长所以通用的解决思路是在调用链的起始端进行同步阻塞而在调用链的结束端通过回调的方式来实现如下图所示在上述图示中链路起始队列处在发送第一次异步消息后会开启一个临时队列并同步阻塞监听该临时队列的回调消息而链路的结束队列在完成逻辑处理后需要回调起始队列监听的临时队列而由于请求线程一直处于阻塞监听该临时队列的状态所以一旦收到回调消息就可以结束阻塞执行后续流程从而完成整个链路的同步响应。虽然常见的消息中间件都可以实现以上逻辑例如小码哥之前所在的公司就基于RabbitMQ通过临时队列的方式实现过消息链路的同步调用但是基于消息中间件的方式多少还是显得有些繁琐对于常见的消息中间件如RocketMQ、RabbitMQ来说异步消息才是其强项如果以大量临时队列的创建和销毁为代价来实现消息调用链路的同步不仅从使用上来说显得有些麻烦并且也会对消息中间件的稳定性带来一些不好的影响。因此在前面提到的IOT系统中我们采用了基于Redis的发布/订阅功能来实现异步消息链路的同步化调用。而由于Redis的高性能以及Redis的应用场景非常丰富并且非常适合数据频繁变动的场景在系统中既可以作为NoSQL数据库来使用同时还支持分布式锁等功能因而维护的性价比也相对较高。接下来我们就基于Spring Boot的开发框架来演示如何利用Redis的发布/订阅来实现异步消息链路的同步回调Redis发布订阅机制Redis本身可以通过发布订阅机制实现一定的消息队列功能在Redis中通过subscribe/publish等命令可以实现发布订阅功能基于此原先的IOT系统处理示意图如下如上图所示在IOT应用端发送异步MQTT消息后会以消息ID组成的Key作为频道****并保持请求线程对该频道的同步监听直到收到Iot业务消息队列的开锁结果上行消息后在消息队列的消费端将该上行消息发布至同样以消息requestId组成的频道中从而实现基于Redis发布订阅机制的异步消息系统同步调用效果。Spring Boot代码实现下面我们基于Spring Boot演示如何通过代码进行实现创建Spring Boot工程后引入Spring Boot Redis集成依赖包如下!--之后在项目的配置文件中添加Redis服务连接信息如下所示spring:此时项目就具备了访问Redis的能力接下来我们通过具体的代码实现来进行功能演示。订阅监听代码如下RestController
在上述代码中我们模拟了一个开锁请求在完成异步消息处理后会开启Redis订阅监听为了实现异步阻塞还需要我们创建消息任务对象代码如下public 在消息任务对象中我们通过JDK1.8新提供的CompletableFuture类实现线程阻塞效果并通过定义消息监听对象及超时时间完善处理机制。此外根据Controller层代码还需要自定义定义消息监听处理对象代码如下public 此时就完成了Redis服务订阅这部分逻辑的编写在后续的逻辑处理中需要完成消息的发布才能正常结束此处的阻塞等待接下来我们写一段代码来模拟消息发布代码如下RestController
此时启动Spring Boot应用调用开锁模拟接口逻辑就会暂时处于订阅等待状态之后再模拟调用开锁回调Redis消息发布逻辑之前的阻塞等待就会因为监听回调而完成同步返回。
