Пример использования Yandex Message Queue на Symfony

PHP-фреймворк Symfony содержит компонент Messenger для работы с очередями. В качестве брокера сообщений можно использовать Message Queue.

УстановкаУстановка

Потребуются следующие инструменты:

• Git;
• PHP версии 8.2 или выше;
• Менеджер пакетов Composer.

Установите Symfony по инструкции на официальном сайте фреймворка.

Подготовка к работеПодготовка к работе

1. Создайте сервисный аккаунт.
2. Назначьте роль editor сервисному аккаунту.
3. Создайте статический ключ доступа.

Создайте очередь в сервисе Message Queue и скопируйте ее URL.

ИнструкцииИнструкции

В этом примере создаются:

• демонстрационное сообщение (Message), в котором хранятся исходные числа;
• обработчик сообщения (MessageHandler), который суммирует два числа из сообщения;
• команда (Command), которая ставит задачу в очередь Message Queue.

Чтобы использовать Message Queue с Symfony Messenger, выполните следующие инструкции:

1. Создайте тестовый проект mq_example:

   symfony new --webapp mq_example
   

2. Установите зависимости для работы с Amazon SQS. Message Queue использует формат совместимый с SQS:

   composer require symfony/amazon-sqs-messenger async-aws/sqs ^1.9
   

3. Создайте сообщение (Message) и обработчик (Handler):

   php bin/console make:message Sum
   

   При выполнении команда спросит Which transport do you want to route your message to? [[no transport]].
   Впишите цифру с вариантом async.

4. Создайте команду (Command) для отправки сообщений в очередь:

   php bin/console make:command app:create
   

5. Откройте созданный файл по пути src/Command/SumCommand.php и приведите его к виду:

   <?php
   
   namespace App\Command;
   
   use App\Message\Sum;
   use Symfony\Component\Console\Attribute\AsCommand;
   use Symfony\Component\Console\Command\Command;
   use Symfony\Component\Console\Input\InputInterface;
   use Symfony\Component\Console\Output\OutputInterface;
   use Symfony\Component\Messenger\MessageBusInterface;
   
   #[AsCommand(
       name: 'app:create',
       description: 'Add a short description for your command',
   )]
   class CreateCommand extends Command
   {
       public function __construct(private readonly MessageBusInterface $messageBus)
       {
           parent::__construct();
       }
   
       protected function execute(InputInterface $input, OutputInterface $output): int
       {
           $this->messageBus->dispatch(new Sum(4,2));
           
           return Command::SUCCESS;
       }
   }
   

6. Откройте файл по пути src\Message\Sum.php и приведите его к виду:

   <?php
   
   namespace App\Message;
   
   final class Sum
   {
       public function __construct(private readonly int $a, private readonly int $b)
       {
       }
   
       public function getA(): int
       {
           return $this->a;
       }
   
       public function getB(): int
       {
           return $this->b;
       }
   }
   

7. Откройте файл по пути src\MessageHandler\SumHandler.php и приведите его к виду:

   <?php
   
   namespace App\MessageHandler;
   
   use App\Message\Sum;
   use Symfony\Component\Messenger\Attribute\AsMessageHandler;
   
   #[AsMessageHandler]
   final class SumHandler
   {
       public function __invoke(Sum $message): void
       {
           printf('Sum of %d and %d is %d',
               $message->getA(),
               $message->getB(),
               $message->getA() + $message->getB()
           );
       }
   }
   
   

8. Откройте файл .env и найдите в нем строку MESSENGER_TRANSPORT_DSN=doctrine://default?auto_setup=0. Приведите ее к следующему виду:

   MESSENGER_TRANSPORT_DSN=sqs://message-queue.api.cloud.yandex.net/b1gvlrnlei4l********/dj6000000000********/symfony-test?access_key=KEY&secret_key=SECRET&region=ru-central1
   

   Часть b1gvlrnlei4l********/dj6000000000********/symfony-test нужно заменить на путь, скопированный в консоли Yandex Cloud.

   В параметрах access_key=KEY и secret_key=SECRET значения KEY и SECRET нужно заменить на значение статического ключа доступа к Message Queue.

9. Откройте файл config/packages/messenger.yaml и приведите его к следующему виду:

   framework:
       messenger:
           failure_transport: failed
   
           transports:
               # https://symfony.com/doc/current/messenger.html#transport-configuration
               async:
                   dsn: '%env(MESSENGER_TRANSPORT_DSN)%'
                   retry_strategy:
                       max_retries: 3
                       multiplier: 2
               failed: 'doctrine://default?queue_name=failed'
               # sync: 'sync://'
   
           routing:
               Symfony\Component\Mailer\Messenger\SendEmailMessage: async
               Symfony\Component\Notifier\Message\ChatMessage: async
               Symfony\Component\Notifier\Message\SmsMessage: async
               App\Message\Sum: async
   
               # Route your messages to the transports
               # 'App\Message\YourMessage': async
   

10. Выполните команду для отправки сообщения в очередь:

    php bin/console  app:create
    

11. Выполните команду для обработки очереди:

    php bin/console messenger:consume async