Что означает значение $channel->wait() в RabbitMQ

Я совершенно новый в RabbitMQ. Я работаю с библиотекой php-amqplib с codeigniter и до сих пор интересуюсь некоторыми знаниями, которых мне не хватает.

  • Зачем $channel->wait() используется?
  • Почему он всегда находится внутри бесконечного цикла while?
  • Как / можно обойти цикл Infinite while.

Как и в ситуации, когда один пользователь моего проекта хочет транслировать новую кампанию на 100 тыс. Лидов, второй пользователь вступает в действие, если ему нужно отправить около 100 писем, второму приходится ждать доставки 100 тыс. Писем, а затем последнего пользователя. получает свою очередь.

Мне нужно решение для одновременных потребителей, которые работают гладко, не влияя на других

Вот мой фрагмент кода:

public function campaign2(){
        $this->load->library('mylibrary');
        for( $i=1;$i<=5;$i++ ) {
            $url = "http://localhost/myproject/rabbit/waiting";
            $param = array('index' => $i);
            $this->waiting($i);
        }          
}

public function waiting($i)
    {
        ini_set('memory_limit','400M');
        ini_set('max_execution_time', 0);
        ini_set('display_errors', 1);

        ${'conn_'.$i} = connectRabbit();
        ${'channel_'.$i} = ${'conn_'.$i}->channel();
        ${'channel_'.$i}->exchange_declare('ha-local-campaign-'.$i.'-exchange', 'fanout', false, true, false);
        $q    = populateQueueName('campaign-'.$i);
        ${'channel_'.$i}->queue_declare($q, false, true, false, false); 
        ${'channel_'.$i}->queue_bind($q, 'ha-local-campaign-'.$i.'-exchange', 'priority.'.$i);
        $consumer_tag = 'campaign_consumer' ;
        function process_message($msg) {
            echo 'Mail Sent';
            $msg->delivery_info['channel']->basic_ack($msg->delivery_info['delivery_tag']);
        }
        function shutdown($channel, $conn){
            echo '['.date('H:i:s').'] Campaign consumer - Shutdown!!'; 
        }

        ${'channel_'.$i}->basic_consume($q, $consumer_tag, false, false, true, false,'process_message');
        while(1) {
           ${'channel_'.$i}->wait();
        }
        register_shutdown_function('shutdown', ${'channel_'.$i}, ${'conn_'.$i});  
    }

Если кто-нибудь любезно проведет меня через этот процесс, я буду благодарен.

1 ответ

Когда вы звоните $channel->wait() ты:

  • Осмотр очередей канала на наличие ожидающих сообщений.

  • Для каждого сообщения вы будете вызывать зарегистрированный обратный вызов для обратного вызова соответствующего канала.

Из "Привет мир пример", шаг за шагом:

// First, you define `$callback` as a function receiving
// one parameter (the _message_).
$callback = function($msg) {
  echo " [x] Received ", $msg->body, "\n";
};

// Then, you assign `$callback` the the "hello" queue.
$channel->basic_consume('hello', '', false, true, false, false, $callback);

// Finally: While I have any callbacks defined for the channel, 
while(count($channel->callbacks)) {
    // inspect the queue and call the corresponding callbacks
    //passing the message as a parameter
    $channel->wait();
}
// This is an infinite loop: if there are any callbacks,
// it'll run forever unless you interrupt script's execution.

Попросите вашего второго пользователя использовать другую очередь. Вы можете иметь столько очередей, сколько хотите.

Другие вопросы по тегам