如何使用php-amqplib在消费者回调中关闭RabbitMQ连接并退出脚本

你的代码没有语法错误，问题出在对php-amqplib的阻塞等待机制理解有偏差：

• $channel->wait() 方法默认是永久阻塞状态，只有收到RabbitMQ服务端发来的AMQP帧（比如新消息、确认帧等）时才会返回
• 你在回调里调用basic_cancel只是向服务端发送了取消消费者的请求，服务端返回cancel_ok帧之后is_consuming()才会变为false
• 如果调用basic_cancel之后没有新的消息推送，wait()会一直阻塞，永远不会进入下一次循环判断is_consuming()的状态，所以脚本不会退出

解决方案 #

可以通过添加退出标记+设置wait超时的方式实现优雅退出，修改后的代码如下：

// 新增全局退出标记
$should_exit = false;

function process_message($message)
{
    global $should_exit;
    // ... 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
    // 建议用__DIR__生成绝对路径，避免CLI模式下工作目录不一致导致检测失效
    if (file_exists(__DIR__ . "/stop.txt")) {
        $message->getChannel()->basic_cancel($message->getConsumerTag());
        unlink(__DIR__ . "/stop.txt");
        // 标记为需要退出
        $should_exit = true;
    }
}

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);

// 循环条件增加退出标记判断，wait设置0.5秒超时（单位：微秒）
while ($channel->is_consuming() && !$should_exit) {
    $channel->wait(null, false, 500000);
}

说明 #

• 第三个参数500000表示wait最多阻塞0.5秒就会主动返回，触发下一次循环条件判断
• 只要$should_exit被标记为true，循环就会终止，脚本会自动执行注册的shutdown方法关闭连接后退出
• 如果你的业务允许更长的退出延迟，可以调大超时参数，反之可以调小
• 如果不需要做额外的收尾处理，检测到stop.txt之后直接调用exit()也可以正常退出，你已经对当前消息做了ack，register_shutdown_function会自动完成连接清理工作，不会产生异常

内容的提问来源于stack exchange，提问作者Richard Slabbert