Spring Integration 在没有轮询器的情况下侦听队列答案

【问题标题】：Spring Integration listen on queue without pollerSpring Integration 在没有轮询器的情况下侦听队列
【发布时间】：2016-12-31 18:57:54
【问题描述】：

我想使用 Spring Integration 实现 HTTP 端点，它监听 http 请求，将请求数据作为消息发送到通道，另一个端点应该监听该通道上的消息并处理它们。

听起来很简单。但我想要实现的是：

应按顺序处理消息。
应尽快处理消息（如果队列已为空，则在 http 请求后立即处理）。
http 请求应在收到消息后立即响应，而不是在处理完之后，因此发送方将只知道收到了该消息以进行处理。
我不想使用外部队列，例如 RabbitMQ。

所以我需要一个QueueChannel。但如果我理解正确，从队列接收消息的唯一方法是poller。所以第2点将不满足。收到消息后到 poller 看到它之前会有一点延迟。

所以问题是：在我没有看到的 Spring Integration 中是否有任何简单的方法可以实现这一点？

当然我可以自己实现。例如创建SmartLifeCycle 组件，它监听DirectChannel 并将消息放入java.util.concurrent.BlockingQueue，并启动一个专用线程，该线程将在此队列上等待并将消息发送到另一个DirectChannel 进行处理。所以不会有延迟，因为只要BlockingQueue 不为空，线程就会被解除阻塞。

这听起来像是一种“模式”——基于专用线程的两个直接通道之间的一些队列。

也许有一种更简单的方法，已经在 Spring Integration 中实现了，由于缺乏这方面的经验，我只是没有看到？

【问题讨论】：

标签： java spring spring-integration

【解决方案1】：

即使使用轮询器，也可以满足第 2 点 - 只需将 fixed-delay 设置为 0 和/或增加接收超时（默认为 1 秒）； poller 线程将阻塞在队列中，直到消息到达；然后立即再次等待。

您还可以使用执行器通道（http 线程移交给执行器线程）。

【讨论】：

谢谢你，加里！我没有意识到receiveTimeout 实际上是一个时间，轮询器的线程将等待支持BlockingQueue 上的新消息，直到它释放调度程序的线程。在此之后，它将根据轮询器的Trigger（例如delay）再次占用线程。因此，0 和 receiveTimeout 的延迟 1s 轮询器将几乎一直阻塞一些调度程序的线程，并尽快处理消息！非常感谢你！ *附注使用executor内部队列的单线程executor的executor通道也是我没有想到的可行方案！
请记住，在服务器崩溃的情况下，这些选项中的任何一个都可能导致消息丢失 - 如果这对您来说是个问题，则需要使用一些持久性存储（例如 rabbitmq）应用程序。
是的，我知道这一点。但是即使使用 RabbitMq，消息也会丢失。调用者和入站 http-endpoint 之间的网络连接，应用程序在收到消息后但在发送到 RabbitMq 的队列之前崩溃，调用者的崩溃 - 仅举几例。因此，与其使用外部队列（无论如何都不能保证“至少一次”），我最好安排在最后 X 分钟内从外部源定期获取所有消息，并以幂等方式处理它们。某种“补偿提取器”来处理可能丢失的消息。
>But message can be lost even when using RabbitMq. - 这并不完全正确；在返回 HTTP 响应之前，您可以使用事务或发布者确认来验证消息已到达代理。但这没关系 - 我只是想确保您（或任何阅读本文的人）知道。
是的，但问题是：将 http 请求发送到我的系统的外部系统（我之前评论中的“调用者”）根本不关心我的响应，或者我是否响应一点也不。 :) 所以它不会将消息（使用另一个 http 请求）重新发送到我的系统。无论如何，感谢您的所有回复！我之前多次阅读 SI 文档，但仍然没有完全理解队列通道是如何工作的。看了你的回答，我真的明白了。