2

Вопрос в следующем. Пытаюсь разобраться с работой компонента Messenger в Symfony. Есть следующие классы:

ACSTask.php

<?php

namespace App\Message;

class ACSTask
{
    private $deviceId;
    private $hostToCheck;
    private $operation;

    public function __construct(?int $deviceId, ?string $hostToCheck, ?string $operation)
    {
        $this->deviceId = $deviceId;
        $this->hostToCheck = $hostToCheck;
        $this->operation = $operation;
    }

    public function getDeviceId(): ?int
    {
        return $this->deviceId;
    }

    public function getHostToCheck(): ?string
    {
        return $this->hostToCheck;
    }

    public function getOperation(): ?string
    {
        return $this->operation;
    }
}

ACSTaskHandler.php

<?php

namespace App\MessageHandler;

use App\Message\ACSTask;
use Symfony\Component\Messenger\Handler\MessageHandlerInterface;
use Symfony\Component\Process\Process;
use Symfony\Component\Process\Exception\ProcessFailedException;

class ACSTaskHandler implements MessageHandlerInterface
{
    public function __invoke(ACSTask $acsTask)
    {
        echo "Start\n";
        $process = new Process([...]);
        $process->run();
        echo "End\n";
    }
}

Метод контроллера:

    public function testAmqp(MessageBusInterface $bus)
    {
        $device_id = 18;
        $address = "google.ru";
        $bus->dispatch(new Envelope(new ACSTask($device_id, $address, "ping"), [

        ]));
        return new JsonResponse([
            'status' => 'OK',
            'ping' => isset($ping) ? $serializer->serialize($ping, 'json') : null,
        ]);
    }

messenger.yaml

framework:
    messenger:
        transports:
            # Uncomment the following line to enable a transport named "amqp"
            amqp:
                dsn: '%env(MESSENGER_TRANSPORT_DSN)%'

        routing:
            # Route your messages to the transports
            'App\Message\ACSTask': amqp

Данная схема работает, но выполняет все задачи поочередно. Я так понимаю, что дело в том, что все задачи шлются в одну очередь. Мне же нужно сгруппировать задачи и добавить их в очереди, соответствующие группам. В официальной документации RabbitMQ сказано, что это можно сделать, но совсем не понимаю как это реализовать в рамках фреймворка. Publish/Subscribe

Я впервые прибегаю к использованию к брокеру сообщений и возможно неправильно понимаю его работу. Возможно компоненты из коробки в Symfony не позволяют задавать нужные параметры из коробки. Прошу совета и помощи: как мне реализовать в рамках фреймворка вышеописанное или какую библиотеку под Symfony мне лучше использовать?

2 ответа 2

1

Messenger прост и разделение очередей основана на классах

messenger.yaml

framework:
    messenger:
        transports:
            # Uncomment the following line to enable a transport named "amqp"
            runTaskPing: %env(MESSENGER_TRANSPORT_DSN)%/task.ping
            runTaskTrace: %env(MESSENGER_TRANSPORT_DSN)%/task.trace
        routing:
            # Route your messages to the transports
            'App\Message\ACSTaskPing': runTaskPing
            'App\Message\ACSTaskTrace': runTaskTrace

Соответственно для каждого типа сообщений должны быть свои Message и свои Handler. Возможны абстрактный Message и Handler в которых вся логика и почти пустые children которые нужны чисто для роутинга по группам.

Ну и вызов для сообщения будет соответствующий

$bus->dispatch(new ACSTaskPing($device_id, $address, "ping"))
$bus->dispatch(new ACSTaskTrace($device_id, $address, "trace"))
0

Попробуйте использовать RabbitMqBundle или enqueue-bundle. Они оба должны подойти для вашей задачи. Также вы можете использовать приоритет на выполнение в пределах одной очереди.

Ваш ответ

By clicking “Отправить ответ”, you agree to our terms of service and acknowledge you have read our privacy policy.

Всё ещё ищете ответ? Посмотрите другие вопросы с метками или задайте свой вопрос.