java - 当使用 RabbitMQ 作为 java 工作队列时，应该如何处理并发和 transient 错误？

Question

我正在考虑设置一个 RabbitMQ 代理来处理 Java Web 应用程序的基本任务处理。基本思想是，生产者是一个 Web 服务器，想要快速处理请求，但也需要做一些数据处理。为了实现这一点，它知道一些可以序列化为 AMQP 消息的作业 DTO，并且在某个地方有一个知道如何处理这些作业的消费者。经典。

阅读 RabbitMQ 文档后，我仍然有几个问题。

1. 我希望消费者应用程序充分利用其 CPU 来处理消息，我想知道 RabbitMQ 是否已经提供了工作池。假设我想要 4 个工作线程，是否足以在同一连接上注册 4 个 channel ，每个 channel 有 1 个消费者？在这种情况下，工作将在 handleDelivery() 中完成，如果成功完成，将发送一个 ack。或者更确切地说，我应该使用 1 个消费者并在我自己的应用程序层管理一组工作人员吗？

2. 消费者将与数据库对话，这意味着会发生暂时性错误；死锁、乐观锁定违规、数据库服务器重启等。如果消费者无法处理工作，预期的行为是什么？发出 nack 等待 broker 重新发送作业，还是延迟 ack 直到确定作业无法完成？还是无所谓？

值得注意的是，作业通常不会超过几秒钟，除非数据库正在重新启动或发生其他情况。我感谢任何指导!

Answer 1

1a.如果您想要固定数量的工作线程，那么您应该打开多个 channel 并为每个 channel 注册一个消费者。 RabbitMQ 每个 channel 有一个调度程序线程，因此一个消费者将阻塞在同一 channel 上注册的其他消费者。

1b。如果您需要动态增长的工作人员池，那么您可以考虑使用 spring-rabbit (假设您使用 Java 实现)。具体来说，SimpleMessageListenerContainer 管理一个根据工作负载动态增长和收缩的消费者池。参见章节 "Listeners Concurrency"在 Spring AMQP 引用指南中。

2a。如果你知道工作人员失败了，你可以 nack 并且消息将被传递给另一个消费者。除非是“自动确认”模式，否则在消费者确认消息之前，消息不会传递给另一个消费者。