17

好的，这是正在发生的事情的概述：

    M <-- Message with unique id of 1234
    |
    +-Start Queue
    |
    |
    | <-- Exchange
   /|\
  / | \
 /  |  \ <-- bind to multiple queues
Q1  Q2  Q3
\   |   / <-- start of the problem is here
 \  |  / 
  \ | /
   \|/
    |
    Q4 <-- Queues 1,2 and 3 must finish first before Queue 4 can start
    |
    C <-- Consumer

所以我有一个推送到多个队列的交换器，每个队列都有一个任务，一旦所有任务都完成了，只有这样队列 4 才能启动。

因此，唯一 id 为 1234 的消息被发送到交换器，交换器将其路由到所有任务队列（Q1、Q2、Q3 等），当消息 id 为 1234 的所有任务都完成后，运行 Q4 获取消息编号 1234。

我该如何实施？

使用 Symfony2、RabbitMQBundle 和 RabbitMQ 3.x

资源：

更新#1

好的，我认为这就是我要寻找的：

带有并行处理的 RPC，但是如何将 Correlation Id 设置为我的唯一 id 来对消息进行分组并识别什么队列？

4

5 回答 5

7

你需要实现这个：http ://www.eaipatterns.com/Aggregator.html但 Symfony 的 RabbitMQBundle 不支持，所以你必须使用底层的 php-amqplib。

来自捆绑包的普通消费者回调将获得 AMQPMessage。从那里您可以访问通道并手动发布到“管道和过滤器”实施中接下来出现的任何交换

于 2012-12-13T21:50:16.927 回答
5

在 RabbitMQ 站点的RPC教程中，有一种方法可以传递一个“相关 id”，它可以识别您的消息给队列中的用户。

我建议您将某种 id 与您的消息一起使用到前 3 个队列中，然后有另一个过程将消息从 3 个队列中取出到某种类型的存储桶中。当这些桶收到我假设完成的 3 个任务时，将最终消息发送到第 4 个队列进行处理。

如果您为一个用户向每个队列发送超过 1 个工作项，您可能需要进行一些预处理以找出特定用户放入队列中的项目数量，以便在 4 之前出队的进程知道在排队之前预期有多少项目向上。

我在 C# 中做我的 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 客户端，那么您需要做的就是让服务器处理用户唯一 ID 的生成并将消息发送到适当的队列：

    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
于 2012-12-13T14:36:44.927 回答
2

我有点不清楚你想在这里实现什么。但我可能会稍微改变设计，以便一旦所有消息从您发布到发布到队列 4 的单独交换中的队列中清除。

于 2012-12-13T14:30:51.817 回答
2

除了基于 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';
    }
}
于 2017-06-22T12:26:02.600 回答
0

我可以向您展示如何使用enqueue-bundle来做到这一点。

因此，使用 composer 安装它并注册为任何其他包。然后配置：

// 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 版本。

于 2017-06-22T11:51:43.537 回答