php rabbitmq消费者重新连接

Question

我有一个使用 RabbitMQ 的 PHP 应用程序。为了冗余，我创建了一对 RabbitMQ 服务器并将它们加入到一个集群中。我还有一个运行 HAProxy 的 VyOS 故障转移集群来负载平衡连接并在发生故障转移时提供重新路由。

昨天，我们的 VyOS 集群决定需要进行故障转移(可能是短暂的网络中断)。 HAproxy 在虚拟 IP 移动的一个 VyOS 上停止，并在另一个节点上重新启动 HAproxy。

之后，我查看了 Rabbit 中的队列，发现每个队列的消费者为零。我检查了运行消费者的机器仍然在运行 PHP。我离开他们一段时间，看看他们是否会重新连接，但他们没有。我不得不终止 PHP 脚本并重新启动它们，然后它们重新连接并立即开始使用。

我认为 RabbitMQ 和 HAproxy 正在按预期工作...现在我需要 PHP 消费者支持故障转移事件...换句话说，它不仅需要挂起，还需要检测断开连接并自动重新连接。

这是我的 RabbitMQ 类。提前感谢您的帮助!

<?php
while(true)
{
    try{getMessages("transcode2");}
    catch(Exception $e){echo($e->getMessage()."\n");}
    sleep(1);
}
require_once("../api/db.php");
require_once("../vendor/autoload.php");
use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;
function sendMessage($msg,$prio)
{
    global $channel;
    $msg=json_encode($msg);
    $queue="transcode2";
    $channel->queue_declare($queue,true,false,false,false);
    $channel->basic_publish(new AMQPMessage($msg,array('priority' => $prio)),'',$queue);
}
function getMessages($queue)
{
    global $connection,$channel;
    $connection=new AMQPStreamConnection(RABBITMQ_SERVER,RABBITMQ_PORT,RABBITMQ_USERNAME,RABBITMQ_PASSWORD);
    $channel=$connection->channel();
    $channel->queue_declare($queue,true,false,false,false);
    $callback=function($msg)
    {
        if(handleMessage(json_decode($msg->body,true)))
        {
            $msg->delivery_info['channel']->basic_ack($msg->delivery_info['delivery_tag']);
        }
        else
        {
            $msg->delivery_info['channel']->basic_nack($msg->delivery_info['delivery_tag'],false,true);
        }
    };
    $channel->basic_qos(null,1,null);
    $channel->basic_consume($queue,'',false,false,false,false,$callback);
    while(count($channel->callbacks))
    {
        try{$channel->wait();}
        catch(Exception $e)
        {
            break;
        }
    }
    $channel->close();
    $connection->close();
}
?>

Answer 1

如果您将超时参数用于$channel->wait();，它可能会起作用