Очереди сообщений

Очереди в Bitrix Framework выполняют задачи в фоне. Это помогает разгрузить систему, если какие-то операции требуют много времени или ресурсов.

Функционал очередей находится в альфа-версии разработки. Очереди можно использовать с версии 25.100.300 главного модуля, но без гарантий обратной совместимости. Разработчики могут изменить функционал в следующих обновлениях.

Как работают очереди

Очередь — это канал передачи данных от отправителя к обработчику. Сообщение помещается в брокер, а обработчик забирает его и выполняет задачу.

Основные компоненты:

  • сообщение — контейнер данных,

  • обработчик — класс, который обрабатывает сообщение,

  • брокер — хранилище сообщений,

  • очередь — логическая группа сообщений, привязанная к обработчику,

  • воркер — запускает обработчики настроенных очередей. Обработчик получает доступные сообщения из брокера и выполняет задачу.

В режиме web система запускает воркер через фоновые задачи после отправки ответа пользователю. В режиме cli воркер запускает консольная команда messenger:consume.

Быстрый старт

Создайте сообщение и обработчик, настройте очередь и отправьте сообщение. Для проверки запустите обработку в консольном режиме и найдите запись в логе.

Для других сценариев можно настроить повторные попытки, отложить обработку и использовать свою таблицу для хранения сообщений.

Пример настройки очереди сообщений

Используйте установленный пользовательский модуль vendor.module. Классы разместите в его папке lib/ по правилам автозагрузки.

1. Создать класс сообщения

Класс AbstractMessage передает публичные свойства для JSON-сериализации и восстанавливает объект перед обработкой. Для полей типов string, int, float, bool и массивов таких значений переопределять методы jsonSerialize() и createFromData() не нужно. Имена параметров конструктора должны совпадать с именами сохраненных полей.

Создайте файл /local/modules/vendor.module/lib/Public/Message/MyMessage.php:

<?php

namespace Vendor\Module\Public\Message;

use Bitrix\Main\Messenger\Entity\AbstractMessage;

class MyMessage extends AbstractMessage
{
    public function __construct(
        public readonly string $text,
        public readonly int $itemId
    )
    {
    }
}

Поле text содержит текст сообщения, а itemId — идентификатор объекта, к которому оно относится.

Если данные требуют собственного преобразования, переопределите jsonSerialize() для сохранения и createFromData() для восстановления объекта.

2. Создать класс обработчика

Обработчик получает сообщение и записывает его данные в лог. Создайте файл /local/modules/vendor.module/lib/Internals/Messenger/Receiver/MyReceiver.php:

<?php

namespace Vendor\Module\Internals\Messenger\Receiver;

use Bitrix\Main\Diag\LoggerFactory;
use Bitrix\Main\Messenger\Entity\MessageInterface;
use Bitrix\Main\Messenger\Internals\Exception\Receiver\UnprocessableMessageException;
use Bitrix\Main\Messenger\Receiver\AbstractReceiver;
use Vendor\Module\Public\Message\MyMessage;

class MyReceiver extends AbstractReceiver
{
    protected function process(MessageInterface $message): void
    {
        if (!($message instanceof MyMessage))
        {
            throw new UnprocessableMessageException($message);
        }

        $logger = (new LoggerFactory())->createDefault();
        $logger->info(
            'Queue message: item {itemId}, text {text}',
            [
                'itemId' => $message->itemId,
                'text' => $message->text,
            ]
        );
    }
}

Для файлового логирования задайте непустую константу LOG_FILENAME в общем файле /local/php_interface/init.php. Укажите абсолютный путь к файлу вне публичной папки сайта, доступный для записи пользователю PHP. Если константа уже задана, используйте существующий путь.

Без LOG_FILENAME метод createDefault() возвращает NullLogger, который не сохраняет сообщения. Подробнее о настройке — в статье Логгеры.

3. Добавить конфигурацию в .settings.php

Добавьте секцию messenger в файл /local/modules/vendor.module/.settings.php. В параметре handler укажите полное имя созданного обработчика. Если файл уже содержит другие настройки, сохраните их:

<?php

return [
    'messenger' => [
        'value' => [
            'queues' => [ // список очередей
                'first_queue' => [ // очередь с именем 'first_queue'
                    'handler' => \Vendor\Module\Internals\Messenger\Receiver\MyReceiver::class,
                ],
            ],
        ],
        'readonly' => true, // защита от перезаписи настроек через API
    ],
];

