PHP如何在RabbitMQ中取消消费者？

Question

我有下一个代码：

<?php

// callback function for recive the message and canceling consumer
function consumer(\AMQPEnvelope $envelope, \AMQPQueue $queue)
{
    $queue->ack($envelope->getDeliveryTag());
    $queue->cancel($envelope->getCorrelationId());
    echo "Message was recived and consumer will be canceled by consumer tag: {$envelope->getCorrelationId()}\n";
}

// generating uniqie exchange and queue
$correlationId = uniqid(str_replace('.', '', (string)microtime(TRUE)) . '_');
$queueName = "databus_response_{$correlationId}";
$consumerTag = "consumer_tag_{$correlationId}";

// establesh connection
$connection = new \AMQPConnection(array('host'=>'127.0.0.1', 'user'=>'guest', 'password'=>'guest'));
$connection->connect();
$channel = new \AMQPChannel($connection);

// declare exchange
$exchange = new \AMQPExchange($channel);
$exchange->setFlags(AMQP_AUTODELETE);
$exchange->setType(AMQP_EX_TYPE_TOPIC);
$exchange->setName($queueName);
$exchange->declareExchange();

// declare queue
$queue = new \AMQPQueue($channel);
$queue->setFlags(AMQP_EXCLUSIVE);
$queue->setName($queueName);
$queue->declareQueue();
$queue->bind($queueName, '#');

// publish message in exchange
$exchange->publish('Test message', NULL, AMQP_PASSIVE, array('correlation_id' => $consumerTag));

// run consumer for getting this echange and canceling consumer after recive the message
$queue->consume('consumer', AMQP_NOWAIT, $consumerTag);

你怎么看，我发送一条消息到队列，并在这个队列上运行消费者。在消费者方法中，您可以看到，在通过“取消”方法接收第一条消息后，我想在此队列上停止消费者，但消费者并未停止。我做错了什么？

• AMQP PECL 模块版本 1.2.0 来自https://github.com/pdezwart/php-amqp/tree/v1.2.0
• PHP 5.4.4-14+deb7u10 (cli)
• Linux v270 3.2.0-4-amd64 #1 SMP Debian 3.2.54-2 x86_64 GNU/Linux

Answer 1

false当您想停止消费时，只需从消费者回调中返回。

从AMQPQueue.php存根文件：

在回调函数返回 FALSE 之前，AMQPQueue::consume() 不会将处理线程返回给 PHP 脚本。

PS：伤心地说，php-amqp扩展仍然缺乏良好的文档，但你可以随时阅读方法的注释中的存根文件或读到这里在波兰语官方PHP网站有点过时文档- http://www.php.net/manual/ pl/book.amqp.php（别担心，没有人翻译它，所以所有敏感的文档都是英文的）。