我的 php amqp 消费者遇到了一个小问题,它在一段时间后停止工作。下面你可以看到我的 silex 命令。我还尝试使用心跳和保活配置来处理断开的网络连接,但它没有改变。消费者不从队列中读取消息的原因可能是什么?脚本并没有退出,它只是似乎在睡觉。
<?php
use Symfony\Component\Console\Input\InputInterface;
use Symfony\Component\Console\Output\OutputInterface;
use Knp\Command\Command as BaseCommand;
use PhpAmqpLib\Message\AMQPMessage;
class RequestWorkerCommand extends BaseCommand
{
protected function configure()
{
$this->setName('queue:worker');
}
protected function execute(InputInterface $input, OutputInterface $output)
{
$app = $this->getSilexApplication();
$amqp = $app['amqp.connection']; /* @var $amqp \PhpAmqpLib\Connection\AMQPStreamConnection */
$channel = $amqp->channel();
$callback = function($message) use ($input, $output) {
return call_user_func_array([$this, 'processMessage'], [$message, $input, $output]);
};
$channel->queue_declare('myqueue', false, true, false, false);
$channel->basic_qos(null, 1, null);
$channel->basic_consume('myqueue', '', false, false, false, false, $callback);
while(count($channel->callbacks)) {
$output->writeln('Waiting for incoming price requests');
$channel->wait();
}
}
protected function processMessage(AMQPMessage $message, InputInterface $input, OutputInterface $output)
{
$app = $this->getSilexApplication();
try {
$data = json_decode($message->body, true);
$request = Request::createFromArray($data); /* create object from data */
$message->delivery_info['channel']->basic_ack($message->delivery_info['delivery_tag']);
$app['distributor']->distribute($request); /* process message */
} catch (\Exception $e) { /* handle error */ }
}
}
我不能确定 PHP,但我在 Python/kombu 上遇到了类似的问题。纯粹的 puython amqplib 从来没有做过心跳,尽管我给了它这样做的指令。当我转而使用 librabbitmq(它包含rabbitmq-c)作为替代品时,心跳不再是一个问题,我的消费者也不再挂断我的电话。
我在 phpamqp-lib 中遇到同样的问题,如何解决?