阻止消费者进入空队列

Question

具体说明

我有一个用 PHP 编写的消费者，它试图消费消息。我的目标很简单 - 如果队列中没有消息，则释放执行并继续，考虑到“没有检索到数据”。

目前的想法

我试过AMQP_NOWAIT标志，比如：

$flag = AMQP_NOWAIT;
$this->queue->consume($callbackFunction, $flag, $this->consumerTag);

但它没有用。到目前为止，我有解决方法，比如 - 我声明 \AMQPConnection 的连接超时，比如说 5 秒，然后以这种方式捕获它：

try {
    $this->consumer->consume($this->consumer->getReadMessageCallback($notifications, $requeue));
} catch (\AMQPConnectionException $connectionException) {
    //based on timeouts. Are there other ways to interrupt empty queue consuming? AMQP_NOWAIT fails, does nothing:
    return [];
}

但是，这是一种非常“hacky”的方式。它对我有用，但是：

• 仍然阻塞代码 timeout 秒
• 显然会失败，如果我有太多消息（即在超时结束之前无法完成）。
• 更多，它甚至没有记录，所以依靠它应该是最后的手段。

接下来 - 我在创建队列时尝试了 AMQP_IFEMPTY | AMQP_PASSIVE。问题是 - 如果那里没有消息，它将删除队列，并且在尝试从那里获取消息时会引发异常（我可能会捕获）。但是出现了一个问题，比如队列被立即删除，我什至不能在那里添加消息。

问题

从空队列读取消息确实是一个常见问题，因此我确信它应该是一种适当解决问题的方法。因此，我该怎么做？

是的，手动链接是/pl/，因为没有“en”链接。但无论如何，它或多或少都是英文的。

Answer 1

如果需要判断队列是否为空，可以调用AMQPQueue::declare()，它是幂等的，结果返回队列中的消息数。请注意，这个数字不太准确（请参阅why）。

此外，您可以直接致电AMQPQueue::get()（就像在管理工具中完成的那样）。

毕竟，正如您所尝试的那样，您可以将AMQPConnection::setReadTimeout() 设置为某个较低的值（在本地网络中1 秒可能就足够了），然后调用AMQPQueue::consume() 并在消费者等待太久时捕获超时异常。

关于糟糕的文档，请参阅此问题的答案：Where can i find the php-amqp documentation。