问题标签 [python-pika]

我需要收听 RabbitMQ 消息，稍微处理每条消息，然后将其提交到另一个交换器。到目前为止，我看到的每个示例都包括以下内容：

或这个：

因为我需要接收和发送消息，我可能需要同时运行两个循环。有没有办法我可以做到这一点？

我在 Ubuntu 14.04 上安装了 RabbitMQ，但无法从网络上的其他计算机连接到 RabbitMQ 服务器。

从安装 RabbitMQ 的机器连接到服务器没有问题，只有从网络上的其他计算机连接。

该页面声明“默认情况下，RabbitMQ 将在所有可用接口上侦听端口 5672”。

当尝试在 python 中使用 pika 从另一个服务器连接时，我收到以下错误：

我添加了一个具有权限的新用户set_permissions newuser ".*" ".*" ".*"并尝试了 URI 身份验证方法，但是我仍然收到相同的错误。

我也无法从网络上的其他计算机连接到 RabbitMQ 管理，但可以在本地 pc 上访问它。

此外，通过检查开放端口，我可以看到以下内容：

我正在使用 RabbitMQ 教程中的 python 代码：

我错过了什么？

我正在使用 RabbitMQ 生产者向消费者发送长时间运行的任务（30 分钟以上）。问题是当与服务器的连接关闭并且未确认的任务重新排队时，消费者仍在处理任务。

通过研究，我了解到可以使用心跳或增加连接超时 来解决此问题。这两种解决方案在尝试它们时都会引发错误。在阅读类似帖子的答案时，我还了解到，自发布答案以来，RabbitMQ 已经实现了许多更改（例如，默认心跳超时已从 RabbitMQ 3.5.5 之前的 580 更改为 60）。

指定心跳和阻塞的连接超时时：

显示以下错误：

在连接参数中指定heartbeat_interval=1000时，会显示类似的错误：TypeError: __init__() got an unexpected keyword argument 'heartbeat_interval'

同样socket_timeout = 1000显示以下错误：TypeError: __init__() got an unexpected keyword argument 'socket_timeout'

我在 Ubuntu 14.04 上运行 RabbitMQ 3.6.1、pika 0.10.0 和 python 2.7。

1. 为什么上述方法会产生错误？
2. 可以在长时间运行的连续任务中使用心跳方法吗？例如，在执行需要 30 多分钟的大型数据库连接时可以使用心跳吗？我赞成心跳方法，因为很多时候很难判断数据库连接等任务需要多长时间。

我已经阅读了类似问题的答案

更新：运行pika 文档中的代码会产生相同的错误。

我正在尝试在容器中使用 rabbitmq。我安装了所需的库，然后尝试创建一个 BlockingConnection，但它失败了。当我在任何虚拟机或物理主机上尝试时，同样的事情。它仅对容器失败。以下是步骤：-

安装所需的软件包：-

接下来，我尝试创建一个 BlockingConnection，但它失败了。不过，同样的事情在虚拟机上也有效。

我不确定容器内有什么问题。我为此使用 ubuntu:15.10 图像。对此的任何反馈都会非常有帮助。

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

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

相同的代码可以在 Ubuntu 上运行，而不是在 Centos 上运行！防火墙已经关闭！

Ubuntu 16.04，python 版本 3.5.2。

Centos 7，python 版本 3.5.2。

Ubuntu和centos是virtualbox新安装的！RabbitMq 配置 tls！</p>

在 Centos 上，如果 connect rabbitmq disable ssl 是可以的，但是如果 connect rabbitmq enable ssl 失败。

你能帮助我吗？非常感谢！

这是 rabbitmq 配置：

这是代码：

这是错误消息：

这个rabbitmq服务器日志：

我有一个回调函数，它在收到每条消息时返回 AMQP 消息的正文。我希望能够将正文的内容分配给一个全局变量，然后在我的程序中进一步使用它。

我在脚本顶部创建了一个全局变量，并在回调函数中使用了 global 关键字，我理解它应该允许我更新脚本顶部的全局变量，但是当我从另一个脚本我从回调函数中返回了'test'的原始值而不是body的当前值......

消费者.py

进口商.py

我不明白为什么 T_MSG 的值没有从回调函数中更新，因为我可以清楚地看到 print 语句正在打印 body 的正确值 - 但即使我在回调函数中将其声明为全局，当我导入 T_MSG 的值并尝试打印它，我得到了最初分配的“测试”值，而不是我期望的正文内容......

我的印象是，每次回调函数接收到一条消息并“处理”它时，应该将 body 的新值分配给 T_MSG （全局）变量，因此我应该能够将它导入到我的 importer.py 脚本中- 但这不起作用，我读过的所有内容都指向在函数中使用 global 关键字来使这个...

任何帮助深表感谢！

是否可以将pika( python-pika) 中的队列声明为循环队列？如果是，如何？

我的意思是循环队列（或环）是一个队列，当一条消息被选中时，它将被重新插入到队列的末尾，而不是从队列中删除。例如，如果队列包含：msg_3, msg_2, msg_1, 那么在一个消费者 get 之后msg_1, 后者将被插入到最后, 所以队列将是: msg_1, msg_3, msg_2(而不是msg_3, msg_2)

编辑：正如IMSoP所建议的，我可以让消费者重新发布每条消费的消息（例如，在回调函数的末尾）。

我有一个想要实现的简单 docker 构建。拉取基础镜像 odoo:9 并安装 pika 库

来自 odoo:9

运行 pip 安装 pika

但无法继续，因为我收到此错误。顺便说一句，在这里输入图像描述 ，我在 Windows 环境中使用 docker。

我有 2 个队列，比如 q1 和 q2，它们对应于 e1 和 e2 交换与绑定密钥 b1 和 b2。我想并行运行消费者函数，比如 c1 和 c2，它们将分别监听 q1 和 q2。我尝试了以下方法：

但是，它只监听 c1 函数和 c2 函数，并没有被执行。如何运行这两个功能？提前致谢。

编辑：我在 2 个不同的模块（文件）中有方法 c1 和 c1