2

Symfony 信使：

https://symfony.com/doc/current/messenger.html

问题：

Pool#1 =（user1创建一个Job，Job拆分为 10 个信使Message）
Pool#2 =（user2创建一个Job，Job拆分为 10 个信使Message）
...
Pool#100 =（user100创建一个Job，Job拆分为 10 个信使Message）

Pool#100 将不会被执行，直到所有先前的 Pool 都不会完成。

目标：

我需要并行队列，所有池都将单独运行，因此每个池都有个人队列。

代码示例：

配置/包/messenger.yaml 
framework:
    messenger:
        transports:
            sync: "%env(MESSENGER_TRANSPORT_DSN)%"
        routing:
            'App\Message\Job': sync
src/消息/Job.php 
<?php

namespace App\Message;

class Job
{
    private $content;

    public function __construct(string $content)
    {
        $this->content = $content;
    }

    public function getContent(): string
    {
        return $this->content;
    }
}
src/MessageHandler/JobHandler.php 
<?php

namespace App\MessageHandler;

use App\Message\Job;
use Symfony\Component\Messenger\Handler\MessageHandlerInterface;

class JobHandler implements MessageHandlerInterface
{
    public function __construct()
    {}

    public function __invoke(Job $message)
    {
        $params = json_decode($message->getContent(), true);
        dump($params);
    }
}
src/Controller/JobController.php 
<?php

namespace App\Controller;

use Symfony\Bundle\FrameworkBundle\Controller\AbstractController;
use Symfony\Component\HttpFoundation\JsonResponse;
use Symfony\Component\HttpFoundation\Request;
use Symfony\Component\Messenger\MessageBusInterface;
use Symfony\Component\Routing\Annotation\Route;

/**
 * @Route("/job")
 */
class JobController extends AbstractController
{
    /**
     * @Route("/create", name="app_job_create")
     * @param Request $request
     * @param MessageBusInterface $bus
     * @return JsonResponse
     */
    public function create(Request $request, MessageBusInterface $bus): JsonResponse
    {
        // ...
        $entityId = $entity->getId();
        // ...

        for ($i = 0; $i < 10; $i++) {
            $params['entityId'] = $entityId;
            $params['counter'] = $i;
            $bus->dispatch(new Job(json_encode($params)));
        }

        return new JsonResponse([]);
    }
}

更多信息：

我想继续使用它，但找不到最简单的解决方案来传递一些唯一的队列名称或 ID，然后告诉工人他必须只处理这个 Pool of Messages.
我找到了自定义传输https://symfony.com/doc/current/messenger/custom-transport.html，但我不确定它是否有帮助。至少我认为只有定制运输是不够的。
我阅读了有关Actor models https://www.brianstorti.com/the-actor-model/的信息，但如果可能的话，我只想使用 Messenger+Redis。

可能这里没有解决方案，这个信使还不能处理并行队列。无论如何，我很高兴得到任何帮助。
谢谢！

4

1 回答 1

1

我最终通过使用动态队列名称解决了这个问题。
不幸的是，我被迫拒绝了信使。

此外，symfony 5（2020 年 6 月）目前不支持 RabbitMQ 的 SDK https://github.com/php-amqplib/RabbitMqBundle，我不确定 100%，但我在第 3 版中使用了它，但是我不能把它放在5号。所以我用了另一个。

这是一个很好的快速入门指南https://blog.programster.org/rabbitmq-job-queue-with-php
在示例中，我将 RABBITMQ_QUEUE_NAME 更改为动态名称，并且效果很好。

然后一切照常，提升 RabbitMQ 并配置主管https://symfony.com/doc/current/messenger.html#supervisor-configuration

如果这对某人有帮助，我会很高兴，谢谢！

于 2020-07-10T08:39:54.737 回答