问题标签 [php-amqplib]

我希望我的 RabbitMQ 应用程序在假定消息已发布之前等待发布者确认。这可以通过将通道设置为确认模式然后等待待处理消息被确认来实现。

如果我想定义一个自定义超时（例如 10 秒），并且在时间过去后，应用程序应该打印一条错误消息，以防留下未确认的消息。

使用PHPAMQPLib v2.6.2 到目前为止我已经尝试在nack 处理程序中通过$this->channel->set_nack_handler($callBack)where$callBack是一个处理错误日志记录的函数来实现这个逻辑。问题是我无法判断这种方法是否有效。

在类初始化方法中，我有：

$this->channel->confirm_select();

在处理消息发布的方法中，我有：

$this->channel->wait_for_pending_acks(static::CONFIRM_TIMEOUT);

上述常数CONFIRM_TIMEOUT设置为 10 秒。

RabbitMQ状态的示例代码

当我们的 $channel 有回调时，我们的代码将阻塞。每当我们收到一条消息时，我们的 $callback 函数都会收到收到的消息。

使用此代码段

这让我很困惑，因为默认超时PhpAmqpLib\Channel\AbstractChannel::wait是永远。

public function wait($allowed_methods = null, $non_blocking = false, $timeout = 0)

因此，如果wait永远阻塞，代码将如何到达while循环的第二次迭代？

是否可以安全地说循环仅在传递 awhile时才是必要的？wait$timeout > 0

我在使用 mqtt 和 amqp 通信时遇到问题

重点是：

我有一个带有 MQTT 插件的 RabbitMQ 实例，我也有一个 mqtt 生产者/消费者和一个 AMQP 生产者/消费者（在 symfony RabbitMqBundle 上使用 php-amqplib）

RabbitMQ 文档说所有 MQTT 主题都适用于 amq.topic 交换。

正确的

我的情况：

MQTT 消费者/生产者订阅了 /topic/test 主题 AMQP 消费者在 amq.topic 交换上，在路由键 .topic.test 上（我不知道这一点，但兔子替换 / for . on mqtt 主题）

AMQP 消费 MQTT 消息：OK AMQP 产生消息到 MQTT：NOT MQTT 到 MQTT：OK AMQP 到 AMQP：OK

我的 mqtt pub/sub 是一个 JS 客户端（经过一些测试）我的 AMQP pub/sub 与 php/symfony

OBS：

如果我去 RabbitMQ，在 amq.topic 交换，.topic.test 队列中发送消息，mqtt 客户端接收正确

问题出在php客户端

似乎这在其他语言和库中不会发生：http: //blog.airasoul.io/the-internet-of-things-with-rabbitmq-node-js-mqtt-and-amqp/

我正在使用vladimir-yuldashev/laravel-queue-rabbitmq库在 Lumen 项目中使用 RabbitMq 队列。

队列功能运行良好，但我在日志文件中看到大量以下错误。

lumen.ERROR：PhpAmqpLib\Exception\AMQPRuntimeException：通道连接已关闭。在 /var/www/html/vendor/php-amqplib/php-amqplib/PhpAmqpLib/Channel/AbstractChannel.php:227

从错误堆栈跟踪来看，队列名称似乎被视为“NULL”。这是我来自queue.php的 rabbitmq 连接配置

我没有使用默认队列。相反，我的每个事件侦听器都为自己声明了一个队列。这是我如何使用队列命令来启动工作程序和侦听器。

工人

听众

这些队列功能中的每一个都运行良好。

我的问题是：

1. 可以为多个听众只启动一个工作人员吗？
2. 为什么我的日志充满了这些错误？我该如何解决这个问题？

还有一点：不确定是否重要，我的事件是连锁的。即，我正在从事件 1 的侦听器中触发事件 2，依此类推。

我正在使用php-amqplib与 PHP 和 RabbitMQ 进行客户端-服务器通信。

