PHP RabbitMQ setTimeout или другой вариант, чтобы прекратить ожидание очереди

Мне необходимо создать простого администратора очередей, чтобы передать номер от отправителя потребителю. Учебник Hello World, предоставленный RabbitMQ, охватывает почти 70 % из них.

Но мне нужно изменить очередь, чтобы не вечно ждать входящих сообщений. Или перестаньте ждать после определенного количества сообщений. Я прочитал и попробовал несколько решений из другого поста, но это не работает.

Rabbitmq AMQP::потреблять() - неопределенный метод. есть еще один метод, wait_frame, но он защищенный.

И другой пост на python, который я не понимаю.

<?php

require_once __DIR__ . '/vendor/autoload.php';
require 'config.php';
use PhpAmqpLib\Connection\AMQPStreamConnection;

function recieveQueue($queueName){
    $connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');

    // try{
    //  $connection->wait_frame(10);
    // }catch(AMQPConnectionException $e){
    //  echo "asdasd";
    // }

    $channel = $connection->channel();

    $channel->queue_declare($queueName, false, false, false, false);

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

    $callback = function($msg) {
        echo " [x] Received ", $msg->body, "\n";

    };

    // $tag = uniqid() . microtime(true);
    // $queue->consume($callback, $flags, $tag);

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

    // $channel->cancel($tag);

    while(count($channel->callbacks)) {
        $channel->wait();
    }

    echo "\nfinish";
}

recieveQueue('vtiger');

?>
Author: Community, 2015-11-26
Source

3 answers

Изменить ожидание() в цикле while:

$timeout = 55;
while(count($channel->callbacks)) {
    $channel->wait(null, false, $timeout);
}
 8
Author: mnv, 2015-11-26 05:36:55

Функция ожидания работает только с сокетами, мы должны поймать исключение:

 $timeout = 5;
    while (count($channel->callbacks)) {
        try{
            $channel->wait(null, false , $timeout);
        }catch(\PhpAmqpLib\Exception\AMQPTimeoutException $e){
            $channel->close();
            $connection->close();
            exit;
        }
    }
 6
Author: Maurizio Brioschi, 2016-06-01 11:54:19

Вот как я сделал, чтобы дать сигнал очереди прекратить потребление входящих сообщений.

Однако это может быть неправильным способом сделать это, так как он выдает ошибку, а не выходит должным образом.

Пожалуйста, предложите лучший ответ, если он есть.

    $callback = function($msg) {
        echo " [x] Received ", $msg->body, "\n";

        // if queue recieve 'stop', stop consume anymore messages
        if ($msg->body == 'stop'){
            $channel->basic_cancel($queueName);
        }
    };

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

    $timeout = 10;
    while(count($channel->callbacks)) {
        // $channel->wait(null, false, $timeout);
        $channel->wait();
    }
 0
Author: jedi, 2016-09-27 09:19:51