Setting up data exchange via message queues (RabbitMQ) in 1C-Bitrix

Configuring Data Exchange via Message Queues (RabbitMQ) for 1C-Bitrix

Synchronous data transfer between 1C-Bitrix and external systems directly inside an event handler is unreliable. If the external system is unavailable, the user receives an error or the event is lost. RabbitMQ solves this: 1C-Bitrix publishes a message to the queue and immediately returns control; a separate worker picks up the message and delivers it to the target system.

When RabbitMQ is Needed

• The external system is periodically unavailable (maintenance, unstable connection).
• Event volume is irregular: thousands per minute at peak, dozens under normal load.
• Delivery guarantee is required: every message must be processed exactly once.
• Multiple consumers must each receive a copy of the same event (fan-out).

Publishing Messages from 1C-Bitrix

For RabbitMQ integration from PHP, use the php-amqplib/php-amqplib library. Install via Composer in /local/:

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

Publisher class:

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, // message survives broker restart
             'content_type'  => 'application/json']
        );
        $channel->basic_publish($msg, $exchange, $routingKey);
        $channel->close();
    }
}

Publishing on 1C-Bitrix Events

// In 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'],
    ]);
});

Consumer Worker

The worker is a separate PHP process (or multiple processes) managed by Supervisor:

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

$connection = new AMQPStreamConnection(/* connection params */);
$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); // at most 5 unacknowledged messages per worker
$channel->basic_consume('order.processor', '', false, false, false, false,
    function($msg) {
        $data = json_decode($msg->getBody(), true);
        try {
            OrderSyncHandler::process($data);
            $msg->ack(); // acknowledge successful processing
        } catch (\Throwable $e) {
            $msg->nack(false, true); // return to queue for retry
        }
    }
);

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

Supervisor config (/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

Messages that fail to process after N attempts are moved to a DLQ (Dead Letter Queue) for manual review. Configured at queue declaration time:

$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], // 1-hour TTL
]);

DLQ monitoring is done via the RabbitMQ Management UI (port 15672) or via alerts triggered when the queue depth grows.

Setting up RabbitMQ for a single event type with a worker takes 1–2 working days, including Supervisor configuration and monitoring.