我有一个看起来工作正常的生产者脚本，但我的消费者没有收到任何东西。

我检查了队列中的条目，sudo rabbitmqctl list_queues每次运行生产者之后，计数器都会增加。

我的消费者开始时没有任何 PHP 错误，看起来像是在等待消息。看起来不好的地方是，它在启动时运行一次回调脚本，传入的消息是空的——然后什么都不做。

这是我的代码：

生产者.php

（出于测试目的，将主机名常量更改为 localhost）

消费者.php

为什么它不起作用？我发现 rabbitmq/php-amqplib 教程和文档非常无用，并且已经完全解决了这个问题半天多。任何帮助表示赞赏。

更新 1

我还从这个站点检查了这个QA，我的代码与之一致。

如果处理结果不适合我，我如何将消息返回到队列。仅找到有关消息确认的信息，但我认为它不适合我。我需要，如果作为处理的结果，我将参数 RETRY 消息添加回队列。然后这个工人或另一个工人再次拿起它并尝试处理它。

例如：

假设，在我的 php 脚本中，我使用 php-amqplib 客户端打开了到 rabbitmq 的连接。发布消息后，我没有明确关闭通道和连接。

我在 rabbitmq 配置中没有看到任何超时连接。连接会发生什么？它是否一直开放？

什么是最佳实践，每秒通过 php 从队列中接收数据？我使用 ajax 查询来执行此操作，每秒调用一次 php 脚本。在那里，每次都会创建一个连接对象并声明一个队列。我第一次尝试将其保存在会话变量中，但是当我第二次调用 PHP 脚本时，我无法接收更多数据。当我调试通道对象时，我看到 is_open 是假的：

这是我的基本 php 测试代码：

当我每次调用脚本时初始化连接和通道时，它就会起作用。

系统布局

我们有三个系统：

1. API 端点（发布者和消费者）
2. RabbitMQ 服务器
3. 主要应用程序/处理器（发布者和消费者）

系统 1 和 3 都使用 Laravel，并使用 PHPAMQPLIB 与 RabbitMQ 交互。

消息的路径

系统 1（API 端点）向 RabbitMQ 服务器发送序列化作业，供系统 3 处理。然后它立即声明一个新的随机命名的队列，将一个交换绑定到具有相关 ID 的该队列 - 并开始侦听消息。

同时，系统 3 完成了该工作，一旦完成，就将该工作的详细信息返回给交易所的 RabbitMQ，并带有相关 ID。

问题和我尝试过的

我经常发现这个过程失败了。作业被发送和接收，响应被发送 - 但系统 1 从未读取此响应，我没有看到它在 RabbitMQ 中发布。

我已经对此进行了一些广泛的调试，但没有找到根本原因。我目前的理论是，系统 3 返回响应的速度非常快，以至于系统 1 甚至还没有声明新的队列和交换绑定。这意味着系统 3 的响应无处可去，结果消失了. 该理论主要基于这样一个事实，即如果我在 System 3 上将作业设置为以较低频率处理，则系统变得更加可靠。作业处理越快，它就越不可靠。

问题是：我怎样才能防止这种情况发生？还是我还缺少其他东西？我当然希望在不破坏请求/响应模式的情况下快速有效地处理这些工作。

我已经记录了两个系统的输出 - 两者都使用相同的相关 ID，并且系统 3 在发布时获得 ACK - 而系统 1 有一个声明的队列，没有最终超时的消息。

代码示例 1：发布消息

代码示例 2：等待消息响应

感谢您提供的任何建议！

我正在为 RabbitMQ 发布/消费消息使用php-amqplib库。

我做了一些研究，但没有找到一种方法来列出分配给 RabbiMQ 交换的所有路由键（绑定）。

管理 API 可以列出绑定，但我更喜欢 PHP 方式。

如何使用 php-amqplib 获取 RabbitMQ 交换的绑定（路由密钥）？