RabbitMQ ждет завершения нескольких очередей

Ответ 1

В учебном пособии RPC на сайте RabbitMQ существует способ передать "Идентификатор корреляции", который может идентифицировать ваши сообщения для пользователей в очередь.

Я бы рекомендовал использовать какой-то идентификатор с вашими сообщениями в первые три очереди, а затем выполнить другой процесс для удаления сообщений из 3 в какие-то ведра. Когда эти ведра получают то, что я предполагаю, это завершение трех задач, отправьте окончательное сообщение в 4 очередь для обработки.

Если вы отправляете более одного рабочего элемента в каждую очередь для одного пользователя, вам может потребоваться небольшая предварительная обработка, чтобы узнать, сколько элементов, которые пользователь помещает в очередь, так что процесс, деинсталлирующий до 4, знает, сколько ожидайте перед очередью.

Я делаю свой rabbitmq на С#, поэтому извините, что мой псевдокод не находится в стиле php

// Client
byte[] body = new byte[size];
body[0] = uniqueUserId;
body[1] = howManyWorkItems;
body[2] = command;

// Setup your body here

Queue(body)

// Server
// Process queue 1, 2, 3
Dequeue(message)

switch(message.body[2])
{
    // process however you see fit
}

processedMessages[message.body[0]]++;

if(processedMessages[message.body[0]] == message.body[1])
{
    // Send to queue 4
    Queue(newMessage)
}

Ответ на обновление # 1

Вместо того, чтобы думать о вашем клиенте как о терминале, может быть полезно подумать о клиенте как о процессе на сервере. Поэтому, если вы настроили клиент RPC на сервере, например этот, то все, что вам нужно сделать, это заставить сервер обрабатывать генерирование уникального идентификатора пользователя и отправлять сообщения в соответствующие очереди:

    public function call($uniqueUserId, $workItem) {
        $this->response = null;
        $this->corr_id = uniqid();

        $msg = new AMQPMessage(
            serialize(array($uniqueUserId, $workItem)),
            array('correlation_id' => $this->corr_id,
            'reply_to' => $this->callback_queue)
        );

        $this->channel->basic_publish($msg, '', 'rpc_queue');
        while(!$this->response) {
            $this->channel->wait();
        }

        // We assume that in the response we will get our id back
        return deserialize($this->response);
    }


$rpc = new Rpc();

// Get unique user information and work items here

// Pass even more information in here, like what queue to use or you could even loop over this to send all the work items to the queues they need.
$response = rpc->call($uniqueUserId, $workItem);

$responseBuckets[array[0]]++;

// Just like above code that sees if a bucket is full or not

Ответ 2

Вам нужно реализовать это: http://www.eaipatterns.com/Aggregator.html, но RabbitMQBundle для Symfony не поддерживает это, поэтому вам придется использовать базовый php-amqplib.

Обычный обратный вызов пользователя из пакета получит AMQPMessage. Оттуда вы можете получить доступ к каналу и вручную опубликовать все последующие обмены в реализации ваших "труб и фильтров"

Ответ 3

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

Ответ 4

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

Идея следующая: все асинхронно, не RPC или другие вещи синхронизации. Каждая задача отправляет даже когда это делается, агрегатор подписывается на это событие. Он в основном учитывает задачи и отправляет сообщение task4, когда счетчик достигает ожидаемого числа (в нашем случае 3). Я выбираю файловую систему как хранилище для счетчиков для Простоты. Вы можете использовать там базу данных.

Производитель выглядит проще. Он просто срабатывает и забывает

<?php
use Enqueue\Client\Message;
use Enqueue\Client\ProducerInterface;
use Enqueue\Util\UUID;
use Symfony\Component\DependencyInjection\ContainerInterface;

/** @var ContainerInterface $container */

/** @var ProducerInterface $producer */
$producer = $container->get('enqueue.client.producer');

$message = new Message('the task data');
$message->setCorrelationId(UUID::generate());

$producer->sendCommand('task1', clone $message);
$producer->sendCommand('task2', clone $message);
$producer->sendCommand('task3', clone $message);

Процессор задач должен отправить событие после выполнения задания:

<?php
use Enqueue\Client\CommandSubscriberInterface;
use Enqueue\Client\Message;
use Enqueue\Client\ProducerInterface;
use Enqueue\Psr\PsrContext;
use Enqueue\Psr\PsrMessage;
use Enqueue\Psr\PsrProcessor;

