异步请求使用reactor中mono和flux所遇到的问题

谁践踏了优雅 2021-09-23 06:00 532阅读 0赞

最近在项目升级改造的过程中，需要用到异步请求，技术选型为reactor的mono和flux，实际用的过程中遇到不少问题，简单记录一下，分享给大家避免踩同样的坑。

**项目需求：**  
1、能够同时发起多个请求，多个请求所耗费的总时间，其实是耗时最长的一个请求所用时间。同时要保证最终的结果数据顺序，和发起的请求顺序保持一致。

2、能够在发起请求后，不阻塞后面的操作。请求与后续操作同时执行。当我们需要请求结果时，直接在后面取出数据即可。实现真正的异步，以减少整个功能的耗时。

**技术选型：**  
Reactor有两种类型， Flux<T> 和 Mono<T> 。  
Mono最多只触发一个事件，可以把Mono<Void>用于在异步任务完成时发出通知。  
Flux可以触发零到多个事件，并根据实际情况结束处理或触发错误。可以将多个mono合并成flux。

方案一：  
    同时发起多个mono请求，然后利用merge(),将多个mono合并成flux。  
    取结果时，通过flux.block()来获取请求的数据。  
    问题：      
    1、利用merge()将多次请求进行合并后，会同时发起，但返回的顺序就会发生改变，和发起请求的顺序不一致。  
    2、block()是阻塞的，当执行请求操作时，是不能往下进行后续操作的。

方案二：  
    为了解决顺序问题,合并请求时使用flux.concat()来合并请求。  
    问题：  
    concat()方法为了保证顺序，实际上是将多个请求顺序执行的，这样就又不能满足我们同时发起请求的需求。

方案三：  
    使用zip()方法，将多个请求压缩成一个，这样可以实现多个请求合并成一个，同时发起。  
    问题：  
    参数不能传递集合，只能一个个传递，实现起来不够简化；也不保证顺序问题。

方案四：

为了实现需求2，让程序能够在发起请求后同时执行后续操作，block()肯定是不能使用的。于是使用注册订阅器的方式。  
    将多个mono合并成flux后，为flux注册一个订阅器，订阅器内可以接收返回的数据，并且可以和后续操作同时执行，当我们取数据时直接从订阅器取返回的数据。  
    问题：  
    需求2解决了，但还是没能实现多个请求同时执行，并保证返回顺序。

方案五：  
    不将mono合并成flux，而是为每个mono注册一个订阅器。这样每个订阅器可以并行执行，并且保证顺序，同时不会阻塞后续操作。

目前项目采用的是方案五，暂时满足了需求。有关reactor的资料比较少，还是建议大家多去参考官方文档。