Налаштування обміну даними через черги повідомлень (RabbitMQ) 1С-Бітрікс

Налаштування обміну даними через черги повідомлень (RabbitMQ) для 1С-Бітрікс

Синхронна передача даних між 1С-Бітрікс та зовнішніми системами прямо в обробнику події — ненадійна. Якщо зовнішня система недоступна, користувач отримує помилку або подія втрачається. RabbitMQ вирішує задачу: 1С-Бітрікс публікує повідомлення в чергу і негайно повертає управління, окремий воркер забирає повідомлення і доставляє в цільову систему.

Коли потрібен RabbitMQ

• Зовнішня система periodically недоступна (обслуговування, нестабільний канал).
• Обсяг подій нерегулярний: у піку — тисячі на хвилину, у звичайний час — десятки.
• Потрібна гарантія доставки: кожне повідомлення має бути оброблено рівно один раз.
• Кілька споживачів повинні отримувати копії однієї події (fan-out).

Публікація повідомлень з 1С-Бітрікс

Для роботи з RabbitMQ з PHP — бібліотека php-amqplib/php-amqplib. Встановлюємо через Composer у /local/:

cd /local && composer require php-amqplib/php-amqplib

Клас-видавець:

use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;

class RabbitMQPublisher {
    private static ?AMQPStreamConnection $connection = null;

    private static function getConnection(): AMQPStreamConnection {
        if (!self::$connection || !self::$connection->isConnected()) {
            self::$connection = new AMQPStreamConnection(
                COption::GetOptionString('site', 'rmq_host', 'localhost'),
                COption::GetOptionInt('site', 'rmq_port', 5672),
                COption::GetOptionString('site', 'rmq_user', 'guest'),
                COption::GetOptionString('site', 'rmq_pass', 'guest'),
                COption::GetOptionString('site', 'rmq_vhost', '/')
            );
        }
        return self::$connection;
    }

    public static function publish(string $exchange, string $routingKey, array $payload): void {
        $channel = self::getConnection()->channel();
        $channel->exchange_declare($exchange, 'topic', false, true, false);

        $msg = new AMQPMessage(
            json_encode($payload),
            ['delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT, // повідомлення не втрачається при рестарті
             'content_type'  => 'application/json']
        );
        $channel->basic_publish($msg, $exchange, $routingKey);
        $channel->close();
    }
}

Публікація при подіях 1С-Бітрікс

// В init.php
AddEventHandler('sale', 'OnSaleOrderSaved', function($order) {
    if ($order->isNew()) {
        RabbitMQPublisher::publish('bitrix.events', 'order.created', [
            'order_id'   => $order->getId(),
            'user_id'    => $order->getUserId(),
            'total'      => $order->getPrice(),
            'timestamp'  => time(),
        ]);
    }
});

AddEventHandler('catalog', 'OnAfterIBlockElementAdd', function($fields) {
    RabbitMQPublisher::publish('bitrix.events', 'product.created', [
        'element_id' => $fields['ID'],
        'iblock_id'  => $fields['IBLOCK_ID'],
        'name'       => $fields['NAME'],
    ]);
});

Воркер-споживач

Воркер — окремий PHP-процес (або кілька), запущений через Supervisor:

// worker.php
require '/local/vendor/autoload.php';
require $_SERVER['DOCUMENT_ROOT'] . '/bitrix/modules/main/include/prolog_before.php';

$connection = new AMQPStreamConnection(/* параметри */);
$channel    = $connection->channel();
$channel->queue_declare('order.processor', false, true, false, false);
$channel->queue_bind('order.processor', 'bitrix.events', 'order.created');

$channel->basic_qos(null, 5, null); // не більше 5 необроблених повідомлень на воркера
$channel->basic_consume('order.processor', '', false, false, false, false,
    function($msg) {
        $data = json_decode($msg->getBody(), true);
        try {
            OrderSyncHandler::process($data);
            $msg->ack(); // підтверджуємо успішну обробку
        } catch (\Throwable $e) {
            $msg->nack(false, true); // повертаємо в чергу для повторної спроби
        }
    }
);

while ($channel->is_consuming()) {
    $channel->wait();
}

Supervisor-конфіг (/etc/supervisor/conf.d/bitrix_worker.conf):

[program:bitrix_order_worker]
command=php /var/www/bitrix.loc/local/workers/order_worker.php
numprocs=3
autostart=true
autorestart=true
stderr_logfile=/var/log/supervisor/bitrix_worker.err.log

Dead Letter Queue

Повідомлення, які не вдалося обробити N разів, переміщуються в DLQ (Dead Letter Queue) для ручного розбору. Налаштовується при оголошенні черги:

$channel->queue_declare('order.processor', false, true, false, false, false, [
    'x-dead-letter-exchange'    => ['S', 'bitrix.dlx'],
    'x-dead-letter-routing-key' => ['S', 'order.failed'],
    'x-message-ttl'             => ['I', 3600000], // TTL 1 година
]);

Моніторинг DLQ — через Management UI RabbitMQ (порт 15672) або через сповіщення при зростанні черги.

Налаштування RabbitMQ для одного типу подій з воркером — 1–2 робочих дні, включаючи Supervisor та моніторинг.