使用 Camel 的异步非阻塞组播答案

【问题标题】：Async Nonblocking Multicast with Camel使用 Camel 的异步非阻塞组播
【发布时间】：2020-03-24 14:12:49
【问题描述】：

我有两个查询： 1. 我们可以在骆驼中拥有非阻塞异步路由吗？我确实看到了与seda 的异步，但是如果卸载工作到其他阻塞的线程。 2. 如果可以，我们可以在这样的路由中使用多播吗？以下是我的多步骆驼路线，似乎可行。但不确定它是异步还是非阻塞异步。

from("direct:multiStep")
             .to("bean:routeHandler?method=monoReturningMethod1")
             .process(new UnwrapStreamProcessor())
             .to("bean:routeHandler?method=monoReturningMethod2")
             .process(new UnwrapStreamProcessor())

以上工作，网络请求有来自monoReturningMethod 的响应。在这种情况下，我想确保所有进程都是非阻塞的。

对于多播，我正在尝试以下路线。不知道把UnwrapStreamProcessor放在哪里。我试图把它放在end() 之后，但它不起作用。我需要自定义 Processor 吗？或者我怎样才能将所有Mono 回报捆绑在一起？

from("direct:incoming")
         .multicast()
         .parallelProcessing()
         .to("bean:routeHandler?method=monoReturningMethod1", "bean:routeHandler?method=monoReturningMethod2")
         .end()

我正在使用带有 spring boot 启动器的 apache `camel 3.0.1。

@Component("routeHandler")
public class RouteHandler {
     Mono<Entity> monoReturningMethod1(Exchange exchange) {
          //make some WebClient request which returns Mono.
     }
     Mono<Entity> monoReturningMethod2(Exchange exchange) {
          //make some WebClient request which returns Mono.
     }
}

此路由处理传入的网络请求。如何使所有路由处理非阻塞和异步。我曾尝试在monoReturningMehtod 之后使用process(new UnwrapStreamProcessor()) 作为流程步骤，如果我按顺序执行它就可以了。但它不适用于多播，并且不允许覆盖原始消息。有什么建议吗？

PS：我正在启动我的异步流程，如下所示： producerTemplate.asyncSend("RouteName", exchange)

【问题讨论】：

标签： asynchronous apache-camel nonblocking spring-camel

【解决方案1】：

According to this page，异步路由引擎支持所有EIPs，但不是所有endpoint types (components)。一些组件只有有限的支持，即它们只能异步消费或生产。

您可以在路由中使用Threads DSL 告诉 Camel，从现在开始，消息应该在新线程中异步路由。

【讨论】：