需要验证的方法：Spring Integration + AMQP + Async

Question

我最近一直在研究 Spring Integration 和 AMQP (RabbitMQ)，因为我需要使用异步方法与两个应用程序（中间件和后端）通信，以便中间件在接收客户端调用时不会阻塞。

我首先采用了更简单的同步实现方法，这意味着我在中间件上有一个网关接口和一个出站网关（requiresReply=true），然后在后端有一个入站网关和一个服务激活器.这种初始方法效果很好（我使用了 Spring Integration XML 配置）。

现在我需要说明以异步方式进行这项工作的方法。

通过查看 RabbitMQ 教程 6，最好使用回调队列和相关 ID，根据我的理解，这类似于调用 Spring RabbitTemplate 的 convertAndSend() 然后 receive()，而不是 convertSendAndReceive( )（在收到响应之前会一直阻塞）。

我查看了 Spring Integration 文档，我需要替换中间件上的网关接口以使其返回 Future 或 ListenableFuture。

Async Gateway

完成后，我还查看了outbound gateway 的文档，其中说它可以与 RabbitTemplate 一起管理correlationID 和replyTo 消息属性。

我的问题是：

1. 为了使这项工作使用异步方法，我是否应该继续使用出站/入站网关，而不是出站/入站消息转换器？
2. 如果遵循出站/入站消息转换器方法（在我看来这类似于 RabbitMQ 教程所示），我如何将网关接口上的 Future 与从入站通道适配器返回的结果相关联？

Answer 1

说实话，您并没有提供原始的业务需求。这可能是一个事实，真的没有理由处理这个asynchandoff，因为你有一个@Gateway作为入口点，它是无线程的，即使它被阻止等待回复它也没有'不影响可能执行类似sendAndReceive 操作的其他线程。在大多数情况下，在同一个请求者线程中完成所有操作就足够了，并且不会因为转移到共享的ThreadPoolExecutor 而降低性能。

没错，Future 允许您释放调用者一点，以便准备好接受同一线程中的新请求。

因为它是一个MessagingGateway 并且无论如何你都想得到一个回复​​，所以有一个与请求相关联的钩子 - TemporaryReplyChannel 标头。这就是<outbound-gateway> 正常工作的原因：它将阻塞回复放置到网关的return（或FutureTask#set()）的通道。

我想说我们可以通过您的异步回复要求实现相同的TemporaryReplyChannel 增益。

1. 您应该使用inbound/outbound 通道适配器对。
2. 在将消息发送到<int-amqp:outbound-channel-adapter> 之前，您应该为<header-enricher> 执行此操作<header-channels-to-string>。
3. 服务器端可能相同 - <int-amqp:inbound-gateway>
4. 您应该使用 fixed replyQueue 作为通过 <int-amqp:outbound-channel-adapter> 发送的邮件的标头
5. <int-amqp:inbound-channel-adapter> 应该配置为 fixed replyQueue。
6. 客户端的<int-amqp:outbound-channel-adapter> 和<int-amqp:inbound-gateway> 都必须为mapped-request-headers="*" 配置，以允许将reply-channel 标头传播到服务器，反之亦然。
7. 客户端的<int-amqp:inbound-channel-adapter> 只会将回复发送到reply-channel，就像<int-amqp:outbound-gateway> 一样
8. 您可能需要手动处理correlationId，因为<int-amqp:inbound-gateway> 可能需要它才能正确生成回复。

嗯，是这样的……

HTH

欢迎提出更多问题。如果我误解了您的问题，请纠正我。