RabbitMQ 队列的可轮询 StreamListener 或 MessageSource

【问题标题】：Pollable StreamListener or MessageSource for RabbitMQ QueueRabbitMQ 队列的可轮询 StreamListener 或 MessageSource
【发布时间】：2017-11-23 15:47:32
【问题描述】：

我有 Spring Cloud Stream Rabbit 项目，该项目将消息发送到 rabbit exchange errorEx，以防流中出现错误。我想定期（每 5 分钟一次）从队列中监听并处理它。

有什么方法可以让我拥有一个 Pollable @StreamListener？或者我可以配置一个 rabbitMQ org.springframework.integration.core.MessageSource 以便我可以使用 Poller 构建一个 IntegrationFlow 吗？

【问题讨论】：

标签： spring rabbitmq spring-integration spring-cloud spring-cloud-stream

【解决方案1】：

Spring Integration AMQP 不提供可轮询适配器。 @StreamListener 只支持Subscribable 频道。

您可以使用自定义 MessageSource，它使用 RabbitTemplate receive() 操作来获取消息。

或者，您可以根据需要使用消息驱动适配器和stop()/start()。

【讨论】：

使用了CustomMessageSource，我可以做到rabbitTemplate.receive()。然后在 IntegrationFlow 中将其与轮询器一起使用。 IntegrationFlows.from(customMessageSource, c -> c.poller(Pollers.fixedDelay(10000, 10000).maxMessagesPerPoll(1))) .channel(retryChannel()).get()