Running Consumers

Two ways to keep queue consumers alive - Magento's cron runner or a process manager - and why one dies without telling you.

Syntaxbin/magento queue:consumers:start <consumer> [--max-messages=N] [--batch-size=N] [--area-code=global] [--single-thread]

A consumer is a PHP process that reads one queue and calls a handler for each message. Nothing in a queue moves until a consumer runs. Magento can start consumers from cron, or you can supervise them yourself with a process manager such as supervisord. Choose one. Both at once gives you duplicate work.

The cron runner

Magento_MessageQueue declares a job named consumers_runner in the consumers cron group, scheduled every minute. It reads cron_consumers_runner from app/etc/env.php: cron_run (default true), max_messages (default 10000), consumers (an allow list; empty means every consumer) and multiple_processes (how many copies of one consumer to start). Each process is started with --single-thread, which takes a lock so a second copy cannot start.

Why consumers exit on purpose

max_messages is deliberate. The process stops after that many messages so PHP releases memory, and cron starts it again within the minute. queue/consumers_wait_for_messages (default 1) keeps it waiting on an empty queue instead of exiting; queue/only_spawn_when_message_available (default true) skips starting a consumer whose queue is empty.

What trips people up

A running consumer holds the code it started with, so it must be restarted after every deploy. If it is killed - an out-of-memory kill, a container restart - the lock can outlive the process, and the hourly messagequeue_clean_outdated_locks job is what frees it. Editing env.php needs no command, but the consumers must be restarted to see the change.

Example

Example · xml
<!-- app/code/Vendor/Feed/etc/queue_consumer.xml -->
<config xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
        xsi:noNamespaceSchemaLocation="urn:magento:framework-message-queue:etc/consumer.xsd">
    <consumer name="vendor.feed.sync"
              queue="vendor.feed.sync"
              handler="Vendor\Feed\Model\SyncItemHandler::process"
              maxMessages="500"
              maxIdleTime="180"
              sleep="10"
              onlySpawnWhenMessageAvailable="1"/>
</config>

When to use it

  • Decide between the cron runner and supervisord for a new production install.
  • Stop a consumer from being started twice on a two-node cluster.
  • Find out why an admin export never produced a file.
  • Add a consumer restart to the deploy script so new handler code is used.

More examples

Deployment configuration for consumers

This is the shape of the configuration, not a whole env.php - merge the keys into the array the file already returns. cron_run false switches the cron runner off, which is what you want when supervisord owns the processes. Setting consumers to a short list stops cron starting every consumer on every node, and consumers_wait_for_messages 0 lets a consumer exit as soon as its queue is empty instead of idling until max_messages.

Example · php
<?php
// app/etc/env.php (fragment)

return [
    'cron_consumers_runner' => [
        'cron_run' => false,
        'max_messages' => 1000,
        'consumers' => [
            'async.operations.all',
            'vendor.feed.sync',
        ],
        'multiple_processes' => [
            'async.operations.all' => 2,
        ],
    ],
    'queue' => [
        'consumers_wait_for_messages' => 0,
        'only_spawn_when_message_available' => 1,
    ],
];

The same consumers under supervisord

supervisord restarts the process the moment it exits after max-messages, which is faster and steadier than waiting for the next cron minute. Keep --single-thread so two copies never read the same queue. Set cron_run to false in env.php first, or you get these two processes plus whatever cron starts, and after every release run supervisorctl restart magento-async-operations: a consumer started before the deploy keeps running the old code.

Example · ini
; /etc/supervisor/conf.d/magento-consumers.conf
[program:magento-async-operations]
command=/usr/bin/php8.4 /var/www/magento/bin/magento queue:consumers:start async.operations.all --max-messages=1000 --single-thread
process_name=%(program_name)s_%(process_num)02d
numprocs=2
autostart=true
autorestart=true
startretries=10
user=www-data
stdout_logfile=/var/log/supervisor/magento-async-operations.log
redirect_stderr=true

Checking what is actually running

queue:consumers:list prints the consumers Magento knows about, which is the quickest way to confirm a new queue_consumer.xml was read; if a name is missing, run bin/magento cache:flush. Starting one in the foreground with a small --max-messages is the normal way to debug a handler, because errors print to the terminal instead of disappearing into var/log. If messages sit unread and no process exists, suspect a lock left by a killed consumer.

Example · bash
bin/magento queue:consumers:list
# async.operations.all
# exportProcessor
# product_action_attribute.update
# vendor.feed.sync

# Drain one queue in the foreground and stop
bin/magento queue:consumers:start vendor.feed.sync --max-messages=50 --single-thread

# Nothing moving? check for a process, then for a stale lock
ps -ef | grep 'queue:consumers:start'

Related

m2 message queuesm2 cronm2 deploy consumersm2 async bulk apim2 debug production

Discussion

  • Be the first to comment on this lesson.