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

В статье вместе с Ильей Рупасовым@rpsv, Григорием Бычеком@gbychek и Мариной Павловой @MarinaPav разбираем один такой сценарий: после изменения события в календаре Битрикс24 нужно синхронизировать его с Google Calendar. Вынесем эту операцию в очередь, создадим сообщение и обработчик, настроим очередь и проверим результат в логе.

❗️ Механизм очередей доступен с версии 25.100.300 главного модуля 

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

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

Вот пример: сотрудник изменил событие в календаре Битрикс24, а его нужно синхронизировать с Google Calendar. При обычной синхронной обработке запрос будет ждать, пока Битрикс24 обратится к Google и получит ответ. Если внешний сервис отвечает медленно, пользователь тоже будет ждать. Если сервис временно недоступен, ошибка может повлиять и на основной запрос.

При асинхронной обработке эти действия разделяются. Битрикс24 сохраняет изменение события и завершает текущий запрос, а задачу синхронизации передаёт в очередь. Она будет выполнена отдельно, когда обработчик получит сообщение.

Для этого механизм очередей использует несколько элементов. 

• Сообщение содержит данные, которые понадобятся для фоновой операции — например, ID события и внешнего календаря.

• Очередь определяет, какой обработчик должен получить сообщение и с какими настройками оно будет обрабатываться.

• Брокер сохраняет сообщения до момента обработки. В текущей версии Bitrix Framework поддерживается брокер типа db, поэтому сообщения хранятся в базе данных.

• Воркер получает доступные сообщения из брокера и передаёт их обработчику. Обработчик выполняет саму фоновую работу. В нашей статье реальную синхронизацию с Google Calendar заменим записью в лог.

Весь путь выглядит так:

изменение события → сообщение → очередь → брокер → воркер → обработчик → лог

Очереди могут работать в режимах web и cli. В режиме web обработка запускается через фоновые задачи после завершения веб-запроса. В режиме cli сообщения получает отдельный консольный воркер. Дальше воспользуемся cli, чтобы явно запустить обработку и проверить результат в логе.

Создаём сообщение

Начнём с сообщения — объекта, в котором хранятся данные для фоновой обработки. В нашем случае нужно передать обработчику информацию о событии Битрикс24, которое затем будет синхронизировано с Google Calendar.

Класс сообщения можно создать вручную или сгенерировать встроенной консольной командой Bitrix Framework:

php bitrix.php make:message GoogleEventSync -m my.module -n

Команда make:message создаёт класс сообщения для механизма очередей. Она доступна с версии main 25.900.0. Подробнее о генерации классов можно прочитать в разделе документации о консольных командах Bitrix Framework.

Добавим в сообщение данные, которые понадобятся при синхронизации — ID события, ID внешнего календаря и тип операции:

use Bitrix\Main\Messenger\Entity\AbstractMessage;
use Bitrix\Main\Messenger\Entity\MessageInterface;

class GoogleEventSyncMessage extends AbstractMessage
{
public function __construct(
public readonly int $eventId,
public readonly string $externalCalendarId,
public readonly string $operation
) {}
}

Здесь мы используем только простые типы данных — int и string. AbstractMessage умеет автоматически преобразовывать такие поля при сохранении сообщения и восстанавливать их при обработке, поэтому дополнительный код для сериализации в нашем случае не нужен.

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

use Bitrix\Main\Messenger\Entity\AbstractMessage;
use Bitrix\Main\Messenger\Entity\MessageInterface;

class GoogleEventSyncMessage extends AbstractMessage
{
public function __construct(
public readonly int $eventId,
public readonly string $externalCalendarId,
public readonly string $operation
) {}

public function jsonSerialize(): mixed
{
return [
'eventId' => $this->eventId,
'externalCalendarId' => $this->externalCalendarId,
'operation' => $this->operation,
];
}

public static function createFromData(array $data): MessageInterface
{
return new static(...$data);
}
}

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

Например, сообщение для нашего сценария может содержать такие данные:

{
"eventId": 123,
"externalCalendarId": "google-calendar-42",
"operation": "update"
}

Само сообщение синхронизацию не запускает. Оно сохраняет данные, которые позже получит обработчик очереди.

Создаём обработчик

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

Класс обработчика можно создать вручную или сгенерировать консольной командой make:messagehandler:

php bitrix.php make:messagehandler GoogleEventSync --handler-module=my.module --message-module=my.module -n

Обработчик наследуется от AbstractReceiver. Основная работа выполняется в методе process() — Bitrix Framework вызовет его, когда сообщение дойдёт до обработки.

use Bitrix\Main\Diag\LoggerFactory;
use Bitrix\Main\Messenger\Entity\MessageInterface;
use Bitrix\Main\Messenger\Receiver\AbstractReceiver;
use Psr\Log\LoggerInterface;
use Psr\Log\NullLogger;

class GoogleEventSyncMessageHandler extends AbstractReceiver
{
    private readonly LoggerInterface $logger;

    public function __construct()
    {
        $this->logger = (new LoggerFactory())->createDefault() ?? new NullLogger();
    }

