我正在使用 Kombu 在 Redis 上构建一个生产者/消费者设置，但我遇到了一个问题。如果我启动一个消费者，然后使用 range(10000) 启动生产者，我可以确认生产者已将所有 10k 项排队，但并非所有 10k 项都被消费者接收。Kombu 或 Redis 是否有我不知道的限制？它似乎可以在 range(9000) 下正常工作，并且所有键/ack 都已正确耗尽。

如果我运行以下代码，我发现我可以从队列中获取消息，但callback无法触发

我试着跑python -m pdb test.py

它循环在

我正在努力将带有 Kombu 的消息传递系统放在一起，但我遇到了以下问题。假设我有消息被路由到路由键“x”和“y”。这在“x”和“y”之间没有依赖关系的情况下非常有用。

但是，考虑另一种情况，我将数据发送到路由键“a”、“b”和“c”，并且单个队列正在从这些路由键中获取消息。如果我需要来自 a、b 和 c 的数据一起处理单个回调，有没有办法将这些消息聚合到单个工作线程中，或者这是消息传递范式的完全混蛋？

我知道我可以将消息缓存在其他地方（例如，Redis），并且仅在我拥有必要的数据时才进行处理，但我想知道 Kombu 是否可以做到这一点，而不必每次都缓存数据并唤醒工作人员。感谢您的任何建议；如果有帮助，我可以给出一些代码示例。

我有两台服务器，分别称为 A 和 B。B 运行 RabbitMQ，而 A 通过 Kombu 连接到 RabbitMQ。如果我在 B 上重新启动 RabbitMQ，kombu 连接会中断，并且不再传递消息。然后我必须重置 A 上的进程以重新建立连接。有没有更好的方法，即 Kombu 有没有办法自动重新连接，即使 RabbitMQ 进程重新启动？

我的基本代码实现如下，在此先感谢！:)

问题

我有一个RabbitMQ 服务器，用作我的一个系统的队列集线器。在过去一周左右的时间里，它的生产者每隔几个小时就会完全停止。

我试过什么

蛮力

• 停止消费者会释放锁几分钟，然后阻塞返回。
• 重启 RabbitMQ 解决了几个小时的问题。
• 我有一些自动脚本可以执行丑陋的重启，但它显然远非正确的解决方案。

分配更多内存

按照cantSleepNow 的回答，我已将分配给 RabbitMQ 的内存增加到 90%。服务器拥有高达 16GB 的内存，并且消息数量不是很高（每天数百万），所以这似乎不是问题。

从命令行：

并与/etc/rabbitmq/rabbitmq.config：

代码与设计

我为所有消费者和生产者使用 Python。

生产者

生产者是服务调用的 API 服务器。每当呼叫到达时，就会打开连接，发送消息并关闭连接。

消费者

消费者之间略有不同，但通常使用以下模式 - 打开连接，并等待它直到消息到达。连接可以长时间保持打开状态（例如，几天）。

设计推理

• 消费者始终需要与队列服务器保持开放连接
• 生产者会话应该只存在于 API 调用的生命周期内

直到大约一周前，这种设计还没有引起任何问题。

Web 视图仪表板

Web 控制台显示消费者进入127.0.0.1并172.31.38.50阻止消费者访问172.31.38.50、172.31.39.120和。172.31.41.38172.31.41.38

系统指标

为了安全起见，我检查了服务器负载。正如预期的那样，平均负载和 CPU 利用率指标很低。

为什么兔子MQ每次都会出现这样的死锁？

我已将我的项目配置为使用 Celery-Kombu0-SQS 和 Celery pidbox 队列的组合，这些队列为我运行 Celery worker 的每个节点创建。我试图抑制/避免这些队列。我浏览了其他博客并了解到 SQS 默认禁用此选项，但我仍然看到正在创建 pidboxes。

如果无法抑制 pidbox 的创建或配置名称，我该如何抑制它？

I'm trying to write some tests where I produce a message to a queue and see if the message gets correctly consumed and handled in the application.

For that I'm playing around with the kombu library and especially the in-memory Transport implementation.

Still I can't get it working, that the produced message gets consumed.

My questions is therefore, if anyone can provide a simple unit test that produces and consumes a message in-memory

过去，我使用过 Celery async python 和 Django 应用程序，其中发送方和接收方都了解发送到 RabbitMQ 以通过 Celery 进行处理的任务（可能是集群中的同一个应用程序等）。

我现在有一个用例，其中 .NET 服务以 JSON 的形式向 RabbitMQ 发布消息，每个队列一种消息类型。.NET 应用程序只会发布一条消息并确保它被 Rabbit 正确接收，然后走开。然后，我将运行一个 Django 应用程序来查看消息，因此我不确定使用这些 JSON 消息的正确方法是什么，因此这个 Django 应用程序可以简单地通过其模型存储数据并确认消息已处理。

使用 Celery/Kombu 我不确定访问队列的最佳方式，以便我们每个队列都有一个直接消费者。我知道 Celery 在幕后使用 Kombu，所以我想我可以在那里创建消费者，但后来我预见不可能像 Celery 和 Flower 那样管理流程，然后在应用程序启动时创建一个流氓线程消费者似乎很不稳定最好。

我在为新的 SO 文档项目记录 Kombu 时偶然发现了这个问题。

考虑以下消费者 Mixin的 Kombu 代码：

代码失败：

因为消息被 ACK 了两次， onprint_even_characters()和print_odd_characters()。

一个可行的简单解决方案是仅确认最后一个回调函数，但如果我想在其他队列或连接上使用相同的函数，它会破坏模块化。

如何确认发送到多个回调函数的排队 Kombu 消息？

我已经使用Kombu了一段时间了 - 今天当我准备一台新机器进行部署时，我TypeError正在import从kombu：