我有一个Web服务接口抽象RabbitMQ服务器(不要问我为什么,我知道这是一个不必要的步骤,但我必须)。也就是说,我通过Web服务调用从队列中轮询消息,而不是直接通过amqp
。
通过basic.consumer
进行消耗会阻塞执行线程,直到队列中有消息为止。这使得Web服务无法返回。
插图代码:
$connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
$channel = $connection->channel();
$channel->queue_declare(QUEUE_NAME, false, true, false, false);
$ret = array('body' => '');
$callback = function($msg) use ($channel, &$ret) {
$ret['body'] = $msg->body;
/*
Here I would basic.cancel the consumer if there were no messages in the queue
*/
};
$channel->basic_consume(QUEUE_NAME, 'tag', false, true, false, false, $callback);
if (count($channel->callbacks)) {
$channel->wait(); // blocks here...
}
return $ret;
如果你想得到队列的大小,可以用php-amqlib调用queue_declare
,return的第二个参数是队列中的消息数。
list($queue, $messageCount, $consumerCount) = $channel->queue_declare(QUEUE_NAME, true);
调用queue_declare()方法时,将$ passive参数设置为true非常重要
我想做的是由basic.get
实现的。
在php-amqplib中:
$channel->basic_get(QUEUE_NAME, true); // the second arg is no_ack
。
第二个参数表示不希望对该消息进行确认。也就是说,您不必将该消息“标记”为RabbitMQ读取的信息,以便自信地将其出列。排除它(具有it = false)导致不弹出顶部消息。
我将RabbitMQ代码包装在http Web服务中。这不是一个好主意(至少对我的用例而言)。当Web服务返回时,因此rabbitmq连接终止,非(尚)确认的消息被重新排队回队列。因此,如果必须采用http包装器,请确保将rabbitmq连接的生命周期与http请求lieftime隔离开来。但是,我没试过这个。