php - How do I split queues in Symfony Messenger? - Stack Overflow

admin管理员组

文章数量:1430279

Could you please explain why, when launching the indexing process using the command $this->bus->dispatch(new ProductIndexMessage($productIds)) (with hundreds of such commands being executed at once), they are evenly distributed across all transports and queues? According to the design, they should only be launched on the searchIndex queue.

PHP: 8.4.5, Symfony 7.2

/usr/local/bin/php -f /var/www/html/app/bin/console messenger:stats
 ---------------- -------
  Transport        Count  
 ---------------- -------
  async            40     
  searchIndex      40     
  genImageThumbs   40     
  failed           6      
 ---------------- -------

File:

<?php

declare(strict_types=1);

namespace App\Message\Search;

use App\Message\SearchIndexMessageInterface;
use App\Utils\Utils;

final readonly class ProductIndexMessage implements SearchIndexMessageInterface
{
    /**
     * @var int[]
     */
    public array $productIds;

    public function __construct(int|array $productIds)
    {
        $this->productIds = Utils::asPosAndNotZeroNotDupIntegerArray($productIds);
    }
}

There is a configured messanger.yaml:

framework:
    messenger:
        failure_transport: failed

        transports:
            async:
                dsn: '%env(MESSENGER_TRANSPORT_DSN)%/async'
                retry_strategy:
                    max_retries: 20
                    multiplier: 2

            searchIndex:
                dsn: '%env(MESSENGER_TRANSPORT_DSN)%/searchIndex'
                options:
                    exchange:
                        name: 'searchIndex'
                retry_strategy:
                    max_retries: 20
                    multiplier: 2

            genImageThumbs:
                dsn: '%env(MESSENGER_TRANSPORT_DSN)%/genImageThumbs'
                options:
                    exchange:
                        name: 'genImageThumbs'
                retry_strategy:
                    max_retries: 1
                    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

            # Route your messages to the transports
            App\Message\AsyncMessageInterface: async
            App\Message\SearchIndexMessageInterface: searchIndex
            App\Message\ImageGenThumbsMessageInterface: genImageThumbs

Commands to start workers:

/usr/local/bin/php /var/www/html/app/bin/console messenger:consume genImageThumbs --limit=1500 --time-limit=86400
/usr/local/bin/php /var/www/html/app/bin/console messenger:consume searchIndex --limit=1500 --time-limit=86400
/usr/local/bin/php /var/www/html/app/bin/console messenger:consume async --limit=1500 --time-limit=86400

Could you please explain why, when launching the indexing process using the command $this->bus->dispatch(new ProductIndexMessage($productIds)) (with hundreds of such commands being executed at once), they are evenly distributed across all transports and queues? According to the design, they should only be launched on the searchIndex queue.

PHP: 8.4.5, Symfony 7.2

/usr/local/bin/php -f /var/www/html/app/bin/console messenger:stats
 ---------------- -------
  Transport        Count  
 ---------------- -------
  async            40     
  searchIndex      40     
  genImageThumbs   40     
  failed           6      
 ---------------- -------

File:

<?php

declare(strict_types=1);

namespace App\Message\Search;

use App\Message\SearchIndexMessageInterface;
use App\Utils\Utils;

final readonly class ProductIndexMessage implements SearchIndexMessageInterface
{
    /**
     * @var int[]
     */
    public array $productIds;

    public function __construct(int|array $productIds)
    {
        $this->productIds = Utils::asPosAndNotZeroNotDupIntegerArray($productIds);
    }
}

There is a configured messanger.yaml:

framework:
    messenger:
        failure_transport: failed

        transports:
            async:
                dsn: '%env(MESSENGER_TRANSPORT_DSN)%/async'
                retry_strategy:
                    max_retries: 20
                    multiplier: 2

            searchIndex:
                dsn: '%env(MESSENGER_TRANSPORT_DSN)%/searchIndex'
                options:
                    exchange:
                        name: 'searchIndex'
                retry_strategy:
                    max_retries: 20
                    multiplier: 2

            genImageThumbs:
                dsn: '%env(MESSENGER_TRANSPORT_DSN)%/genImageThumbs'
                options:
                    exchange:
                        name: 'genImageThumbs'
                retry_strategy:
                    max_retries: 1
                    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

            # Route your messages to the transports
            App\Message\AsyncMessageInterface: async
            App\Message\SearchIndexMessageInterface: searchIndex
            App\Message\ImageGenThumbsMessageInterface: genImageThumbs

Commands to start workers:

/usr/local/bin/php /var/www/html/app/bin/console messenger:consume genImageThumbs --limit=1500 --time-limit=86400
/usr/local/bin/php /var/www/html/app/bin/console messenger:consume searchIndex --limit=1500 --time-limit=86400
/usr/local/bin/php /var/www/html/app/bin/console messenger:consume async --limit=1500 --time-limit=86400

• php
• symfony
• rabbitmq

Share Improve this question Follow asked Mar 17 at 9:49 Denis ShlyapnikovDenis Shlyapnikov 1911 silver badge22 bronze badges 0

Add a comment  | 

1 Answer 1

Sorted by: Reset to default 0

Solution: It is necessary to add an indication of the queue to the configuration, for example:

                    queues:
                        searchIndex: ~

本文标签： phpHow do I split queues in Symfony MessengerStack Overflow