Rabbitmq-php amqp破裂的管道错误

Question

我正在处理一个巨大的xml文档（包含大约一百万个条目），然后使用rabbitmq将一个格式化版本导入数据库。每次发布大约200000个条目后，我都会收到一个断管错误，rabbitmq无法从中恢复。

通知错误：f写（）：发送2651字节失败，errno=11资源暂时不可用在[/var/www/ribbon/app/控制台/命令/lib/php_amqplib/amqp.inc，第439行]

注意错误：fwrite（）：发送33字节失败，错误号为104，对等方在[/var/www/ribbon/app/Console/Command/lib/php_amqplib/amqp.inc，第439行]中重置连接

通知错误：f写（）：发送19个字节失败，errno=32在[/var/www/ribbon/app/控制台/命令/lib/php_amqplib/amqp.inc，第439行]

这随后会导致节点关闭错误，需要手动杀死进程才能从中恢复。

以下是我的课堂方法：-

public function publishMessage($message) {
    if (!isset($this->conn)) {
        $this->_createNewConnectionAndChannel();
    }
    try {
        $this->ch->basic_publish(
            new AMQPMessage($message, array('content_type' => 'text/plain')), 
            $this->defaults['exchange']['name'], 
            $this->defaults['binding']['routing_key']
        );
    } catch (Exception $e) {
        echo "Caught exception : " . $e->getMessage();
        echo "Creating new connection.";
        $this->_createNewConnectionAndChannel();
        $this->publishMessage($message); // try again
    }
}

protected function _createNewConnectionAndChannel() {
    if (isset($this->conn)) {
        $this->conn->close();
    }

    if(isset($this->ch)) {
        $this->ch->close();
    }

    $this->conn = new AMQPConnection(
        $this->defaults['connection']['host'], 
        $this->defaults['connection']['port'], 
        $this->defaults['connection']['user'], 
        $this->defaults['connection']['pass']
    );
    $this->ch = $this->conn->channel();
    $this->ch->access_request($this->defaults['channel']['vhost'], false, false, true, true);
    $this->ch->basic_qos(0 , 20 , 0); // fair dispatching

    $this->ch->queue_declare(
        $this->defaults['queue']['name'],
        $this->defaults['queue']['passive'],
        $this->defaults['queue']['durable'],
        $this->defaults['queue']['exclusive'],
        $this->defaults['queue']['auto_delete']
    );

    $this->ch->exchange_declare(
        $this->defaults['exchange']['name'],
        $this->defaults['exchange']['type'],
        $this->defaults['exchange']['passive'],
        $this->defaults['exchange']['durable'],
        $this->defaults['exchange']['auto_delete']
    );

    $this->ch->queue_bind(
        $this->defaults['queue']['name'],
        $this->defaults['exchange']['name'],
        $this->defaults['binding']['routing_key']
    );
}

任何帮助都将不胜感激。

Answer 1

这个问题发生在我与RabbitMQ的连接断开时（原因无关紧要，在我的情况下，我故意停止RabbitMQ服务进行一些失败测试），我试图通过关闭旧连接并初始化新连接来重新连接RabbitMQ，但我收到了管道断裂或连接关闭的错误消息。我解决此问题的方法是在我的连接上使用重新连接（）方法：

$channel->reconnect();

Answer 2

实际上，当你的消息中有大量内容时，这个问题就会发生，而你的消费者花费了太多时间来处理一条消息，这就是响应兔子的“确认字符”并尝试消费另一条消息的问题。

例如，当我遇到这个问题时，我会尝试“适应”我的消息，因为它是一个产品工人，每条消息都有一些类似1k的产品id，所以我改为100个产品，效果非常好。

你可以在这里阅读更多关于用心跳检测死TCP连接的信息

Answer 3

确保已在Rabbit MQ上为用户添加virtualhost访问权限。我创建了新用户，但忘记了为默认使用的“/”主机设置访问权限。

您可以通过管理面板（您的主机：15672）完成此操作

P. S.我假设您的RabbitMQ服务正在运行，用户存在，密码正确。