问题标签 [pika]

Pika 库支持 tornado 适配器，这里是一个关于如何使用异步适配器发布消息的示例。

我想在tornado应用程序中使用pika，只是一个例子，我想把tornado请求数据放到RabbitMQ，但是不知道怎么做。

两个问题不知道怎么解决。

1 Pika使用tornado适配器有自己的ioloop，

Tornado 应用程序有自己的 ioloop，

如何结合这两个ioloop？

2 Pika 示例一次又一次地发布相同的消息，但我想发布请求数据，如何将请求数据传递给发布方法？

发布到队列时，我需要将元数据附加到每条消息。问题是哪种方法更有效？

1. 将自定义字段添加到每个邮件正文
2. 为每条消息添加自定义标题

以防万一：

• 发布者在 AWS m1.small 上
• 消息速率小于 500 msgs/s
• 兔子库：pika（python）

我有一个从 RabbitMQ 读取消息的分布式系统。在我现在的情况下，我需要处理不超过 N msgs/s。

例如：想象一下A发送短信 (SMS) 的服务。该服务只能处理 100 msgs/s。我的系统有多个从 RabbitMQ 读取数据的消费者。每条消息都需要先处理，然后发送到服务A。处理时间总是不同的。

所以问题：

• 是否可以将队列配置为向多个消费者分派不超过 100 msgs/s 的速度？

我想知道是否可以设置队列中的最大消息数？

假设我希望队列 Foo 中的消息不超过 100 个，有可能吗？

我有 Flask 应用程序并在其中实现了 RPC。

引擎.py

api.py

当 Flask 应用程序初始化时，它运行 method init(host)。

现在我需要测试我的应用程序如何响应 RPC 调用。所以我写了以下 RPC 客户端：

客户端.py

然后我基于 Flask-Testing 框架编写了测试。它在每个测试方法之间初始化 Flask 应用程序，因此我们可以与之交互。

测试.py

此测试在拨打电话时挂起。它在这里无限迭代：

来自客户端.py

这意味着on_response()客户端上的该方法从未调用过。

如果我用 CTRL-C 中断我的测试，我将看到以下堆栈跟踪：

我尝试运行应用程序并从单独的脚本访问它：

如您所见，代码与测试中的代码相同，但在这种情况下它可以工作。

这个问题的原因可能是什么？可能是因为 Flask-Testing 在一个进程中同时运行应用程序和客户端？如何检查它/编写正确的测试？

我正在使用 pika 0.9.13 编写一些 python 代码来连接到 RabbitMQ。我正在创建许多子进程（可能有 1000 个），并希望每个进程都能够发送到 RabbitMQ。仔细阅读，似乎最好的方法是创建一个连接，然后在该连接中创建通道。

任何人都可以建议如何最好地做到这一点？通常，代码如下所示：-

我应该使用什么类型的适配器（BlockingConnection？）...

提前致谢 ！

我正在尝试不断地监听队列，但是大约一分钟后（假设我的队列是空的）我被这个错误断开了：

这是我的代码：

我正在寻找任何帮助。我想修复一个insert_order_queue()功能，以便能够将消息重新发送到RabbitMQ如果消息实际上没有传递到服务器，

这是我当前的代码：

这是我的_connect()方法：

我在测试 pika 时看到了这个异常

当我只使用默认值时它可以工作。但是，当我在连接参数中添加凭据时，它引发了异常。

它给了我这个错误：

有什么可以帮助我的吗？谢谢。

我正在将 RabbitMQ 与 pika (Python) 一起使用，我想知道是否有办法确定消息被重新排队的次数？

一般的想法是，我希望我的工作人员能够告诉它正在处理的消息之前已被拒绝/重新排队 X 次，然后采取适当的措施来进行一些错误处理，然后明确地从如果达到 Y 次重试的限制，则队列。

我知道我可能会复制消息，增加优先级属性的值（甚至增加标头中的值），然后将副本发布到队列，然后确认原始消息（类似于方法我发现了这个例子），但我更愿意明确地取消并重新排队消息。

有任何想法吗？