RabbitMQ:用慢速消费者限制快速生产者对抗大队列

Question

我们目前正在使用 RabbitMQ，其中一个持续超快的生产者与受有限资源(例如慢速 MySQL 插入)限制的消费者配对。

我们不喜欢用 x-max-length 声明队列，因为一旦达到限制，所有消息都将被丢弃或死信，我们不想丢失消息。

添加更多的消费者很容易，但他们都将受到一个共享资源的限制，所以这是行不通的。问题仍然存在:如何减慢生产者的速度？

当然，我们可以在 Redis、memcached、MySQL 或其他东西中放置一个流控制标志，生产者读取为 pointed out in an answer to a similar question ，或者也许更好，生产者可以定期测试队列长度和 throttle 本身，但这些对我来说似乎是黑客。

我主要是质疑我是否有根本的误解。我原以为这是一个常见的场景，所以我想知道:

限制生产者的最佳实践是什么？ RabbitMQ 是如何做到的？或者你以完全不同的方式做到这一点？

背景

假设制作人实际上知道如何通过正确的输入放慢速度。例如。硬件传感器或硬件随机数生成器，可以根据需要生成任意数量的事件。

在我们特定的真实案例中，我们有一个 API，用户可以使用它来添加消息。如果队列“已满”，我们希望通过让我们的 API 返回错误来应用背压，而不是吞噬和丢弃消息，因此调用者/用户知道要退避，或者让 API 阻塞直到消费者 catch 。我们不控制我们的用户，所以无论消费者有多快，我都可以创建一个更快的生产者。

我希望有类似 TCP 套接字的 API 之类的东西，其中 write()可以屏蔽和哪里有select()可用于确定句柄是否可写。因此，要么使用 RabbitMQ API 块，要么让它在队列已满时返回错误。

Answer 1

对于 x-max-length 属性，您说过您不希望消息被丢弃或死信。我看到有一个更新，为此添加了更多功能。正如我所见，它在文档中指定:

“使用溢出设置来配置队列溢出行为。如果溢出设置为拒绝发布，最近发布的消息将被丢弃。此外，如果发布者确认已启用，发布者将通过基本通知被拒绝。消息”

因此，据我所知，您可以使用队列限制来拒绝来自发布者的新消息，从而将一些背压推到上游。