В глобальном файле /bitrix/.settings.php настройте брокер default и режим cli. Добавьте секцию в существующий массив настроек или дополните ее, если она уже есть:

'messenger' => [
    'value' => [
        'run_mode' => 'cli',
        'brokers' => [
            'default' => [
                'type' => 'db',
                'params' => [
                    'table' => \Bitrix\Main\Messenger\Internals\Storage\Db\Model\MessengerMessageTable::class,
                ],
            ],
        ],
    ],
    'readonly' => true,
],

Режим run_mode действует на все очереди проекта. Если используется web, переключение на cli требует запуска консольного обработчика и для остальных очередей.

Подробнее — в разделе Конфигурация очередей.

4. Отправить сообщение

В коде сайта после подключения ядра подключите модуль и отправьте сообщение. Имя очереди в send() должно совпадать с ключом в конфигурации:

use Bitrix\Main\Loader;
use Vendor\Module\Public\Message\MyMessage;

Loader::requireModule('vendor.module');

$message = new MyMessage('Мое первое сообщение!', 12345);
$message->send('first_queue');

Метод send() сохраняет сообщение через брокер. В режиме cli оно будет ждать запуска консольного обработчика.

5. Запустить обработку и проверить результат

Выполните команду из папки /bitrix/ сайта. Для проверки ограничьте время работы десятью секундами:

php bitrix.php messenger:consume first_queue —time-limit 10

Откройте файл, указанный в LOG_FILENAME, и найдите сообщение Queue message: item 12345, text Мое первое сообщение!. Форматтер может добавлять к нему дату и другие данные. Если записи нет, проверьте настройки логирования, права на файл и совпадение имени очереди в конфигурации и вызове send().

Создать классы через консоль

Для создания классов можно использовать команды make:message и make:messagehandler. Они доступны с версии 25.900.0 главного модуля. Выполните команды из папки /bitrix/ сайта для установленного модуля vendor.module:

php bitrix.php make:message My -m vendor.module -n
php bitrix.php make:messagehandler My —handler-module=vendor.module —message-module=vendor.module -n

Первая команда создает класс MyMessage, вторая — MyMessageHandler. Параметр -m задает модуль сообщения, —handler-module — модуль обработчика, —message-module — модуль сообщения, а -n отключает интерактивные вопросы.

Добавьте поля сообщения и код обработки в сгенерированные классы. В обработчике подключите класс сообщения через use Vendor\Module\Public\Message\MyMessage;.

Генератор размещает обработчик в пространстве имен Vendor\Module\Internals\Integration\Vendor\Module\MessageHandler. Укажите полное имя \Vendor\Module\Internals\Integration\Vendor\Module\MessageHandler\MyMessageHandler::class в параметре handler конфигурации.

Конфигурация очередей

Систему очередей нужно настроить в файле .settings.php. Конфигурация может быть:

  • глобальной — в файле bitrix/.settings.php для общих настроек и межмодульных очередей,

  • модульной — в файле .settings.php модуля, для очередей из этого модуля.

Конфигурацию очередей нужно указывать в файле .settings.php того модуля, к которому относятся очереди. В глобальном файле указываются только те очереди, которые одновременно используются в нескольких модулях.

Глобальная конфигурация

Глобальную конфигурацию настраивают, когда очереди и брокер планируют использовать в разных модулях, например, для интеграции между модулями.

По умолчанию в /bitrix/.settings.php создается секция messenger с параметрами run_mode и brokers, где описан один брокер по умолчанию default.

return [
    'messenger' => [
        'value' => [
            'run_mode' => 'web', // web, cli
            'brokers' => [
                'default' => [
                    'type' => DbBroker::TYPE_CODE, // сейчас один тип — db
                    'params' => [
                        'table' => MessengerMessageTable::class,
                    ]
                ],
                'db_calendar' => [
                    'type' => 'db',
                    'params' => [
                        'table' => 'b_calendar_messages',
                        'module' => 'calendar',
                    ],
                ],
            ],
            'queues' => [
                'first_queue' => [
                    'handler' => MyReceiver::class,
                ],
                'google_section_sync' => [
                    'broker' => 'db_calendar',
                    'handler' => GoogleSectionReceiver::class,
                ],
            ],
        ],
        'readonly' => true,
    ],
];

Параметры

Режим run_mode — определяет как обрабатывать сообщения:

  • web — в фоновых задачах,

  • cli — в консольном скрипте.

При значении cli нужно настроить запуск скрипта вручную командой:

php bitrix.php messenger:consume

