将 HTTP 请求/响应与 RabbitMQ 中的异步消息集成答案

【问题标题】：Integrate HTTP request/response with asynchronous messages in RabbitMQ将 HTTP 请求/响应与 RabbitMQ 中的异步消息集成
【发布时间】：2020-01-16 05:14:01
【问题描述】：

我们的应用程序是一个消息处理系统，具有多个与 RabbitMQ 队列连接的组件。所以消息处理是异步的。现在我需要添加一个与系统通信的 HTTP 适配器。由于 HTTP 与请求/响应同步，因此我需要一种连接同步和异步流的方法。目前的解决方案是：

HTTP 请求被发送到一个队列。每个请求都有一个唯一的关联请求 ID。
HTTP 请求被CompletableFuture 阻止。
处理请求并将响应发送回另一个队列。
队列消费者使用响应来完成CompletableFuture与请求ID的匹配。

HTTP 适配器是使用 Akka HTTP 实现的。使用handleWithAsyncHandler() 处理请求，函数类型为Function<HttpRequest, CompletionStage<HttpResponse>>。

问题在于 HTTP 适配器需要管理所有待处理请求的映射 (Map<String, CompletableFuture>)。对于每个请求，都会创建一个新的 CompletableFuture 对象并将其放入映射中。当队列中收到响应时，匹配的CompletableFuture就完成了请求。这在代码中似乎是一种难闻的气味，因为我需要仔细管理这张地图。例如，如果无法为请求生成响应，则需要从地图中删除该请求。

我想知道除了使用地图来跟踪所有待处理的请求之外，是否还有其他方法。

【问题讨论】：

虽然对于这个用例来说它可能有点重量级，但 Apache Camel 能够相对轻松地完成这个确切的概念。对于最基本的用例，我想您只需几行代码或 XML 配置即可将其连接起来。
它看起来像一个long polling HTTP 服务器。我认为这个discussion 很有帮助。

标签： java rabbitmq akka-http

【解决方案1】：

基本上，akka-http 可以是异步样式。您不需要实现该队列来映射请求 Id。

需要考虑的一点是不要使用默认调度程序。

最好定义一个阻塞调度程序来处理 CompletableFuture.supplyAsync

例如

my-blocking-dispatcher {
  type = Dispatcher
  executor = "thread-pool-executor"
  thread-pool-executor {
    fixed-pool-size = 16
  }
  throughput = 1
}

import static akka.http.javadsl.server.Directives.completeWithFuture;
import static akka.http.javadsl.server.Directives.post;

// GOOD (the blocking is now isolated onto a dedicated dispatcher):
final Route routes = post(() -> {
    final MessageDispatcher dispatcher = system.dispatchers().lookup("my-blocking-dispatcher");
    return completeWithFuture(CompletableFuture.supplyAsync(() -> {
                try {
                    Thread.sleep(5000L);
                } catch (InterruptedException e) {
                }
                return HttpResponse.create()
                        .withEntity(Long.toString(System.currentTimeMillis()));
            }, dispatcher // uses the good "blocking dispatcher" that we
            // configured, instead of the default dispatcher to isolate the blocking.
    ));
});

【讨论】：