dtnbjjq51949 2016-04-08 14:58 采纳率: 0%
浏览 841

PHP AMQP消费者在一段时间后没有回应

我遇到了一个 php amqp consumer的小问题,它最近罢工了,在下面你可以看到我的 silex 命令。 我还尝试使用 heartbeat 和 keepalive 配置来处理断开的网络连接,但不奏效。Consumer不从队列中读取消息的原因是什么?脚本也没有删除,但就是不起作用。

<?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 */ }
    }
}
  • 写回答

1条回答 默认 最新

  • douduidui1046 2016-04-08 15:49
    关注

    I cannot be sure of PHP but I faced a similar problem with Python/kombu. the pure puython amqplib never did heartbeats, even though I gave it directives to do so. When I switched to using librabbitmq (which wraps around rabbitmq-c) as a replacement, heartbeats stopped being a problem and my consumers stopped hanging up on me.

    评论

报告相同问题?

悬赏问题

  • ¥15 arduino控制ps2手柄一直报错
  • ¥15 使用rabbitMQ 消息队列作为url源进行多线程爬取时,总有几个url没有处理的问题。
  • ¥85 maple软件,solve求反函数,出现rootof怎么办?
  • ¥15 求chat4.0解答一道线性规划题,用lingo编程运行,第一问要求写出数学模型和lingo语言编程模型,第二问第三问解答就行,我的ddl要到了谁来求了
  • ¥15 Ubuntu在安装序列比对软件STAR时出现报错如何解决
  • ¥50 树莓派安卓APK系统签名
  • ¥65 汇编语言除法溢出问题
  • ¥15 Visual Studio问题
  • ¥20 求一个html代码,有偿
  • ¥100 关于使用MATLAB中copularnd函数的问题