Брокеры brokers — доступные хранилища сообщений. Ключ — имя брокера, а значение — его параметры.

  • type — тип брокера. Сейчас поддерживается только значение db, то есть хранение в базе данных.

  • params — параметры брокера. Для типа db:

    • table — таблица для хранения сообщений,

    • module — модуль, с которым связана таблица. По умолчанию имеет значение main.

Брокер по умолчанию default должен всегда быть в глобальной конфигурации.

Очереди queues — список доступных очередей с указанием обработчиков.

Модульная конфигурация

В .settings.php модуля указываются очереди, используемые этим модулем.

return [
    ...
    'messenger' => [
        'value' => [
            'queues' => [
                'google_section_sync' => [
                    'broker' => 'db_calendar',
                    'handler' => \Bitrix\Calendar\Synchronization\Internals\Messenger\Receiver\GoogleSectionReceiver::class,
                ],
                'google_event_sync' => [
                    'handler' => \Bitrix\Calendar\Synchronization\Internals\Messenger\Receiver\GoogleEventReceiver::class,
                    'limit' => 30,
                    'total_processing_limit' => 50,
                ],
            ],
        ],
        'readonly' => true,
    ],
];

Параметры

Брокеры brokers — доступные хранилища сообщений. Структура объекта полностью идентична глобальной конфигурации.

Очереди queues — доступные очереди. Ключ — идентификатор очереди, а значение — параметры очереди:

  • handler — класс-обработчик сообщений,

  • broker — брокер очереди. Если его не указать, используется брокер по умолчанию,

  • limit — количество обрабатываемых обработчиком сообщений за раз, по умолчанию — 50,

  • total_processing_limit — максимальное количество одновременно обрабатываемых сообщений в очереди, по умолчанию — 100,

    Значение не может быть меньше значения параметра limit, иначе при обработке очереди будет выброшено исключение.

  • retry_strategy — настройки повторной обработки сообщений в случае неудачной попытки обработки,

  • module — модуль, к которому относится очередь. Всегда имеет значение идентификатора модуля.

Ограничение количества обрабатываемых сообщений

Количество обрабатываемых сообщений настраивается двумя параметрами очереди.

  1. Параметр limit задает количество сообщений, которые обработчик выбирает из очереди за один раз.

    Пример. Если обработка выполняется на хите в режиме web и обработка одного сообщения занимает около 10 секунд, значение 50 может создавать лишнюю нагрузку. Достаточно обрабатывать по 2-3 сообщения за хит, чтобы очередь разбиралась равномерно и не влияла на работу другого функционала.

  2. Параметр total_processing_limit задает максимальное количество сообщений, которые могут одновременно обрабатываться в очереди всеми обработчиками.

    Пример. Если limit равен 30 и total_processing_limit — 50, то при наличии 100 готовых сообщений первый обработчик заберет 30, а второй сможет забрать только 20. Параметр помогает ограничить общую нагрузку на систему и избежать ситуации, когда длительная обработка сообщений блокирует работу портала.

Значение параметра total_processing_limit не должно быть меньше, чем значение параметра limit, иначе при обработке очереди будет выброшено исключение, чтобы подсветить проблему в конфигурации.

Повторная обработка сообщений

Сообщение может не обработаться с первой попытки, например, из-за временной недоступности сервиса. Тогда сообщение нужно обработать повторно.

Используйте повторную обработку:

  • при работе с внешними API,

  • при нестабильных соединениях,

  • для ресурсоемких задач.

Параметры

Для каждой очереди можно указать в конфигурации параметр retry_strategy с настройками повторной обработки.

  • max_retries — максимальное количество попыток обработки, включая исходную попытку. По умолчанию — 3.

  • delay — базовая задержка между попытками в секундах. По умолчанию — 1.

  • multiplier — множитель задержки между попытками. По умолчанию — 2.

  • max_delay — максимальное значение задержки между повторами в секундах. Ограничивает задержку, но не отменяет повторные попытки. По умолчанию — 0, то есть без ограничения.

Пример конфигурации

return [
    'messenger' => [
        'value' => [
            'queues' => [
                'second_queue' => [
                    'handler' => MyReceiver::class,
                    'retry_strategy' => [
                        'max_retries' => 10,
                        'delay' => 5,       // базовая задержка — пять секунд
                        'multiplier' => 2,  // удваиваем с каждой попыткой
                        'max_delay' => 300  // максимум пять минут между попытками
                    ],
                ],
            ],
        ],
        'readonly' => true,
    ],
];

  • Первая повторная попытка станет доступна через 10 секунд после ошибки. Задержка равна delay * multiplier, то есть 5 * 2.

  • Вторая — через 20 секунд после следующей ошибки.

  • Третья — через 40 секунд. Далее задержка увеличивается с учетом multiplier.

  • Шестая и последующие повторные попытки станут доступны через 300 секунд после ошибки. Расчетная задержка для шестого повтора равна 5 * 2 ** 6 = 320 секундам, поэтому применяется ограничение max_delay.

