3

我想从工作人员的回调中记录工作状态，并在左侧队列中包含一些消息。

到目前为止，我发现的唯一解决方案是获取queue_declare结果数组的第二个成员，但这应该在每次工作人员启动时调用一次，并且我需要更新每条新消息的信息。

UPD：基于IMSoP 答案的解决方案：

<?php
require_once __DIR__ . '/../vendor/autoload.php';
use PhpAmqpLib\Connection\AMQPStreamConnection;
$connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
$channel = $connection->channel();
$channel->queue_declare('test1');
echo "[*] Waiting for messages. To exit press CTRL+C\n";
$callback = function ($msg) use ($channel) {
    list (, $cn) = $channel->queue_declare('test1', true);
    echo ' [x] Received ', $msg->body, " $cn left";
    for ($i = 0; $i < $msg->body; ++$i) {
        sleep(1);
        echo '.';
    }
    echo "\n";
};
$channel->basic_qos(null, 1, null);
$channel->basic_consume('test1', '', false, true, false, false, $callback);
while (count($channel->callbacks)) {
    $channel->wait();
}

出于某种原因，总是给出 0 作为消息计数。

4

1 回答 1

4

该queue_declare方法有一个称为“被动”的参数，可用于此目的：它仅按名称检查队列是否存在，并忽略任何其他参数。

根据AMQP 文档：

如果设置，如果队列已经存在同名，服务器将回复 Declare-Ok，如果不存在则引发错误。客户端可以使用它来检查队列是否存在，而无需修改服务器状态。设置后，除了 name 和 no-wait 之外的所有其他方法字段都将被忽略。带有被动和无等待的声明无效。比较参数的语义等价性。

请注意，这Declare-Ok不仅仅是一个状态，而是完整响应结构的名称，包括字段queue、message-count和consumer-count。

在 PHP-AMQPLib 中，您可以使用它来记录一组队列的状态，如下所示：

foreach ( $this->registeredQueues as $queueName ) {
    // The second parameter to queue_declare is $passive
    // When set to true, everything else is ignored, so need not be passed
    list($queueName, $messageCount, $consumerCount)
        = $this->rabbitChannel->queue_declare($queueName, true);

    $this->logger->info(
        "Queue $queueName has $messageCount messages and $consumerCount active consumers."
    );
}
于 2019-04-15T09:31:04.147 回答