    /**
      @param GoogleEventSyncMessage $message
     /
    protected function process(MessageInterface $message): void
    {
        $this->logger->info(
            "Google Calendar sync: event {$message->eventId}, "
            . "calendar {$message->externalCalendarId}, "
            . "operation {$message->operation}"
        );
    }
}

В process() доступны данные, которые мы сохранили в GoogleEventSyncMessage: ID события, ID внешнего календаря и тип операции. Пока обработчик только записывает их в лог. Логгер создаём через LoggerFactory, а сообщение записываем методом info().

Чтобы проверить результат примера, файловое логирование должно быть настроено заранее. Подробнее о настройке можно прочитать в документации по логгерам Bitrix Framework.

Если process() завершился без исключения, сообщение считается успешно обработанным. Обработку ошибок и повторные попытки в этом сценарии настраивать не будем — для них в документации по очередям тоже есть подробный раздел.

Регистрируем очередь в конфигурации модуля

Обработчик мы создали, но система пока не знает, какую очередь он должен обслуживать. Очереди конкретного модуля настраиваются в его файле .settings.php. Зарегистрируем очередь google_event_sync и свяжем её с созданным GoogleEventSyncMessageHandler. Такая структура соответствует модульной конфигурации очередей в Bitrix Framework.

Если в .settings.php уже есть другие настройки, секцию messenger нужно добавить в существующий массив конфигурации.

return [
'messenger' => [
'value' => [
'queues' => [
// Очередь для синхронизации событий с Google Calendar.
'google_event_sync' => [
// Класс обработчика сообщений этой очереди.
'handler' =>

\My\Module\Internals\Integration\My\Module\MessageHandler\GoogleEventSyncMessageHandler::class,


],
],
],

// Запрещаем изменять настройки через API.
'readonly' => true,
],
];

Ключ google_event_sync — идентификатор очереди. Его будем использовать дальше, когда отправим сообщение на обработку. Параметр handler связывает очередь с GoogleEventSyncMessageHandler: сообщения из этой очереди Bitrix Framework будет передавать этому обработчику. 

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

Включаем очереди на уровне проекта

В предыдущем разделе мы зарегистрировали очередь google_event_sync в конфигурации модуля. Теперь настроим механизм очередей на уровне всего проекта.

Глобальные настройки находятся в файле /bitrix/.settings.php. Для нашего сценария нужно указать режим обработки и брокер default, в котором будут храниться сообщения. 

// цитата // Если в /bitrix/.settings.php уже есть другие настройки, секцию messenger нужно добавить в существующий массив конфигурации.

В статье будем использовать консольный режим cli: на последнем шаге вручную запустим обработку очереди и проверим результат в логе.

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

Параметр run_mode определяет способ обработки сообщений. Значение cli означает, что их будет забирать консольный воркер, который мы запустим позже командой messenger:consume.

В brokers перечислены хранилища сообщений. Сейчас Bitrix Framework поддерживает брокер типа db, который сохраняет сообщения в базе данных. Брокер с именем default используется автоматически, если для конкретной очереди не указан другой. Для нашей очереди google_event_sync отдельный брокер мы не задавали, поэтому сообщения попадут именно сюда.

Очередь и обработчик настроены. Теперь система знает две вещи: очередь google_event_sync должен обслуживать GoogleEventSyncMessageHandler, а сообщения этой очереди нужно хранить через брокер default. 

Отправляем сообщение в нужном месте

Сейчас нужно поставить задачу на синхронизацию в тот момент, когда событие Битрикс24 уже изменено и его данные можно передать на фоновую обработку. В этом месте создаём объект GoogleEventSyncMessage и передаём в него данные события.

В контексте нашего примера мы предполагаем, что $eventId и $externalCalendarId уже получены в коде, который обрабатывает изменение события:

$message = new GoogleEventSyncMessage(
$eventId,
$externalCalendarId,
'update'
);

Затем отправляем сообщение в очередь google_event_sync методом send():

$message->send('google_event_sync');

Идентификатор в send() должен совпадать с именем очереди, которое мы указали в .settings.php модуля. В предыдущем разделе мы зарегистрировали именно google_event_sync и связали её с GoogleEventSyncMessageHandler.

После вызова send() Bitrix Framework сериализует данные GoogleEventSyncMessage и передаёт сообщение брокеру default. Основной код может продолжить работу, а сообщение будет ждать обработки в брокере.

В нашем сценарии полный путь теперь выглядит так:

изменили событие → создали GoogleEventSyncMessage → отправили в google_event_sync → брокер default сохранил сообщение

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

На предыдущем шаге мы отправили GoogleEventSyncMessage в очередь google_event_sync. Сообщение уже хранится у брокера и ждёт обработки.

В глобальной конфигурации мы выбрали режим cli, поэтому запустим консольный воркер и укажем ему нашу очередь:

php bitrix.php messenger:consume google_event_sync

Команда messenger:consume запускает обработку очередей. Воркер получает доступные сообщения из брокера и передаёт их связанным с очередями обработчикам. Если указать идентификатор очереди после команды, воркер будет обрабатывать именно её.

Сейчас воркер найдёт GoogleEventSyncMessage в очереди google_event_sync и передаст его в GoogleEventSyncMessageHandler. Метод process() выполнится и запишет в лог данные, которые мы передали в сообщении:

Google Calendar sync: event 123, calendar google-calendar-42, operation update

Так мы проверяем весь путь сообщения:

`GoogleEventSyncMessage → 

google_event_sync → 

брокер default → 

GoogleEventSyncMessageHandler → лог`

Открываем лог, настроенный для AddMessage2Log(), и проверяем, что запись появилась.

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

Послесловие

В реальной интеграции с Google Calendar стоит учитывать, что внешний сервис может временно не ответить. Для таких случаев у очереди можно настроить retry_strategy — правила повторной обработки сообщения с количеством попыток и задержками между ними.

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

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