При значении max_retries = 10 обработчик может выполнить исходную попытку и до девяти повторов для обычных исключений. Если все десять попыток завершатся ошибкой, брокер удалит сообщение из очереди. Для исключения RecoverableMessageException ограничение количества попыток не действует.

Своя таблица для сообщений

При работе с DBBroker сообщения сохраняются в базу данных. По умолчанию используется таблица b_main_messenger_message. Можно создать отдельную таблицу для своего модуля.

  1. Создайте класс-наследник MessengerMessageTable.

    use Bitrix\Main\Messenger\Internals\Storage\Db\Model\MessengerMessageTable;

class CalendarMessagesTable extends MessengerMessageTable
{
    public static function getTableName(): string
    {
        return 'b_calendar_messenger_message';
    }
}

  2. Укажите свою таблицу в конфигурации брокера.

    return [
    'messenger' => [
        'value' => [
            'brokers' => [
                'db_calendar' => [
                    'type' => 'db',
                    'params' => [
                        'table' => CalendarMessagesTable::class,
                    ],
                ],
            ],
        ],
        'readonly' => true,
    ],
];

  3. Укажите имя брокера db_calendar в настройках очередей. После этого сообщения будут сохраняться в созданную таблицу.

За создание и удаление собственной таблицы отвечает модуль, к которому относится таблица. Поэтому обязательно добавьте запрос создания таблицы в установщик модуля.

Запуск обработчика очередей

Система поддерживает два режима обработки.

  1. web — обработка через фоновые задачи. Работает по умолчанию и без дополнительных настроек.

  2. cli — консольный режим. Требует дополнительной настройки и предоставляет больше возможностей для управления обработкой. Для производственной среды рекомендуется использовать cli-режим с супервизором.

Консольный режим

Чтобы использовать консольный режим, установите в конфигурации 'run_mode' => 'cli'.

Основная команда для обработки:

php bitrix.php messenger:consume

Она запускает обработчики для всех очередей в бесконечном цикле.

Параметры команды:

  • Выбор очередей. Можно указать несколько:

    php bitrix.php messenger:consume first_queue second_queue

  • Дополнительные опции:

    # Обрабатывает очередь second_queue 60 секунд с паузой 2 секунды между проверками
php bitrix.php messenger:consume second_queue -t 60 —sleep 2

    • -t, —time-limit — ограничить время работы, в секундах,

    • —sleep — установить паузу между итерациями. По умолчанию — 1 секунда.

Обработчики сообщений

Каждая очередь использует свой обработчик. Он должен:

  • обработать сообщение,

  • выбросить исключение при ошибке.

Чтобы создать обработчик, наследуйтесь от класса AbstractReceiver и реализуйте метод process. Метод возвращает void, когда сообщение успешно обработано, или бросает исключение.

class SomeReceiver extends AbstractReceiver
{
    protected function process(MessageInterface $message): void
    {
        if ($message instanceof SectionMessage)
        {
            $section = $this->sectionRepository->getById($message->getSectionId());
            (new SendSectionCommand($section))->run();
            return;
        }

        throw new UnprocessableMessageException($message->getId(), $this->queueId);
    }
}

Типы исключений

Используйте разные исключения для разных сценариев.

  • UnprocessableMessageException — сообщение не подходит для этого обработчика. Пример: в очередь для обработки задач пришло сообщение о пользователе.

  • UnrecoverableMessageException — сообщение устарело и повторная обработка не нужна. Пример: сообщение содержит ID удаленного объекта.

  • RecoverableMessageException — временная ошибка, стоит повторить попытку. Можно указать задержку в секундах через метод getRetryDelay(). Если возвращает null, используются настройки повтора из конфигурации очереди.

  • Другие исключения — критические ошибки обработки. Система применит к ним настройки повторной обработки retry_strategy из конфигурации очереди.

Пример обработчика сообщений в консольном режиме

Создайте обработчик SomeReceiver, который выводит данные полученного сообщения в формате JSON:

use Bitrix\Main\Messenger\Entity\MessageInterface;
use Bitrix\Main\Messenger\Receiver\AbstractReceiver;