class Task1Processor implements PsrProcessor, CommandSubscriberInterface
{
    private $producer;

    public function __construct(ProducerInterface $producer)
    {
        $this->producer = $producer;
    }

    public function process(PsrMessage $message, PsrContext $context)
    {
        // do the job

        // same for other
        $eventMessage = new Message('the event data');
        $eventMessage->setCorrelationId($message->getCorrelationId());

        $this->producer->sendEvent('task_is_done', $eventMessage);

        return self::ACK;
    }

    public static function getSubscribedCommand()
    {
        return 'task1';
    }
}

И агрегатор-процессор:

<?php

use Enqueue\Client\TopicSubscriberInterface;
use Enqueue\Psr\PsrContext;
use Enqueue\Psr\PsrMessage;
use Enqueue\Psr\PsrProcessor;
use Symfony\Component\Filesystem\LockHandler;

class AggregatorProcessor implements PsrProcessor, TopicSubscriberInterface
{
    private $producer;
    private $rootDir;

    /**
     * @param ProducerInterface $producer
     * @param string $rootDir
     */
    public function __construct(ProducerInterface $producer, $rootDir)
    {
        $this->producer = $producer;
        $this->rootDir = $rootDir;
    }

    public function process(PsrMessage $message, PsrContext $context)
    {
        $expectedNumberOfTasks = 3;

        if (false == $cId = $message->getCorrelationId()) {
            return self::REJECT;
        }

        try {
            $lockHandler = new LockHandler($cId, $this->rootDir.'/var/tasks');
            $lockHandler->lock(true);

            $currentNumberOfProcessedTasks = 0;
            if (file_exists($this->rootDir.'/var/tasks/'.$cId)) {
                $currentNumberOfProcessedTasks = file_get_contents($this->rootDir.'/var/tasks/'.$cId);

                if ($currentNumberOfProcessedTasks +1 == $expectedNumberOfTasks) {
                    unlink($this->rootDir.'/var/tasks/'.$cId);

                    $this->producer->sendCommand('task4', 'the task data');

                    return self::ACK;
                }
            }

            file_put_contents($this->rootDir.'/var/tasks/'.$cId, ++$currentNumberOfProcessedTasks);

            return self::ACK;
        } finally {
            $lockHandler->release();
        }
    }

    public static function getSubscribedTopics()
    {
        return 'task_is_done';
    }
}

Ответ 5

Я могу показать вам, как вы можете это сделать с помощью enqueue-bundle.

Поэтому установите его с композитором и зарегистрируйте как любой другой пакет. Затем настройте:

// app/config/config.yml

enqueue:
  transport:
    default: 'amnqp://'
  client: ~

Этот подход основан на RPC. Вот как вы это делаете:

<?php
use Enqueue\Client\ProducerInterface;
use Symfony\Component\DependencyInjection\ContainerInterface;

/** @var ContainerInterface $container */

/** @var ProducerInterface $producer */
$producer = $container->get('enqueue.client.producer');

$promises = new SplObjectStorage();

$promises->attach($producer->sendCommand('task1', 'the task data', true));
$promises->attach($producer->sendCommand('task2', 'the task data', true));
$promises->attach($producer->sendCommand('task3', 'the task data', true));

while (count($promises)) {
    foreach ($promises as $promise) {
        if ($replyMessage = $promise->receiveNoWait()) {
            // you may want to check the response here
            $promises->detach($promise);
        }
    }
}

$producer->sendCommand('task4', 'the task data');

Потребительский процессор выглядит следующим образом:

use Enqueue\Client\CommandSubscriberInterface;
use Enqueue\Consumption\Result;
use Enqueue\Psr\PsrContext;
use Enqueue\Psr\PsrMessage;
use Enqueue\Psr\PsrProcessor;

class Task1Processor implements PsrProcessor, CommandSubscriberInterface
{
    public function process(PsrMessage $message, PsrContext $context)
    {
        // do task job

        return Result::reply($context->createMessage('the reply data'));
    }

    public static function getSubscribedCommand()
    {
        // you can simply return 'task1'; if you do not need a custom queue, and you are fine to use what enqueue chooses. 

        return [
          'processorName' => 'task1',
          'queueName' => 'Q1',
          'queueNameHardcoded' => true,
          'exclusive' => true,
        ];
    }
}

Добавьте его в свой контейнер в качестве службы с тегом enqueue.client.processor и выполните команду bin/console enqueue:consume --setup-broker -vvv

Здесь простая версия PHP.