New in Symfony 8.2 Faster Messenger Workers

Parallel Message Processing

Contributed by Nicolas Grekas in #63650

A Symfony Messenger worker handles one message at a time. While a handler waits for an HTTP response or a database query, its worker cannot handle another message. If you wanted to process several messages at once, you had to run several messenger:consume processes. In Symfony 8.2, you can use the new --concurrency option:

1
2
3
4
5

# first, install this dependency once in your project
$ composer require amphp/parallel

# handle up to 4 messages at the same time
$ php bin/console messenger:consume async --concurrency=4

When using concurrency, the worker relies on the amphp/parallel library to start a pool of child processes which boot your application and handle the messages (if the parallel PHP extension is also installed, it uses threads instead of processes). The worker still fetches the messages itself, so it only needs one connection to the transport, no matter how many messages are handled in parallel.

In a benchmark with messages that take 20 ms each, --concurrency=8 was 7.5 times faster than a single worker.

Batch handlers are an exception. All the messages of a batch must be handled by the same child process, so --concurrency doesn't make them faster. To handle batched messages in parallel, keep running several workers as before.

If you used a batch handler only to handle messages faster (and not because handling them together is cheaper), consider replacing it with a regular handler and the new --concurrency option.

Faster AMQP Transport

Contributed by Nicolas Grekas in #65924 and #65168

The AMQP transport asks the broker for one message at a time, which requires a network round trip per message. RabbitMQ considers this the least efficient way to consume messages. Symfony 8.2 adds a prefetch_count option which registers a real consumer on each queue and lets the broker push messages to it in advance:

1
2

# .env
MESSENGER_TRANSPORT_DSN=amqp://guest:guest@localhost:5672/%2f/messages?prefetch_count=20

In benchmarks against a local broker, this consumed messages 13 to 18 times faster. Registered consumers also appear in the RabbitMQ management UI. Set prefetch_count higher than the worker's fetch size, which defaults to the --concurrency value.

Prefetching is disabled by default because it changes how messages are consumed: the broker now decides the order across queues, and any prefetched messages still waiting to be handled are redelivered when the worker stops.

Delayed messages also become cheaper to send. AMQP has no native delays, so Symfony creates a delay queue for each distinct delay value. The default retry strategy adds jitter to these delays, resulting in more than a thousand different values. Each retry therefore usually declares and binds a new queue.

In Symfony 8.2, delays are rounded up to two significant digits (e.g. 5234 ms becomes 5300 ms). This reduces the number of delay queues by orders of magnitude, while ensuring that messages are never released before the requested delay and at most 10% later. Use the new delay[granularity] option to round delays to a fixed number of milliseconds instead, or set it to 1 to disable rounding:

1
2

# .env
MESSENGER_TRANSPORT_DSN=amqp://guest:guest@localhost:5672/%2f/messages?delay[granularity]=1000

Logging Processing Time and Memory Usage

Contributed by Marie Charles in #64621

This feature won't make your workers faster, but it will help you find which messages are slow. The new optional logging middleware records processing time and memory usage for each message. Enable it on the buses you want to monitor:

1
2
3
4
5
6
7

# config/packages/messenger.yaml
framework:
    messenger:
        buses:
            messenger.bus.default:
                middleware:
                    - logging

1
2
3
4
5
6
7
8
9
10
11
12
13
14

// config/packages/messenger.php
namespace Symfony\Component\DependencyInjection\Loader\Configurator;

return App::config([
    'framework' => [
        'messenger' => [
            'buses' => [
                'messenger.bus.default' => [
                    'middleware' => ['logging'],
                ],
            ],
        ],
    ],
]);

The middleware measures everything that runs after it in the stack. It logs messages in the messenger channel with the class, duration_ms and memory_usage (in bytes) context keys, so you can build dashboards and alerts from them:

1
2

messenger.INFO: "App\Message\GenerateInvoice" message successfully handled. {"class":"App\\Message\\GenerateInvoice","duration_ms":1834,"memory_usage":12582912}
messenger.ERROR: Unable to handle "App\Message\SyncCatalog" message. {"class":"App\\Message\\SyncCatalog","duration_ms":30012,"memory_usage":524288,"exception":"..."}