1

我正在使用php-amqplib与 PHP 和 RabbitMQ 进行客户端-服务器通信。

我有一个看起来工作正常的生产者脚本，但我的消费者没有收到任何东西。

我检查了队列中的条目，sudo rabbitmqctl list_queues每次运行生产者之后，计数器都会增加。

我的消费者开始时没有任何 PHP 错误，看起来像是在等待消息。看起来不好的地方是，它在启动时运行一次回调脚本，传入的消息是空的——然后什么都不做。

php consumer.php 
string(47) " [*] Waiting for messages. To exit press CTRL+C"
string(1) "
"
string(10) "Received: "

这是我的代码：

生产者.php

public function sendDataToRabbitMQ()
{
    $id = $_POST['id'];
    $ipAddress = $_POST['ip'];
    $date = date("Y-m-d h:i:s");
    $status = false;

    $connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest'); //host: RABBITMQ_HOST
    $channel = $connection->channel();

    $channel->queue_declare('first_queue', false, true, false, false);

    if(is_array($argv)) {
        $data = implode(' ', array_slice($argv, 1));
    }

    if (empty($data)) {
        $data = "$ipAddress,$id";
    }

    $msg = new AMQPMessage($data, array('delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT));

    $channel->basic_publish($msg, '', 'first_queue');

    echo " [x] Sent data:", "\n", $data, "\n";

    $channel->close();
    $connection->close();

    return $result;
}

（出于测试目的，将主机名常量更改为 localhost）

消费者.php

<?php

require_once '../vendor/autoload.php';

use PhpAmqpLib\Connection\AMQPStreamConnection;

class Consumer
{
    private $token;

    private function getQueue()
    {
        $connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
        $channel = $connection->channel();

        $channel->queue_declare('ban_queue', false, true, false, false);

        var_dump(' [*] Waiting for messages. To exit press CTRL+C', "\n");

        $callback = function($msg) {

            $message = explode(',', $msg->body);

            var_dump('Received: '.$message[0]);

            $msg->delivery_info['channel']->basic_ack($msg->delivery_info['delivery_tag']);
        };

        $channel->basic_qos(null, 1, null);
        $channel->basic_consume('first_queue', '', false, false, false, false, $callback);
        while (count($channel->callbacks)) {
            $channel->wait();
        }
        $channel->close();
        $connection->close();
    }

    public function processResult()
    {
        $this->getQueue();
    }

}

$consumer = new Consumer();
$consumer->processResult();

为什么它不起作用？我发现 rabbitmq/php-amqplib 教程和文档非常无用，并且已经完全解决了这个问题半天多。任何帮助表示赞赏。

更新 1

我还从这个站点检查了这个QA，我的代码与之一致。

4

1 回答 1

2

经过一段时间的研究和测试，我找到了解决上述问题的方法：

我变了

$channel->basic_consume('first_queue', '', false, false, false, false, $callback);

到

$channel->basic_consume('first_queue', '', false, true, false, false, $callback);

这就是诀窍。

于 2017-09-01T11:17:22.090 回答