使用 RabbitMQ 和 Camel 一条一条地消费消息

Question

我有一个 RabbitMQ 代理和一个使用消息的 Camel Java 应用程序。消费者在 ca 中处理一条消息。 1秒。我产生 10'000 条消息；这几乎是瞬间的。 RabbitMQ 控制台立即报告队列不包含更多内容。 7'000 条消息。这对我来说是个问题，因为如果消费者失败，大约有 3'000 条消息会丢失。

我尝试了几个选项：

• 流缓存：camelContext.setStreamCaching(false); // 或 true
• 限制：from(queue).throttle(1)
• 预取：from(rabbitmq:exchange&queue=q&prefetchEnabled=true&prefetchCount=1&prefetchGlobal=false&prefetchSize=0

一切都无济于事。我从未观察到队列中的消息数量有规律地减少（10'000，然后是 9'999，9'998 等）。恰恰相反：我总是可以看到消息以大块（通常是 3'000 条消息）从队列中出来，尽管消费者实际上处理它们的速度很慢。

我怎样才能一一消费消息？

Answer 1

我知道这对你来说可能有点晚了，但我也一直在与同样的问题作斗争，我终于找到了一个解决方案，让 Camel RabbitMq 组件一个接一个地消费消息。

您只需将autoAck 属性设置为false。

下面是整个路线定义的示例：

rabbitmq:your-exchange?routingKey=rtkey&queue=qname&autoDelete=false&durable=true&guaranteedDeliveries=true&threadPoolSize=1&autoAck=false

那么重要的部分就是：

guaranteedDeliveries=true&threadPoolSize=1&autoAck=false

threadPoolSize 和 guaranteedDeliveries 属性不是实现您想要的东西所必需的，我只是为了完整起见将它们放在这里。

这是link to the documentation

Answer 2

在事务中运行此路由。这样一来，失败的消费者最终会将消息回滚到队列中。