class SomeReceiver extends AbstractReceiver
{
    protected function process(MessageInterface $message): void
    {
        echo json_encode($message, JSON_THROW_ON_ERROR) . PHP_EOL;
    }
}

Зарегистрируйте обработчик для нужной очереди в параметре handler и запустите команду messenger:consume в режиме cli.

Метод Worker::process() выполняет один проход по настроенным очередям. Цикл повторных проходов и паузы между ними организует команда messenger:consume. Для постоянной обработки в режиме cli используйте эту команду.

Собственная реализация

Для особых сценариев обработки можно реализовать собственный обработчик через интерфейс ReceiverInterface.

interface ReceiverInterface
{
    /**
     * Основной метод обработки
     * @throws ReceiverException
     */
    public function run(): void;

    // Установка лимита сообщений за одну итерацию
    public function setLimit(int $limit): self;

    // Назначение очереди
    public function setQueueId(string $queueId): self;

    // Выбор брокера
    public function setBroker(BrokerInterface $broker): self;
}

При реализации интерфейса необходимо самостоятельно управлять:

  • получением сообщений из брокера,

  • подтверждением успешной обработки ack,

  • обработкой ошибок reject.

Методы интерфейса:

  • run() — получает сообщения из очереди, выполняет обработку и выбрасывает ReceiverException при неудачной обработке.

  • setLimit(int $limit) — устанавливает лимит сообщений для одновременной обработки. Например, setLimit(10) задает обработку по 10 сообщений.

  • setQueueId(string $queueId) — назначает очередь для обработки.

  • setBroker(BrokerInterface $broker) — подключает брокера сообщений, например, setBroker(new RedisBroker()).

Работа с сообщениями

Сообщения — это контейнеры данных для передачи через очередь. Они должны реализовывать \Bitrix\Main\Messenger\Entity\MessageInterface, но лучше расширять класс \Bitrix\Main\Messenger\Entity\AbstractMessage.

Ключевые требования к сообщениям:

  • Данные должны преобразовываться в JSON, то есть быть сериализуемыми.

  • Можно использовать любые поля и структуры данных.

  • Необходимо учитывать возможную задержку обработки.

Отправить сообщение

Создайте объект сообщения и отправьте в очередь.

$message = new MyMessage(...);
$message->send('first_queue');

Отложить обработку сообщения

Параметр DelayParam(3600) откладывает обработку на 1 час (3600 секунд). Сообщение появится в очереди, но обработчик получит его только через указанное время.

use Bitrix\Main\Messenger\Entity\ProcessingParam\DelayParam;

$message = new MyMessage(...);
$message->send('first_queue', [new DelayParam(3600)]);

Отправить сообщение с дополнительным параметром

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

  • DelayParam — отложить обработку на 1 час,

  • ItemIdParam — отметить сообщение идентификатором myid-123.

use Bitrix\Main\Messenger\Entity\ProcessingParam\DelayParam;
use Bitrix\Main\Messenger\Entity\ProcessingParam\ItemIdParam;

$message = new MyMessage(...);
$message->send('first_queue', [new DelayParam(3600), new ItemIdParam('myid-123')]);

Создать свой параметр

Чтобы добавить новый параметр, создайте класс, реализующий интерфейс \Bitrix\Main\Messenger\Entity\ProcessingParam\ProcessingParamInterface:

interface ProcessingParamInterface
{
    public function applyTo(MessageBox $messageBox): MessageBox;
}

Интерфейс требует всего один метод — applyTo. Он принимает объект класса \Bitrix\Main\Messenger\Entity\MessageBox — обертку для сообщения, которое отправляется в брокер очередей.

Особенности работы с отложенной обработкой

Сообщения могут обрабатываться не сразу, что требует особого подхода к их формированию.

Если сообщение обрабатывается с задержкой, в него нужно включать все данные, которые потребуются обработчику.

Пример. В системе есть задача с ID = 123, связанная с внешним сервисом ID = trk-042. При удалении задачи нужно уведомить внешний сервис. Если передать только task_id, то к моменту обработки задача уже может быть удалена из базы данных, и внешний идентификатор станет недоступен.

В таком случае в сообщение нужно сразу включить все необходимые данные.

$message->setData([
    'task_id' => 123,
    'external_id' => 'trk-042',
]);

Такой подход гарантирует, что обработчик получит все данные даже после удаления исходной сущности. Это особенно важно для операций удаления, синхронизации с внешними системами и сценариев, в которых важна согласованность данных.

Предыдущая
События
Следующая
Временное хранение