Перейти к основному содержимому
Версия: 1.0.5

Продюсеры

Producer публикует сообщения в RabbitMQ через потокозависимый блокирующий канал. Один экземпляр привязывается к конкретному exchange и очереди при создании и переиспользует эту привязку при каждом вызове publish.

Создание продюсера​

from django_rmq.producer import Producer
from django_rmq.queues.queue_config import QueueConfig

# Default exchange, queue named by string
producer: Producer = Producer(queue='orders')

# Named exchange, no fixed queue (exchange-only mode)
producer: Producer = Producer(exchange='events', queue='')

# With a QueueConfig (carries dead-letter settings)
queue_config: QueueConfig = QueueConfig(
    name='orders',
    dead_letter_exchange='dlx-orders',
    dead_letter_routing_key='dlq-orders',
)
producer: Producer = Producer(queue=queue_config)

# Explicit connection alias (required when multiple connections are configured)
producer: Producer = Producer(queue='orders', using='payments')

Параметры

ПараметрТипПо умолчаниюОписание
exchangestr''Имя exchange. Пустая строка означает exchange по умолчанию (direct).
queueQueueConfig | str''Очередь для публикации. Используется также как ключ маршрутизации по умолчанию. Передайте пустую строку для публикации только в exchange с явным routing_key.
usingstr | NoneNoneПсевдоним подключения из RABBITMQ_CONNECTIONS. Не указывайте, если настроено одно подключение.

Producer можно создавать на уровне модуля — при __init__ подключение не открывается. Соединение и канал получаются лениво при первом вызове publish.

Публикация сообщения​

import json
from django_rmq.producer import Producer

producer: Producer = Producer(queue='orders')

# String body — encoded to UTF-8 bytes automatically
producer.publish(body='{"order_id": 42}')

# Bytes body — passed through unchanged
producer.publish(body=b'{"order_id": 42}')

# Override routing key (exchange-only mode)
Producer(exchange='events', queue='').publish(
    body=json.dumps({'event': 'order.created'}),
    routing_key='payments.created',
)

Сигнатура publish

def publish(
    self,
    body: str | bytes,
    routing_key: str | None = None,
    properties: BasicProperties | None = None,
) -> None: ...
  • body — str кодируется в UTF-8; bytes отправляется без изменений.
  • routing_key — если не указан, используется self.queue (имя очереди).
  • properties — свойства AMQP-сообщения; см. Custom properties ниже.

Постоянная доставка (Persistent Delivery)​

Каждое сообщение, опубликованное через Producer, является постоянным (delivery_mode=2). Это означает, что брокер записывает сообщение на диск, и оно переживает перезапуск брокера при условии, что сама очередь является durable.

Если передать объект BasicProperties без указания delivery_mode, продюсер принудительно устанавливает DeliveryMode.Persistent. Единственный способ отправить нестойкое сообщение — явно указать delivery_mode=DeliveryMode.Transient в объекте свойств.

from pika import DeliveryMode
from pika.spec import BasicProperties
from django_rmq.producer import Producer

# Explicitly transient (unusual — only if you know what you are doing)
Producer(queue='ephemeral').publish(
    body='ping',
    properties=BasicProperties(delivery_mode=DeliveryMode.Transient.value),
)

Ленивое объявление очереди​

Очередь объявляется на брокере один раз, при первом вызове publish. После этого флаг _is_queue_declared предотвращает повторные объявления.

Режим объявления зависит от типа аргумента queue:

Тип queueРежим объявленияЭффект
str (непустая)passive=TrueПроверяет наличие очереди; выбрасывает ChannelClosedByBroker (404), если очередь не существует.
QueueConfigАктивное объявление с durable + argumentsСоздаёт очередь, если отсутствует; идемпотентна при повторном вызове с теми же параметрами.
'' (пустая строка)Пропускается полностьюИспользуйте явный routing_key; продюсер сразу переходит к basic_publish.

Используйте QueueConfig, когда очередь имеет dead-letter аргументы или когда нужно, чтобы продюсер создал очередь, если она ещё не существует. Используйте обычную строку, когда существование очереди гарантировано (объявлена setup-функцией или другим сервисом).

Надёжность: confirmation, mandatory и ретраи​

Подтверждения публикации включены на каждом канале продюсера (confirm_delivery()). После каждого basic_publish брокер подтверждает получение сообщения. Если ни одна очередь не соответствует ключу маршрутизации, брокер возвращает сообщение, и pika выбрасывает UnroutableError.

mandatory=True всегда установлен. Сообщение, опубликованное в exchange без подходящей привязки, не отбрасывается молча — брокер возвращает его, и pika выбрасывает UnroutableError.

from pika.exceptions import UnroutableError
from django_rmq.producer import Producer

producer: Producer = Producer(exchange='', queue='')

try:
    producer.publish(body=b'nowhere', routing_key='no-such-queue')
except UnroutableError:
    # The broker had no queue for this routing key.
    ...

Повтор при временных ошибках. Когда publish сталкивается с AMQP-ошибкой (AMQPConnectionError, ConnectionClosed, ChannelClosed, ChannelClosedByBroker, StreamLostError, ConnectionResetError), он сбрасывает кэшированные канал и соединение, затем выполняет повтор ровно один раз. Если повтор также завершается неудачей, исключение передаётся вызывающему коду.

Невосстанавливаемые исключения (например, ValueError, UnroutableError) не повторяются и немедленно передаются вызывающему коду.

Продюсер как декоратор​

Экземпляр Producer может использоваться как декоратор функции. Декоратор автоматически публикует возвращаемое значение функции после её выполнения.

import json
from django_rmq.producer import Producer

order_producer: Producer = Producer(queue='orders')

@order_producer
def create_order(order_id: int) -> str:
    # business logic here
    return json.dumps({'order_id': order_id})

# Calling create_order publishes the returned JSON and also returns it.
result: str = create_order(order_id=42)

Контракт возвращаемого значения:

Тип возвратаПоведение
str или bytesПубликуется и возвращается вызывающему коду.
NoneПубликация пропускается; возвращается None.
Любой другой типНемедленно выбрасывается TypeError.

Строгость в отношении других типов намеренна — молчаливая сериализация dict или list скрывала бы ошибки контракта.

Публикация только в exchange​

Для публикации в exchange без фиксированной очереди (например, fanout или topic exchange) установите queue='' и всегда передавайте routing_key в publish:

from django_rmq.producer import Producer

events_producer: Producer = Producer(exchange='domain.events', queue='')

events_producer.publish(
    body=b'{"type": "order.shipped", "order_id": 7}',
    routing_key='order.shipped',
)

В этом режиме объявление очереди не выполняется.

Custom properties​

Передайте экземпляр BasicProperties, чтобы переопределить тип контента, заголовки, correlation ID или другие AMQP-свойства. delivery_mode принудительно устанавливается в Persistent, если вы явно не задаёте его.

from pika.spec import BasicProperties
from django_rmq.producer import Producer

Producer(queue='orders').publish(
    body=b'{"order_id": 1}',
    properties=BasicProperties(
        content_type='application/json',
        correlation_id='req-abc-123',
        headers={'x-source': 'checkout-service'},
    ),
)

Смотрите также​