如何使用 Spring Integration 轮询 AMQP 队列答案

【问题标题】：How to poll an AMQP queue using Spring Integration如何使用 Spring Integration 轮询 AMQP 队列
【发布时间】：2014-07-28 08:09:24
【问题描述】：

我有一个用例，我需要等待 2 个小时才能使用来自 AMQP（我们使用 Rabbit）队列的消息。

编辑：为了澄清我的用例...我需要每条消息等待 2 小时才能被阅读。例如。消息 1 于上午 10 点到达，消息 2 于 10:15 到达。我需要在下午 12 点阅读消息 1，在下午 12:15 阅读消息 2。

我们正在使用 Spring Integration 3.x。

int-amqp:inbound-channel-adapter 是消息驱动的，没有我能找到的轮询选项。

我想到的几件事：

将 auto-startup 设置为 false 并使用石英作业手动启动入站通道适配器。
创建我自己的基于轮询的自定义 SimpleMessageListenerContainer（不确定这有多容易）
使用这个方法在rabbitmq中配置一个“延迟队列”：How to create a delayed queue in RabbitMQ?
编辑：添加第 4 个选项：使用 delayer 将每条消息延迟 2 小时：http://docs.spring.io/spring-integration/docs/3.0.2.RELEASE/reference/html/messaging-endpoints-chapter.html#delayer

有什么建议吗？

【问题讨论】：

标签： spring rabbitmq spring-integration spring-amqp

【解决方案1】：

我们目前没有轮询入站适配器。 #1 很简单。对于 #2，最简单的方法是使用 RabbitTemplate 并从 POJO 中的 inbound-channel-adapter 调用 receive()。

我会选择#1；你不需要quartz，你可以使用一个简单的Spring定时任务和一个控制总线来启动适配器。

【讨论】：

感谢您的信息。在周末重新考虑之后，我不确定这两种解决方案都能解决我的问题。我没有很好地说明我的用例。我需要让每条消息等待 2 小时。我认为我的问题中的选项 3 或新添加的选项 4 可能是最好的解决方案。
对于那个用例（每条消息延迟 2 小时），那么选项 #3 是要走的路；我不会介绍延迟器。

【解决方案2】：

另一个技巧即将使用PollableAmqpChannel：

<int-amqp:channel id="myQueueName" message-driven="false"/>

并为该频道的订阅者提供<poller>。

没有理由向该频道发送消息（因为您将从 Rabbit Queue 轮询消息），对，它看起来像 anti-pattern，但它是一个钩子，如何避免直接使用 RabbitTemplate 的任何变通办法通过 SpEL。

更新

<delayer> 可以帮助您，但这取决于您的要求。如果你不想从 RabbitMQ 轮询消息，你应该使用上面的解决方法。但是，如果您只是不想在一段时间后处理消息，则可以将其“延迟”到那个时间。

不要忘记添加持久化message-store，以避免在此期间丢失消息和意外应用程序失败。

【讨论】：

谢谢。这是一个很好的解决方法。现在的一个问题...我不确定可轮询通道是否适合我，因为如果我每 2 小时轮询一次仅 1 条消息，则队列可能会得到真正的备份。更新了我的问题以澄清我的用例。
添加了关于此事的 cmets

【解决方案3】：

仅供参考，我是如何解决这个问题的。（使用的解决方案 #3）。

<rabbit:queue name="delayQueue" durable="true">
  <rabbit:queue-arguments>
    <entry key="x-message-ttl">
       <value type="java.lang.Long">7200000</
     </entry>
     <entry key="x-dead-letter-exchange" value="finalDestinationTopic"/>
     <entry key="x-dead-letter-routing-key" value="finalDestinationQueue"/>
  </rabbit:queue-arguments>
</rabbit:queue>

<rabbit:topic-exchange name="finalDestinationTopic">
  <rabbit:bindings>
    <rabbit:binding queue="finalDestinationQueue" pattern="finalDestinationQueue"/>
  </rabbit:bindings>
</rabbit:topic-exchange>

【讨论】：