如何使用php-amqplib关闭来自消费者回调的rabbitmq连接

Question

我已经能够使用php-amqplib库设置一个连接到RabbitMQ的脚本。但是，如果需要的话，我希望能够退出脚本。

我们认为创建一个stop.txt文件就可以解决这个问题。基本上，一旦接收到消息，如果脚本看到该文件，它必须停止接收任何进一步的消息，删除该文件并退出脚本。

但是，尽管它正在删除该文件，但它不会阻止接收进一步的消息，也不会退出脚本。

正如您所看到的，我使用与此官方示例完全相同的函数https://github.com/php-amqplib/php-amqplib/blob/master/demo/amqp_consumer.php来退出脚本。参见第54行。

我是不是做错了什么或者不理解什么？我是这个图书馆的新手。

这是我正在使用的脚本。为了简短起见，我省略了部分内容。

function process_message($message)
{
    // ... do some processing ... //
    // ... insert data into DB ... //

    // send an acknowledgement back to the producer: message successfully received
    $message->ack();
    
    // if a stop file exists then exit the script
    if (file_exists("./stop.txt")) {
        $message->getChannel()->basic_cancel($message->getConsumerTag());
        unlink("./stop.txt"); // delete stop file
    }
}

function shutdown($channel, $connection)
{
    $channel->close();
    $connection->close();
}

$connection = new AMQPStreamConnection(HOST, PORT, USERNAME, PASSWORD, VHOST);
$channel = $connection->channel();

$channel->queue_declare(QUEUE, false, true, false, false);

// make sure only one message is sent at a time
$channel->basic_qos(null, 1, null);

// consume messages from the queue
$channel->basic_consume(QUEUE, 'consumer', false, false, false, false, 'process_message');

register_shutdown_function('shutdown', $channel, $connection);

// Loop as long as the channel has callbacks registered
while ($channel->is_consuming()) {
    $channel->wait();
}

提前谢谢你，

理查德

Answer 1

这似乎关闭了一切。我不知道这是不是最好的解决方案。我基本上是从回调中关闭通道和连接。

// if a stop file exists then exit the script
if (file_exists("./stop.txt")) {
    $message->getChannel()->basic_cancel($message->getConsumerTag());
    // NEW CODE //
    $channel = $message->getChannel();
    $connection = $message->getChannel()->getConnection();
    $channel->close();
    $connection->close();
    // END NEW CODE //
    unlink("./stop.txt"); // delete stop file
}