Брокеры сообщений и Celery #

1. Что такое Celery #

Что такое Celery #

Celery — это Python-фреймворк для выполнения задач вне основного процесса приложения: в фоне, по расписанию или на отдельных серверах.

Приложение не выполняет тяжёлую операцию непосредственно. Оно отправляет сообщение с заданием в очередь, после чего отдельный процесс Celery Worker получает и выполняет его. Celery называет такую единицу работы task.

Клиент
   │
   ▼
FastAPI / Django
   │ отправляет задачу
   ▼
Broker: Redis / RabbitMQ
   │
   ▼
Celery Worker
   │ выполняет задачу
   ▼
Result Backend — при необходимости

Пример задачи #

Пользователь регистрируется, после чего ему нужно отправить письмо.

Без Celery:

@app.post("/register")
def register(data: RegisterData):
    user = create_user(data)

    send_email(user.email)  # запрос ждёт отправки письма

    return {"user_id": user.id}

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

С Celery:

@app.post("/register")
def register(data: RegisterData):
    user = create_user(data)

    send_email.delay(user.email)

    return {"user_id": user.id}

Метод .delay() не отправляет письмо самостоятельно. Он помещает сообщение о задаче в брокер, после чего свободный Worker выполняет её отдельно.

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

1. Producer #

Приложение, которое создаёт задачу:

send_email.delay("user@example.com")

Producer может быть:

FastAPI;
Django;
CLI-команда;
другой Worker;
периодический планировщик Celery Beat.

2. Broker #

Broker хранит и передаёт сообщения с задачами:

Приложение → Broker → Worker

Часто используют:

• RabbitMQ;

• Redis;

• Amazon SQS.

RabbitMQ является полноценным брокером сообщений; Redis в Celery может использоваться и как broker, и как result backend.

Broker не выполняет задачу. Он только передаёт сообщение Worker-у.

Пример содержимого сообщения:

{
  "task": "tasks.send_email",
  "args": ["user@example.com"],
  "kwargs": {}
}

3. Worker #

Worker — отдельный процесс, который постоянно получает задачи из очереди и выполняет их:

celery -A tasks worker --loglevel=INFO

Можно запустить несколько Worker-ов:

                  ┌── Worker 1
Redis/RabbitMQ ───┼── Worker 2
                  ├── Worker 3
                  └── Worker 4

Это позволяет распределять задачи между несколькими процессами или машинами. Celery поддерживает работу с несколькими Worker-ами и горизонтальное масштабирование.

4. Result Backend #

Result Backend хранит состояние и результат задачи:

PENDING
STARTED
SUCCESS
FAILURE
RETRY
REVOKED

Например:

result = add.delay(10, 20)

print(result.id)
print(result.status)
print(result.get())

В качестве backend могут использоваться Redis, SQL-база, RPC и другие поддерживаемые хранилища. Result Backend необязателен: если результат выполнения не нужен, его можно не настраивать.

Минимальный пример #

Установка:

pip install celery redis

Файл tasks.py:

from celery import Celery

app = Celery(
    "tasks",
    broker="redis://localhost:6379/0",
    backend="redis://localhost:6379/1",
)


@app.task
def add(left: int, right: int) -> int:
    return left + right

Запуск Worker:

celery -A tasks worker --loglevel=INFO

Отправка задачи:

from tasks import add

result = add.delay(10, 20)

print(result.id)

Происходит следующее:

1. add.delay(10, 20) создаёт сообщение
2. Сообщение попадает в Redis
3. Worker получает сообщение
4. Worker вызывает add(10, 20)
5. Результат 30 сохраняется в Redis backend

Получение результата:

print(result.get(timeout=10))

Результат:

30

В веб-приложении обычно не следует вызывать result.get() прямо внутри HTTP-запроса: это снова заставит запрос ждать завершения фоновой задачи.

Что обычно выполняют через Celery #

Типичные задачи:

отправка электронной почты;
обработка загруженных файлов;
генерация PDF и отчётов;
изменение размера изображений;
импорт и экспорт данных;
обращения к медленным внешним API;
отправка уведомлений;
периодическая очистка данных;
массовая обработка записей;
повторные попытки после временных ошибок.

Например:

@app.task
def generate_report(report_id: int) -> None:
    report = load_report(report_id)
    file_path = create_pdf(report)
    upload_to_storage(file_path)

HTTP-запрос только запускает работу:

task = generate_report.delay(report_id)

return {
    "task_id": task.id,
    "status": "processing",
}

Повторные попытки #

Celery позволяет повторить задачу после временной ошибки:

from celery import Celery

app = Celery("tasks")


@app.task(
    autoretry_for=(ConnectionError,),
    retry_backoff=True,
    retry_kwargs={"max_retries": 5},
)
def call_external_api() -> None:
    send_request()

Например, если внешний сервис временно недоступен, Celery повторит операцию позднее. Celery поддерживает автоматический retry через параметры декоратора задачи.

Задачи с повторными попытками желательно делать идемпотентными:

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

Например, вместо безусловного создания платежа:

create_payment(order_id)

нужно проверять уникальный идентификатор операции:

if payment_exists(idempotency_key):
    return

create_payment(
    order_id=order_id,
    idempotency_key=idempotency_key,
)

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

Периодические задачи и Celery Beat #

Celery Beat — планировщик, который отправляет задачи по расписанию:

Celery Beat → Broker → Celery Worker

Сам Beat не выполняет задачу. Её выполняет Worker.

Пример:

app.conf.beat_schedule = {
    "clear-expired-tokens": {
        "task": "tasks.clear_expired_tokens",
        "schedule": 3600.0,
    },
}

Задача будет отправляться каждый час.

Запуск:

celery -A tasks beat --loglevel=INFO

Celery Beat поддерживает интервальные и периодические расписания; в каждый момент обычно должен работать один экземпляр планировщика для конкретного расписания, иначе задачи могут создаваться повторно.

Несколько очередей #

Разные типы задач можно разделить:

emails  → Worker для писем
reports → Worker для отчётов
images  → Worker для изображений

Например:

app.conf.task_routes = {
    "tasks.send_email": {"queue": "emails"},
    "tasks.generate_report": {"queue": "reports"},
}

Запуск отдельных Worker-ов:

celery -A tasks worker -Q emails --loglevel=INFO

celery -A tasks worker -Q reports --loglevel=INFO

Это позволяет независимо масштабировать разные типы работы.

Celery не создаёт отдельный поток в FastAPI #

Важно различать:

async def

и:

Celery task

async def позволяет эффективно ожидать I/O внутри процесса приложения.

Celery передаёт работу другому процессу, Worker-у или серверу:

FastAPI process
      │
      ▼
Broker
      │
      ▼
Celery worker process

Поэтому Celery применяют, когда работа должна:

• продолжаться после завершения HTTP-запроса;

• переживать перезапуск веб-приложения при корректно настроенном брокере;

• выполняться отдельными Worker-ами;

• распределяться между машинами;

• повторяться после ошибки;

• запускаться по расписанию.

Недостатки Celery #

Celery добавляет инфраструктурную сложность:

нужен брокер;
нужно запускать Worker-ы;
нужно контролировать очереди;
нужны мониторинг и логирование;
нужно обрабатывать повторные задачи;
необходимо следить за зависшими задачами;
нужно проектировать идемпотентность;
нужно управлять результатами и сроком их хранения.

Поэтому Celery не требуется для любой маленькой фоновой операции.

Итог #

Celery — система распределённого выполнения фоновых задач.

Приложение создаёт задачу.
Broker передаёт её.
Worker выполняет.
Result Backend при необходимости сохраняет результат.
Celery Beat создаёт задачи по расписанию.

Основная схема:

FastAPI / Django
        │
        │ task.delay(...)
        ▼
Redis / RabbitMQ
        │
        ▼
Celery Worker
        │
        ├── отправляет письмо
        ├── создаёт отчёт
        ├── обрабатывает файл
        └── вызывает внешний API

2. Для каких целей применяется Celery? #

Для каких целей применяется Celery #

Celery применяют для выполнения задач вне основного процесса веб-приложения и распределения работы между отдельными Worker-процессами или серверами.

Общая схема:

FastAPI / Django
        │
        │ отправляет сообщение о задаче
        ▼
Redis / RabbitMQ / другой broker
        │
        ▼
Celery Worker
        │
        └── выполняет задачу

Приложение публикует сообщение в очередь, брокер доставляет его Worker-у, а Worker выполняет функцию, зарегистрированную как Celery task. Несколько Worker-ов позволяют распределять работу и горизонтально масштабировать обработку.

1. Длительные фоновые операции #

Celery используют, когда HTTP-запрос не должен ждать завершения операции:

@app.post("/reports")
async def create_report(report_id: int):
    generate_report.delay(report_id)

    return {
        "report_id": report_id,
        "status": "processing",
    }

Типичные задачи:

генерация PDF и отчётов;
обработка изображений и видео;
импорт и экспорт данных;
архивация файлов;
массовая обработка записей;
обращение к медленным внешним API.

Пользователь получает ответ сразу, а обработка продолжается отдельным Worker-ом.

2. Отправка писем и уведомлений #

Отправка электронной почты, push-уведомлений или сообщений во внешние системы может быть медленной либо временно завершаться ошибкой:

@app.task
def send_registration_email(user_id: int) -> None:
    user = load_user(user_id)
    email_client.send(user.email)

Запуск:

send_registration_email.delay(user.id)

Это отделяет бизнес-операцию от доступности SMTP-сервера или стороннего провайдера.

Важно сначала сохранить основное состояние в БД, а затем надёжно опубликовать задачу. Простая последовательность:

save_user()
send_email.delay(user.id)

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

3. Повторные попытки после временных ошибок #

Celery поддерживает повторный запуск задач:

@app.task(
    autoretry_for=(ConnectionError, TimeoutError),
    retry_backoff=True,
    retry_kwargs={"max_retries": 5},
)
def request_external_service(order_id: int) -> None:
    external_client.send_order(order_id)

Повторные попытки полезны при временной недоступности:

внешнего API;
SMTP-сервера;
объектного хранилища;
сетевого ресурса;
удалённого микросервиса.

При вызове retry Celery отправляет новое сообщение с тем же идентификатором задачи и сохраняет исходную очередь; автоматический retry настраивается через autoretry_for.

Задача должна быть идемпотентной, поскольку повторное выполнение не должно создавать второй платёж, второй заказ или повторную запись:

@app.task
def process_payment(payment_id: str) -> None:
    if payment_already_processed(payment_id):
        return

    execute_payment(payment_id)

Документация Celery прямо связывает позднее подтверждение сообщения через acks_late с требованием идемпотентности задачи.

4. Параллельная обработка больших объёмов данных #

Большой набор данных можно разделить на независимые части:

1 000 000 записей
        ↓
батчи по 1 000 записей
        ↓
1 000 Celery-задач
        ↓
несколько Worker-ов

Пример:

@app.task
def process_batch(record_ids: list[int]) -> int:
    processed = 0

    for record_id in record_ids:
        process_record(record_id)
        processed += 1

    return processed

Celery предоставляет Canvas-примитивы для организации подобных процессов:

group  — параллельный запуск задач;
chain  — последовательное выполнение;
chord  — группа задач с финальной задачей;
chunks — разбиение данных на части.

Эти механизмы входят в официальный Canvas API Celery.

Пример параллельной обработки:

from celery import group

workflow = group(
    process_batch.s(batch)
    for batch in batches
)

workflow.apply_async()

5. Выполнение цепочек задач #

Иногда следующая задача должна запускаться после предыдущей:

получить файл
    ↓
проверить файл
    ↓
обработать данные
    ↓
сформировать отчёт
    ↓
уведомить пользователя

Это можно представить через chain:

from celery import chain

workflow = chain(
    download_file.s(file_id),
    validate_file.s(),
    process_file.s(),
    generate_report.s(),
)

workflow.apply_async()

Для параллельной обработки с итоговой агрегацией используется chord:

from celery import chord

chord(
    process_batch.s(batch)
    for batch in batches
)(
    aggregate_results.s(job_id)
)

Callback у chord запускается после завершения группы задач; для работы chord обычно требуется подходящий result backend.

6. Периодические задачи #

Celery Beat используется как планировщик:

Celery Beat
     │ создаёт задачи по расписанию
     ▼
Broker
     ▼
Celery Worker

Пример:

app.conf.beat_schedule = {
    "delete-expired-sessions": {
        "task": "tasks.delete_expired_sessions",
        "schedule": 3600.0,
    },
}

Типичные периодические операции:

очистка истёкших токенов;
формирование ежедневных отчётов;
синхронизация с внешней системой;
проверка неоплаченных заказов;
удаление временных файлов;
обновление кеша.

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

7. Разделение задач по очередям #

Разные виды задач можно направлять в разные очереди:

emails     → Worker-ы для писем
reports    → Worker-ы для отчётов
payments   → Worker-ы для платёжных операций
images     → Worker-ы для обработки изображений

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

app.conf.task_routes = {
    "tasks.send_email": {
        "queue": "emails",
    },
    "tasks.generate_report": {
        "queue": "reports",
    },
    "tasks.resize_image": {
        "queue": "images",
    },
}

Запуск Worker-а только для определённой очереди:

celery -A application worker -Q reports --loglevel=INFO

Celery поддерживает маршрутизацию задач в разные очереди через task_routes и конфигурацию очередей.

Это позволяет:

независимо масштабировать Worker-ы;
не блокировать быстрые задачи тяжёлыми;
задавать разную concurrency;
использовать разные лимиты ресурсов;
разделять задачи по приоритету и назначению.

8. Отложенный запуск #

Задачу можно запланировать на определённое время или через интервал:

send_reminder.apply_async(
    args=[order_id],
    countdown=300,
)

Либо:

send_reminder.apply_async(
    args=[order_id],
    eta=scheduled_datetime,
)

Это подходит для коротких отложенных операций:

отправить напоминание через пять минут;
повторить проверку позднее;
снять временную блокировку;
проверить статус внешней операции.

Для большого количества задач, запланированных далеко в будущем, зачастую надёжнее хранить расписание в БД и периодически выбирать готовые к запуску записи, а не удерживать множество ETA-задач внутри инфраструктуры Celery.

9. Сохранение статуса и результата задачи #

При настроенном result backend можно получить:

идентификатор задачи;
текущее состояние;
результат;
информацию об ошибке.

Пример:

result = generate_report.delay(report_id)

return {
    "task_id": result.id,
}

Позже:

from celery.result import AsyncResult

result = AsyncResult(task_id)

return {
    "status": result.status,
    "result": result.result if result.successful() else None,
}

Broker и result backend выполняют разные задачи:

Broker         — передаёт сообщения Worker-ам;
Result Backend — хранит состояния и результаты.

Celery поддерживает различные брокеры и result backend; например, RabbitMQ обычно используется как broker, а Redis может выступать и broker, и backend.

Для длительных бизнес-процессов часто лучше хранить пользовательский статус отдельно в своей БД:

PENDING
PROCESSING
COMPLETED
PARTIALLY_COMPLETED
FAILED

Celery-состояние описывает техническое выполнение задачи, а запись в доменной БД — состояние бизнес-операции.

Когда Celery оправдан #

Celery обычно полезен, когда необходимо хотя бы одно из следующего:

выполнить работу после завершения HTTP-запроса;
распределить обработку между процессами или серверами;
буферизовать всплески нагрузки;
повторять временно неудачные операции;
разделять задачи по очередям;
выполнять задачи по расписанию;
строить цепочки и группы фоновых операций;
масштабировать Worker-ы независимо от API.

Когда Celery может быть избыточен #

Celery не всегда нужен для:

очень короткой некритичной операции;
обычного async I/O внутри текущего запроса;
задачи, результат которой необходим до ответа клиенту;
небольшого проекта без потребности в очередях и Worker-ах.

Например, asyncio помогает конкурентно ожидать I/O внутри процесса, но не предоставляет отдельную устойчивую очередь, независимые Worker-ы, маршрутизацию и повторную доставку.

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

Итог #

Celery применяется как инфраструктура фонового и распределённого выполнения:

веб-приложение
      │
      │ публикует задачу
      ▼
broker
      │
      ▼
один или несколько Worker-ов
      │
      ├── выполняют работу
      ├── повторяют временные ошибки
      ├── обрабатывают задачи параллельно
      └── сохраняют результат при необходимости

Celery особенно оправдан для длительных, повторяемых, распределяемых и независимо масштабируемых задач, которые не должны выполняться внутри HTTP-запроса.

3. Для каких целей применяется RabbitMQ? #

Что такое RabbitMQ #

RabbitMQ — брокер сообщений. Он принимает сообщения от отправителей, маршрутизирует их через exchange, помещает в очереди и передаёт приложениям-получателям.

Producer
   │ публикует сообщение
   ▼
Exchange
   │ маршрутизирует
   ▼
Queue
   │ хранит до обработки
   ▼
Consumer

RabbitMQ не выполняет бизнес-задачу самостоятельно. Он хранит и доставляет сообщения, а обработку выполняют Consumer-ы: Celery Worker, микросервис или отдельное приложение. RabbitMQ поддерживает очереди задач, публикацию событий нескольким получателям, маршрутизацию и модель request/reply.

1. Асинхронная обработка #

RabbitMQ позволяет отправителю не ждать завершения операции:

API
 │ создаёт сообщение
 ▼
RabbitMQ
 │
 ▼
Worker обрабатывает позднее

Например:

HTTP-запрос
    ↓
создание задания
    ↓
публикация сообщения
    ↓
ответ клиенту: 202 Accepted
    ↓
фоновая обработка Worker-ом

Это применяется для:

• отправки писем и уведомлений;

• генерации отчётов;

• обработки файлов;

• импорта данных;

• обращения к медленным внешним системам;

• массовой обработки записей.

RabbitMQ в таком сценарии отделяет создание задания от его выполнения.

2. Очередь фоновых задач #

Несколько Worker-ов могут читать одну очередь:

                   ┌── Worker 1
Producer → Queue ──┼── Worker 2
                   └── Worker 3

Каждое сообщение из рабочей очереди передаётся одному Consumer-у. Это позволяет распределять длительные задачи между несколькими Worker-ами — паттерн competing consumers.

Например:

10 000 батчей
      ↓
batch_processing_queue
      ↓
20 Worker-ов обрабатывают батчи параллельно

3. Буферизация всплесков нагрузки #

Допустим, API за минуту создаёт 10 000 задач, но Worker-ы способны обрабатывать только 2 000 задач в минуту.

Без очереди приложение может перегрузить обработчики или зависимые системы:

Producer ─────────────> Worker
10 000 сообщений        перегрузка

С RabbitMQ:

Producer
   │ быстро публикует
   ▼
Queue: 10 000 сообщений
   │ выдаёт постепенно
   ▼
Worker-ы

Очередь временно накапливает сообщения, а Consumer-ы забирают их в доступном темпе. При ручных подтверждениях параметр prefetch ограничивает количество сообщений, одновременно находящихся у Consumer-а без подтверждения, что помогает не перегружать обработчик.

RabbitMQ при этом не отменяет необходимость контролировать длину очередей, скорость Producer-ов, TTL и ресурсные лимиты.

4. Ослабление связанности компонентов #

Без брокера один сервис напрямую зависит от другого:

Order Service ──HTTP──> Notification Service

Если Notification Service недоступен, запрос может завершиться ошибкой или ожиданием тайм-аута.

Через RabbitMQ:

Order Service
      │ OrderCreated
      ▼
RabbitMQ
      │
      ▼
Notification Service

Order Service публикует сообщение и не обязан знать:

• где запущен получатель;

• сколько экземпляров Consumer-а работает;

• когда именно сообщение будет обработано;

• каким внутренним способом отправляется уведомление.

Это снижает временную связанность: отправитель и получатель необязательно должны быть доступны одновременно.

5. Передача событий нескольким системам #

Одно событие может понадобиться сразу нескольким сервисам:

                ┌──> Notification Queue
OrderCreated ───┼──> Analytics Queue
                ├──> Delivery Queue
                └──> Audit Queue

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

Это паттерн publish/subscribe. В отличие от рабочей очереди, где одно сообщение получает один Worker, при publish/subscribe одно опубликованное событие может быть доставлено нескольким независимым Consumer-ам через разные очереди.

6. Маршрутизация сообщений #

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

Producer
   │ routing_key = payment.succeeded
   ▼
Exchange
   ├──> notification_queue
   ├──> accounting_queue
   └──> audit_queue

Exchange определяет, в какие очереди направить сообщение, исходя из своих привязок и routing key.

Основные варианты маршрутизации:

direct  — точное совпадение routing key;
topic   — совпадение с шаблоном;
fanout  — отправка во все связанные очереди;
headers — маршрутизация по заголовкам.

Официальные руководства RabbitMQ отдельно рассматривают routing, topic routing и publish/subscribe как базовые модели взаимодействия.

Пример topic-маршрутизации:

payment.succeeded
payment.failed
order.created
order.cancelled

Подписка:

payment.*

получит события, связанные с платежами.

7. Подтверждение обработки сообщений #

При ручном подтверждении Consumer отправляет ack после успешной обработки:

RabbitMQ → Consumer: сообщение
Consumer выполняет операцию
Consumer → RabbitMQ: ACK
RabbitMQ удаляет сообщение из очереди

Если соединение или Consumer завершатся до подтверждения, неподтверждённое сообщение может быть повторно доставлено. Подтверждение означает, что Consumer принял ответственность за доставку.

Из этого следует важное требование:

Consumer должен быть идемпотентным.

Например:

def handle_payment(message: PaymentMessage) -> None:
    if processed_messages.exists(message.id):
        return

    process_payment(message)
    processed_messages.save(message.id)

RabbitMQ не следует воспринимать как механизм гарантированного выполнения «ровно один раз». При сбоях и повторной доставке сообщение может быть обработано повторно.

8. Подтверждение публикации #

Consumer acknowledgements подтверждают обработку со стороны получателя.

Для Producer существует отдельный механизм — publisher confirms:

Producer → RabbitMQ: сообщение
RabbitMQ → Producer: сообщение принято

Он позволяет отправителю понять, принял ли RabbitMQ публикацию. Для надёжной доставки обычно комбинируют:

publisher confirms;
durable queue;
persistent messages;
consumer acknowledgements;
идемпотентную обработку.

RabbitMQ подчёркивает, что долговечность очереди и сообщений настраивается отдельно: durable-очередь восстанавливается после перезапуска, а сохраняемые сообщения должны быть опубликованы как persistent.

При этом даже такая конфигурация требует корректной обработки ошибок и повторов на уровне приложения.

9. Повторные попытки и проблемные сообщения #

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

основная очередь
      ↓ ошибка
retry queue
      ↓ задержка
основная очередь

Если сообщение окончательно не обрабатывается, его можно направить в Dead Letter Exchange:

processing_queue
      │ reject / TTL / delivery limit
      ▼
Dead Letter Exchange
      ▼
dead_letter_queue

Сообщение может попасть в DLX, например, после отрицательного подтверждения без повторной постановки, истечения TTL, превышения длины очереди или лимита доставок quorum queue.

Dead-letter queue позволяет:

• не блокировать основную очередь;

• сохранить сведения об ошибке;

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

• повторно обработать его после исправления причины.

10. Отложенная обработка и TTL #

RabbitMQ поддерживает TTL для сообщений и очередей. После истечения TTL сообщение может быть удалено или отправлено в DLX при соответствующей настройке.

Это используют для:

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

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

11. Межсервисное взаимодействие #

В микросервисной архитектуре RabbitMQ применяют для асинхронных команд и событий:

Команда:
SendNotification
ReserveProduct
GenerateReport

Событие:
OrderCreated
PaymentSucceeded
UserRegistered

Различие по смыслу:

Команда:
конкретному получателю предлагается выполнить действие.

Событие:
фиксируется уже произошедший факт;
получателей может быть несколько.

Например:

Payment Service
      │ PaymentSucceeded
      ▼
RabbitMQ
      ├──> Order Service
      ├──> Notification Service
      └──> Analytics Service

12. Работа RabbitMQ с Celery #

При использовании Celery RabbitMQ обычно выступает в роли broker:

FastAPI / Django
      │ task.delay(...)
      ▼
RabbitMQ
      │ сообщение о задаче
      ▼
Celery Worker
      │
      ▼
выполнение функции

Разделение ответственности:

Celery:
определяет задачи;
запускает Worker-ы;
поддерживает retries, Canvas и расписание;
управляет выполнением задач.

RabbitMQ:
принимает сообщения;
маршрутизирует их;
хранит в очередях;
доставляет Worker-ам;
обрабатывает acknowledgements.

RabbitMQ не заменяет Celery, а Celery не является брокером сообщений.

13. Высокая доступность очередей #

Для очередей, которым требуется репликация и высокая доступность, RabbitMQ предлагает quorum queues. Они используют Raft и предназначены для надёжного хранения и быстрого выбора лидера при отказах. Официальная документация рекомендует рассматривать quorum queue как основной выбор, когда необходима реплицируемая высокодоступная очередь.

Но репликация увеличивает:

• использование диска;

• сетевой трафик;

• задержку подтверждения;

• инфраструктурную сложность.

Поэтому тип очереди выбирают исходя из требований к сохранности сообщений и доступности.

Где RabbitMQ не нужен #

RabbitMQ может быть избыточен, когда:

• операция должна завершиться до ответа клиенту;

• достаточно обычного вызова функции;

• компоненты находятся внутри одного процесса;

• нет потребности в буферизации или повторной доставке;

• требуется простой синхронный запрос-ответ;

• объём системы не оправдывает эксплуатацию брокера.

RabbitMQ поддерживает паттерн RPC, но для обычных синхронных межсервисных запросов HTTP или gRPC часто проще: RabbitMQ RPC требует correlation ID, reply queue, тайм-аутов, дедупликации и обработки потерянных ответов. Сам RabbitMQ приводит RPC как отдельный request/reply-паттерн, а не как основной режим работы брокера.

Итог #

RabbitMQ применяется для:

асинхронного выполнения работы;
очередей фоновых задач;
буферизации всплесков нагрузки;
распределения задач между Worker-ами;
ослабления связанности сервисов;
публикации событий нескольким системам;
маршрутизации сообщений;
повторной доставки после сбоев;
dead-letter и retry-механизмов;
горизонтального масштабирования Consumer-ов.

Основная роль RabbitMQ:

Надёжно передать сообщение
от Producer к одному или нескольким Consumer-ам,
не заставляя их работать синхронно и напрямую зависеть друг от друга.

4. Как разделить большую Celery task на несколько маленьких? | Celery chain, group, chord для разделения задач #

Основной принцип #

Большую Celery-задачу следует делить на самостоятельные идемпотентные этапы, каждый из которых можно выполнить отдельно, повторить после ошибки и распределить между Worker-ами.

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

Большая операция
      ↓
разбиение данных на батчи
      ↓
маленькие Celery tasks
      ↓
параллельная или последовательная обработка
      ↓
финализация результата

Canvas-примитивы #

Эти примитивы являются частью Celery Canvas и могут комбинироваться между собой.

chain: последовательные задачи #

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

загрузить файл
      ↓
проверить файл
      ↓
обработать данные
      ↓
сохранить результат

from celery import chain

workflow = chain(
    download_file.s(file_id),
    validate_file.s(),
    process_file.s(),
    save_result.s(),
)

workflow.apply_async()

По умолчанию результат предыдущей задачи передаётся первым аргументом следующей:

@app.task
def download_file(file_id: int) -> str:
    return f"/tmp/{file_id}.csv"


@app.task
def validate_file(file_path: str) -> str:
    validate(file_path)
    return file_path


@app.task
def process_file(file_path: str) -> dict:
    return parse_file(file_path)


@app.task
def save_result(parsed_data: dict) -> int:
    return repository.save(parsed_data)

download_file → "/tmp/42.csv"
                     ↓
validate_file("/tmp/42.csv")
                     ↓
process_file("/tmp/42.csv")

Celery определяет chain как последовательность signatures, где каждая задача вызывается после предыдущей.

Immutable signatures #

Иногда результат предыдущей задачи передавать не нужно. Тогда используется .si() вместо .s():

workflow = chain(
    create_order.s(order_id),
    send_notification.si(order_id),
)

В этом примере send_notification получит только order_id, а не результат create_order.

@app.task
def send_notification(order_id: int) -> None:
    ...

Immutable signature не принимает дополнительные аргументы, передаваемые предыдущей задачей Canvas.

group: параллельные задачи #

group применяют, когда задачи независимы друг от друга:

             ┌── batch 1
             ├── batch 2
Обработка ───┼── batch 3
             └── batch N

from celery import group

workflow = group(
    process_batch.s(batch_id)
    for batch_id in batch_ids
)

result = workflow.apply_async()

Все задачи группы могут выполняться параллельно разными Worker-процессами:

@app.task
def process_batch(batch_id: int) -> dict:
    batch = batch_repository.get(batch_id)

    processed = 0
    failed = 0

    for record_id in batch.record_ids:
        try:
            process_record(record_id)
            processed += 1
        except ProcessingError:
            failed += 1

    return {
        "batch_id": batch_id,
        "processed": processed,
        "failed": failed,
    }

Celery определяет group как набор задач, которые должны применяться параллельно.

Пример разбиения списка на батчи #

from collections.abc import Iterator, Sequence
from typing import TypeVar

T = TypeVar("T")


def split_into_batches(
    values: Sequence[T],
    batch_size: int,
) -> Iterator[list[T]]:
    if batch_size <= 0:
        raise ValueError("batch_size must be greater than zero")

    for offset in range(0, len(values), batch_size):
        yield list(values[offset : offset + batch_size])

Запуск:

batches = split_into_batches(record_ids, batch_size=1_000)

workflow = group(
    process_records.s(batch)
    for batch in batches
)

workflow.apply_async()

@app.task
def process_records(record_ids: list[int]) -> int:
    processed = 0

    for record_id in record_ids:
        process_record(record_id)
        processed += 1

    return processed

Однако передавать большие массивы идентификаторов через брокер не всегда эффективно. Для крупных объёмов лучше предварительно сохранить батчи в БД или объектном хранилище и передавать только batch_id.

workflow = group(
    process_batch.s(batch_id)
    for batch_id in batch_ids
)

Celery рекомендует передавать идентификаторы объектов, а актуальные данные повторно загружать внутри задачи: переданные ранее объекты или данные могут устареть до момента выполнения.

chord: параллельная обработка с финализацией #

chord состоит из:

header — группа параллельных задач
body   — callback после завершения всей группы

process_batch 1 ─┐
process_batch 2 ─┤
process_batch 3 ─┼──> aggregate_results
process_batch N ─┘

from celery import chord

workflow = chord(
    process_batch.s(batch_id)
    for batch_id in batch_ids
)(
    aggregate_results.s(job_id)
)

Результаты всех задач группы передаются в callback первым аргументом:

@app.task
def aggregate_results(
    batch_results: list[dict],
    job_id: int,
) -> dict:
    processed = sum(
        result["processed"]
        for result in batch_results
    )
    failed = sum(
        result["failed"]
        for result in batch_results
    )

    job_repository.complete(
        job_id=job_id,
        processed=processed,
        failed=failed,
    )

    return {
        "job_id": job_id,
        "processed": processed,
        "failed": failed,
    }

chord запускает callback только после завершения всех задач header-группы. Для chord требуется result backend, а задачи header и callback не должны игнорировать результаты. RPC result backend для chord не поддерживается.

Недостаток chord #

Синхронизация chord имеет стоимость. Официальная документация рекомендует избегать chord там, где итоговая барьерная синхронизация не нужна.

Особенно осторожно следует применять chord, когда:

сотни тысяч задач;
каждая задача возвращает большой результат;
callback получает огромный список результатов;
result backend испытывает значительную нагрузку.

Не следует возвращать из каждой задачи целые обработанные объекты:

# Плохо
return processed_records

Лучше возвращать компактную статистику:

return {
    "processed": 950,
    "failed": 50,
}

Либо сохранять частичные результаты в БД и возвращать только идентификатор:

return batch_result_id

chunks: автоматическое разбиение аргументов #

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

@app.task
def process_record(record_id: int) -> bool:
    process(record_id)
    return True

Создание чанков по 100 записей:

arguments = (
    (record_id,)
    for record_id in record_ids
)

workflow = process_record.chunks(
    arguments,
    100,
)

workflow.apply_async()

Если имеется 10 000 элементов и размер чанка равен 100, Celery создаст 100 задач. Внутри каждой задачи 100 вызовов будут выполнены последовательно. chunks уменьшает количество сообщений брокера и накладные расходы по сравнению с отдельным сообщением на каждый элемент.

Важно понимать отличие:

group:
каждая signature становится отдельной задачей.

chunks:
одна задача последовательно обрабатывает
несколько наборов аргументов.

group или chunks #

Для 100 000 записей:

# 100 000 сообщений
group(
    process_record.s(record_id)
    for record_id in record_ids
)

# 100 сообщений при chunk_size=1000
process_record.chunks(
    ((record_id,) for record_id in record_ids),
    1_000,
)

chunks подходит для простой однородной обработки. Собственный process_batch удобнее, когда внутри батча нужны:

• одна транзакция;

• один SQL-запрос на группу ID;

• bulk_create или bulk_update;

• собственная статистика;

• особая обработка ошибок;

• контроль состояния батча.

Комбинирование chain, group и chord #

Распространённая схема: каждый батч проходит несколько последовательных этапов, а батчи выполняются параллельно.

batch 1: validate → process → persist ─┐
batch 2: validate → process → persist ─┼──> finalize
batch 3: validate → process → persist ─┘

from celery import chain, chord

batch_workflows = [
    chain(
        validate_batch.s(batch_id),
        process_batch.s(),
        persist_batch.s(),
    )
    for batch_id in batch_ids
]

workflow = chord(batch_workflows)(
    finalize_job.s(job_id)
)

workflow.apply_async()

Задачи:

@app.task
def validate_batch(batch_id: int) -> int:
    validate_batch_data(batch_id)
    return batch_id


@app.task
def process_batch(batch_id: int) -> int:
    process_batch_data(batch_id)
    return batch_id


@app.task
def persist_batch(batch_id: int) -> dict:
    result = persist_batch_result(batch_id)

    return {
        "batch_id": batch_id,
        "processed": result.processed,
        "failed": result.failed,
    }


@app.task
def finalize_job(
    batch_results: list[dict],
    job_id: int,
) -> None:
    ...

Canvas-примитивы являются signatures и могут вкладываться и объединяться в составные workflow.

Практическая архитектура большой обработки #

Для операции с миллионом записей лучше не передавать миллион ID непосредственно в Celery Canvas.

1. API создаёт Job
2. Приложение записывает Batch-строки в БД
3. В Celery передаются только batch_id
4. Worker загружает данные батча
5. Worker сохраняет результат батча
6. Финальная задача обновляет Job

Структура данных:

processing_jobs
├── id
├── status
├── total_batches
├── completed_batches
├── failed_batches
└── created_at

processing_batches
├── id
├── job_id
├── status
├── records_from
├── records_to
├── processed_count
└── failed_count

Запуск:

@app.task
def start_job(job_id: int) -> None:
    batch_ids = batch_repository.list_ids(job_id)

    chord(
        process_batch.s(batch_id)
        for batch_id in batch_ids
    )(
        finalize_job.s(job_id)
    )

Обработка:

@app.task(
    bind=True,
    autoretry_for=(TemporaryProcessingError,),
    retry_backoff=True,
    retry_jitter=True,
    max_retries=5,
    acks_late=True,
)
def process_batch(
    self,
    batch_id: int,
) -> dict:
    batch = batch_repository.lock_for_processing(batch_id)

    if batch.status == "completed":
        return {
            "batch_id": batch.id,
            "processed": batch.processed_count,
            "failed": batch.failed_count,
        }

    result = batch_processor.process(batch)

    batch_repository.mark_completed(
        batch_id=batch.id,
        processed=result.processed,
        failed=result.failed,
    )

    return {
        "batch_id": batch.id,
        "processed": result.processed,
        "failed": result.failed,
    }

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

Как обрабатывать ошибки #

Ошибка одной задачи в chain #

Последующие задачи цепочки не выполняются:

validate → process FAILED → persist не запускается

Можно назначить errback:

workflow = chain(
    validate_batch.s(batch_id),
    process_batch.s(),
    persist_batch.s(),
)

workflow.apply_async(
    link_error=mark_batch_failed.s(batch_id),
)

Ошибка задачи внутри group #

Остальные задачи группы продолжают выполняться независимо.

batch 1 SUCCESS
batch 2 FAILURE
batch 3 SUCCESS

Ошибка внутри chord #

Если одна из header-задач завершилась ошибкой, callback chord обычно не выполняется как успешная финализация. Ошибку необходимо учитывать через errback и состояние общей операции.

workflow = chord(header, finalize_job.s(job_id))

workflow.apply_async(
    link_error=mark_job_failed.s(job_id),
)

Для бизнес-процесса лучше хранить состояние Job в собственной БД, а не полагаться исключительно на Celery result backend.

Выбор размера батча #

Универсального размера нет.

Слишком маленький батч:

много сообщений;
нагрузка на broker;
нагрузка на result backend;
расходы на сериализацию;
большое количество task metadata.

Слишком большой батч:

долгое выполнение;
плохая балансировка Worker-ов;
дорогой повтор после ошибки;
долгое удержание транзакций;
блокировка очереди тяжёлыми задачами.

Celery рекомендует делить работу на небольшие задачи, но учитывать накладные расходы слишком мелкой гранулярности.

Практически размер подбирают по метрикам:

время обработки батча;
размер сообщения;
длина очереди;
CPU и память Worker-а;
нагрузка на БД;
стоимость повторного выполнения;
скорость Producer-а и Consumer-ов.

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

Основные ошибки #

# Плохо: одна task обрабатывает миллион записей
@app.task
def process_all(ids: list[int]) -> None:
    for record_id in ids:
        process(record_id)

Проблемы:

огромное сообщение;
один Worker занят надолго;
нет нормального параллелизма;
при ошибке повторяется вся обработка;
сложно отслеживать прогресс.

Также не следует:

передавать ORM-объекты в task;
возвращать огромные результаты;
вызывать result.get() внутри Worker-а;
создавать задачу на каждую микроскопическую операцию;
считать повторную доставку невозможной;
использовать chord без result backend;
запускать неидемпотентные задачи с acks_late.

Какой примитив выбирать #

Этапы зависят друг от друга
→ chain

Задачи независимы и выполняются параллельно
→ group

Параллельные задачи требуют общей финализации
→ chord

Одинаковая задача применяется к большому списку аргументов
→ chunks

Несколько последовательных этапов для каждого батча,
затем общая финализация
→ chord из chain

Итоговая схема #

workflow = chord(
    chain(
        validate_batch.s(batch_id),
        process_batch.s(),
        save_batch.s(),
    )
    for batch_id in batch_ids
)(
    finalize_job.s(job_id)
)

workflow.apply_async()

Это даёт:

параллельность между батчами;
последовательность этапов внутри батча;
независимые retry;
ограниченный объём повторной работы;
финализацию после завершения всех батчей;
возможность горизонтально масштабировать Worker-ы.

5. Преимущества очередей (Celery) #

Преимущества очередей задач в Celery #

Celery использует брокер сообщений — например, RabbitMQ или Redis — чтобы передавать задачи от приложения отдельным Worker-процессам.

FastAPI / Django
      │ task.delay(...)
      ▼
Broker
      │ очередь задач
      ▼
Celery Worker
      │
      ▼
Выполнение операции

1. Быстрый ответ клиенту #

Долгая операция не выполняется внутри HTTP-запроса:

@app.post("/reports", status_code=202)
async def create_report(report_id: int):
    generate_report.delay(report_id)

    return {
        "report_id": report_id,
        "status": "processing",
    }

API только ставит задачу в очередь, а Worker выполняет её отдельно. Это уменьшает время HTTP-ответа и не удерживает процесс веб-приложения занятой фоновой работой. В Celery клиент публикует сообщение в очередь, брокер передаёт его одному из Worker-ов.

2. Буферизация всплесков нагрузки #

Очередь отделяет скорость создания задач от скорости их выполнения:

API создаёт 10 000 задач
          ↓
      очередь
          ↓
Worker-ы обрабатывают их
с доступной скоростью

Без очереди резкий поток запросов пришлось бы немедленно обрабатывать приложению, БД или внешнему сервису. Очередь временно накапливает работу и позволяет Worker-ам забирать её постепенно.

Но очередь не устраняет перегрузку автоматически: необходимо контролировать её длину, скорость поступления задач и производительность Worker-ов.

3. Горизонтальное масштабирование #

Несколько Worker-ов могут обрабатывать одну очередь:

                 ┌── Worker 1
Queue ───────────┼── Worker 2
                 ├── Worker 3
                 └── Worker 4

Если задач становится больше, можно добавить Worker-процессы или отдельные серверы без масштабирования HTTP-приложения. Celery рассчитан на распределение работы между потоками или машинами и поддерживает несколько Worker-ов и брокеров.

4. Независимое масштабирование разных типов работы #

Задачи можно разделить по очередям:

emails       → Worker-ы для писем
reports      → Worker-ы для отчётов
images       → Worker-ы обработки изображений
integrations → Worker-ы внешних API

Например:

app.conf.task_routes = {
    "tasks.send_email": {
        "queue": "emails",
    },
    "tasks.generate_report": {
        "queue": "reports",
    },
}

Worker можно запустить только для выбранных очередей:

celery -A app worker -Q reports --loglevel=INFO

Celery поддерживает маршрутизацию задач по очередям и запуск Worker-а с ограниченным набором потребляемых очередей.

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

5. Изоляция веб-приложения от тяжёлой работы #

CPU-, I/O- и memory-intensive операции выполняются не в процессе FastAPI или Django:

Web application:
принимает запросы

Celery Worker:
генерирует PDF;
обрабатывает файл;
изменяет изображения;
вызывает внешние API.

Сбой конкретной фоновой задачи не обязан приводить к падению HTTP-запроса или всего веб-приложения.

При этом Celery не превращает CPU-bound задачу в быструю автоматически. Производительность зависит от количества Worker-процессов, оборудования и характера задачи.

6. Повторные попытки #

При временной ошибке задачу можно выполнить повторно:

@app.task(
    autoretry_for=(ConnectionError, TimeoutError),
    retry_backoff=True,
    retry_jitter=True,
    retry_kwargs={"max_retries": 5},
)
def send_to_external_api(operation_id: int) -> None:
    external_client.send(operation_id)

Это полезно при временной недоступности:

• внешнего API;

• SMTP-сервера;

• объектного хранилища;

• сетевого ресурса;

• другого микросервиса.

Celery поддерживает повтор задачи через retry() и автоматические retries.

Повторные попытки требуют идемпотентности:

@app.task
def process_payment(payment_id: str) -> None:
    if payment_repository.is_processed(payment_id):
        return

    payment_service.process(payment_id)

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

7. Восстановление после падения Worker-а #

Celery и брокер поддерживают acknowledgements — подтверждения получения или обработки сообщений.

При подходящей настройке задача подтверждается после выполнения:

@app.task(
    acks_late=True,
)
def process_batch(batch_id: int) -> None:
    ...

Если Worker аварийно завершится до подтверждения, сообщение может быть возвращено в очередь и обработано повторно. Именно поэтому задача с поздним подтверждением должна быть идемпотентной. Официальная документация отдельно связывает acks_late и возможность повторного выполнения задачи после сбоя Worker-а.

Это не гарантия «ровно один раз». В практической системе нужно исходить из модели:

Задача может быть доставлена
и выполнена более одного раза.

8. Управление справедливым распределением нагрузки #

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

Например:

app.conf.worker_prefetch_multiplier = 1

Это особенно полезно для долгих задач: один Worker не должен заранее забрать слишком много работы, пока другие Worker-ы простаивают. Документация Celery описывает связь prefetch multiplier с количеством зарезервированных задач и рекомендует разделять долгие и короткие задачи по Worker-ам и очередям.

9. Параллельная обработка больших объёмов #

Большую операцию можно разбить на батчи:

1 000 000 записей
        ↓
1 000 батчей
        ↓
очередь Celery
        ↓
несколько Worker-ов

from celery import group

workflow = group(
    process_batch.s(batch_id)
    for batch_id in batch_ids
)

workflow.apply_async()

Преимущества:

части обрабатываются параллельно;
ошибка одного батча не требует повторять все данные;
можно отслеживать прогресс;
нагрузка распределяется между Worker-ами;
можно добавлять вычислительные ресурсы.

10. Ослабление временной связанности #

Producer и Worker не обязаны одновременно выполнять одну операцию:

API опубликовал задачу
        ↓
Worker может обработать её позднее

Веб-приложению не нужно знать:

• какой конкретно Worker выполнит задачу;

• на каком сервере он находится;

• сколько экземпляров Worker-а запущено;

• когда именно задача начнёт выполняться.

Это упрощает независимое масштабирование и развёртывание компонентов.

11. Управление приоритетами и классами задач #

Разные очереди позволяют отделить важные операции от менее срочных:

critical   → платёжные статусы
default    → обычная обработка
low        → отчёты и архивирование

Даже без брокерных приоритетов раздельные очереди позволяют выделить отдельное количество Worker-ов каждому классу задач.

Например:

celery -A app worker -Q critical --concurrency=4
celery -A app worker -Q reports --concurrency=2

12. Построение сложных workflow #

Celery Canvas позволяет объединять задачи:

chain — последовательность;
group — параллельная группа;
chord — группа с общей финализацией;
chunks — пакетная обработка аргументов.

Например:

обработать батчи параллельно
          ↓
дождаться всех
          ↓
сформировать итоговый отчёт

from celery import chord

chord(
    process_batch.s(batch_id)
    for batch_id in batch_ids
)(
    finalize_report.s(job_id)
)

13. Выполнение задач по расписанию #

С помощью Celery Beat задачи можно публиковать периодически:

app.conf.beat_schedule = {
    "remove-expired-files": {
        "task": "tasks.remove_expired_files",
        "schedule": 3600.0,
    },
}

Celery Beat
     ↓
Broker
     ↓
Celery Worker

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

Ограничения #

Очереди Celery добавляют собственную сложность:

нужен Redis или RabbitMQ;
нужно обслуживать Worker-ы;
нужно следить за длиной очередей;
возможны повторные выполнения;
нужна идемпотентность;
нужны мониторинг и алерты;
нужно проектировать тайм-ауты и retries;
нужно контролировать размер сообщений.

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

• настроек брокера;

• persistence сообщений;

• acknowledgements;

• обработки повторов;

• идемпотентности;

• мониторинга;

• способа публикации задачи.

Например, операция:

save_order()
send_notification.delay(order_id)

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

Итог #

Главные преимущества очередей Celery:

1. Быстрый ответ HTTP API
2. Фоновое выполнение длительных операций
3. Буферизация всплесков нагрузки
4. Горизонтальное масштабирование Worker-ов
5. Разделение задач по очередям
6. Повторные попытки после временных ошибок
7. Возможность повторной доставки после сбоя
8. Параллельная обработка данных
9. Независимое масштабирование разных типов задач
10. Построение последовательных и параллельных workflow
11. Запуск задач по расписанию
12. Снижение прямой связанности компонентов

Очередь особенно полезна, когда работу необязательно завершать внутри текущего HTTP-запроса и её можно выполнить отдельно, повторить или распределить между несколькими Worker-ами.

6. Гарантии доставки сообщений в брокерах #

Что означают гарантии доставки #

Гарантия доставки описывает, сколько раз сообщение может быть передано и обработано при сбоях:

Producer → Broker → Consumer → БД / внешний сервис

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

Важно различать:

доставка сообщения
≠
выполнение бизнес-операции

Брокер может гарантировать передачу сообщения Consumer-у, но не способен автоматически гарантировать, что запись в PostgreSQL, платёж или HTTP-запрос во внешнюю систему произошли ровно один раз.

At-most-once — не более одного раза #

Сообщение обрабатывается максимум один раз:

сообщение потеряно
или
обработано один раз

Повторная доставка не выполняется.

Пример:

Broker передал сообщение
        ↓
сразу считает его обработанным
        ↓
Consumer упал до выполнения операции
        ↓
сообщение потеряно

В RabbitMQ это близко к автоматическому подтверждению:

auto_ack=True

При automatic acknowledgement RabbitMQ считает сообщение успешно доставленным после отправки его Consumer-у, не ожидая подтверждения успешной обработки. Это обеспечивает высокую скорость, но при обрыве соединения или падении Consumer-а сообщение может быть потеряно.

Такая модель подходит для некритичных данных:

метрики;
часть логов;
временные обновления интерфейса;
телеметрия, потеря которой допустима.

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

At-least-once — хотя бы один раз #

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

сообщение обработано один раз
или
сообщение обработано несколько раз

Типичный сценарий:

1. Consumer получил сообщение
2. Consumer изменил данные в БД
3. Consumer упал до отправки ACK
4. Broker не получил подтверждение
5. Broker доставил сообщение повторно
6. Операция выполняется второй раз

RabbitMQ автоматически возвращает неподтверждённые сообщения в очередь при закрытии канала или соединения. Поэтому ручные acknowledgements обеспечивают повторную доставку, но создают возможность дубликатов.

Это наиболее распространённая модель для очередей задач:

лучше получить дубликат,
чем полностью потерять задачу

Amazon SQS Standard также официально предоставляет at-least-once: копия сообщения иногда может быть получена повторно, поэтому AWS требует проектировать Consumer-ы идемпотентными.

Exactly-once — ровно один раз #

На практике термин exactly-once часто понимают неправильно.

Недостаточно сделать так, чтобы Broker отправил сообщение Consumer-у один раз. Нужно гарантировать, что весь наблюдаемый бизнес-эффект произошёл ровно один раз:

получить сообщение
        +
изменить БД
        +
отправить следующее событие
        +
подтвердить сообщение

Все эти действия должны быть согласованы.

Проблема возникает в критическом промежутке:

Consumer изменил PostgreSQL
        ↓
Consumer должен отправить ACK

Если Consumer упал между этими действиями, Broker не знает, завершилась ли бизнес-операция:

БД обновлена
ACK не получен
        ↓
повторная доставка

Если изменить порядок:

сначала ACK
потом обновить БД

то возникает противоположная проблема:

ACK отправлен
Consumer упал
БД не обновлена
        ↓
сообщение потеряно

Поэтому обычная очередь плюс обычная БД не дают настоящей сквозной exactly-once семантики автоматически.

Apache Kafka поддерживает транзакционную exactly-once обработку для сценария, где приложение читает из Kafka, обрабатывает данные и записывает результат обратно в Kafka. Но сама документация Kafka подчёркивает, что подобные гарантии имеют конкретные границы и не автоматически распространяются на произвольные внешние базы и системы.

На уровне бизнес-приложения обычно реализуют не физическое «доставить один раз», а:

доставлять at-least-once
+
обрабатывать идемпотентно
=
эффект exactly-once

Гарантия состоит из нескольких участков #

Нельзя настроить только одну опцию Broker-а и получить надёжную доставку.

Полный путь:

1. Producer создал сообщение
2. Сообщение дошло до Broker-а
3. Broker сохранил сообщение
4. Broker передал его Consumer-у
5. Consumer выполнил бизнес-операцию
6. Consumer подтвердил обработку

Для каждого участка нужны отдельные механизмы.

Producer → Broker: publisher confirms #

Обычный вызов публикации ещё не означает, что Broker принял сообщение.

Producer отправил сообщение
        ↓
соединение оборвалось
        ↓
неизвестно:
Broker получил сообщение или нет

В RabbitMQ для этого используются publisher confirms:

Producer → RabbitMQ: сообщение
RabbitMQ → Producer: confirm

Confirm означает, что RabbitMQ принял ответственность за сообщение. Для persistent-сообщения в durable-очереди подтверждение отправляется после сохранения сообщения; для quorum queue — после принятия сообщения кворумом реплик.

Однако потеря самого подтверждения всё равно возможна:

RabbitMQ сохранил сообщение
RabbitMQ отправил confirm
соединение разорвалось
Producer confirm не получил
        ↓
Producer публикует сообщение повторно

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

Поэтому Producer тоже должен использовать:

message_id;
event_id;
idempotency key;
таблицу outbox.

Сохранность внутри Broker-а #

Чтобы сообщение переживало перезапуск RabbitMQ, обычно нужны одновременно:

durable queue;
persistent message;
publisher confirms.

Одна durable-очередь недостаточна: она сохраняет определение очереди, но сообщения также должны публиковаться как persistent. Publisher confirms позволяют Producer-у узнать, когда RabbitMQ принял ответственность за сообщение.

Для повышенной отказоустойчивости RabbitMQ предоставляет quorum queues. Confirm для такой очереди отправляется после репликации сообщения на кворум узлов.

Схема:

Producer
   │ persistent message
   ▼
Quorum Queue
├── replica 1
├── replica 2
└── replica 3
   │
   ▼
publisher confirm

Это защищает от некоторых отказов Broker-а, но не устраняет дубликаты на уровне Consumer-а.

Broker → Consumer: acknowledgements #

Consumer acknowledgements сообщают RabbitMQ, когда сообщение можно удалить.

ACK после успешной операции #

def handle(message):
    process(message)
    channel.basic_ack(message.delivery_tag)

Это приближает систему к at-least-once:

успех → ACK → сообщение удаляется
сбой до ACK → сообщение доставляется повторно

RabbitMQ рекомендует manual acknowledgements, когда необработанные сообщения должны возвращаться в очередь и повторно обрабатываться другим Consumer-ом.

NACK или reject #

При ошибке Consumer может:

ACK                  → удалить сообщение;
NACK + requeue       → вернуть в очередь;
NACK без requeue     → удалить или направить в DLQ.

Бесконечно возвращать неисправное сообщение в ту же очередь опасно:

получение → ошибка → requeue → получение → ошибка → ...

Поэтому обычно применяют:

retry queue;
ограниченное число повторов;
exponential backoff;
Dead Letter Queue.

Идемпотентный Consumer #

Для at-least-once Consumer должен выдерживать повторную доставку.

Пример таблицы дедупликации:

CREATE TABLE processed_messages (
    consumer_name TEXT NOT NULL,
    message_id UUID NOT NULL,
    processed_at TIMESTAMPTZ NOT NULL DEFAULT now(),

    PRIMARY KEY (consumer_name, message_id)
);

Обработка выполняется в одной локальной транзакции:

def handle_message(message: Message) -> None:
    with database.transaction():
        inserted = processed_messages.try_insert(
            consumer_name="payment_consumer",
            message_id=message.id,
        )

        if not inserted:
            return

        payment_service.apply_result(
            payment_id=message.payment_id,
            status=message.status,
        )

первая доставка:
message_id отсутствует → операция выполняется

повторная доставка:
message_id уже существует → операция пропускается

Ключевой момент: запись о дедупликации и бизнес-изменение должны находиться в одной транзакции.

Плохо:

mark_as_processed(message.id)
update_payment(message)

Если процесс упадёт между вызовами, сообщение будет считаться обработанным, хотя бизнес-операция не завершилась.

Transactional Outbox на стороне Producer-а #

Проблема Producer-а:

save_order(order)
publish_event("OrderCreated", order.id)

Возможен сбой:

заказ сохранён
событие не опубликовано

Либо обратный порядок:

событие опубликовано
заказ не сохранён

Transactional Outbox решает это через одну транзакцию БД:

with database.transaction():
    save_order(order)

    outbox_repository.add(
        event_id=uuid4(),
        event_type="OrderCreated",
        payload={"order_id": str(order.id)},
    )

Отдельный Publisher читает outbox и отправляет события в Broker:

Бизнес-транзакция
├── orders
└── outbox

Outbox Worker
      ↓
RabbitMQ

Publisher может отправить одно событие повторно, например если RabbitMQ принял сообщение, но confirm потерялся. Поэтому event_id остаётся необходимым для дедупликации Consumer-а.

Celery и гарантии доставки #

По умолчанию Celery обычно подтверждает задачу до её фактического выполнения. Это снижает риск повторного выполнения, но задача может быть потеряна, если Worker упадёт после ACK и до завершения работы.

Для идемпотентных задач можно включить позднее подтверждение:

@app.task(
    acks_late=True,
)
def process_order(order_id: int) -> None:
    ...

Тогда задача подтверждается после возврата из task, что приближает обработку к at-least-once. Документация Celery прямо требует идемпотентности для задач с acks_late.

Но есть важная особенность: даже при acks_late=True Celery по умолчанию может подтвердить задачу, если дочерний процесс Worker-а аварийно завершился или был принудительно остановлен. Для возврата такой задачи в очередь существует:

task_reject_on_worker_lost = True

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

Практическая конфигурация для критичных идемпотентных задач:

app.conf.update(
    task_acks_late=True,
    task_reject_on_worker_lost=True,
)

Но конфигурация сама по себе не заменяет:

идемпотентность;
ограничение retries;
тайм-ауты;
DLQ;
publisher confirms;
надёжный Broker;
Transactional Outbox.

Практическое сравнение #

At-most-once #

ACK до обработки

Плюсы:

нет повторного выполнения;
простая обработка;
меньше накладных расходов.

Минус:

при сбое возможна потеря сообщения.

At-least-once #

обработка → commit → ACK

Плюсы:

сообщение не теряется при обычном падении Consumer-а;
можно повторить временно неуспешную операцию.

Минусы:

возможны дубликаты;
нужна идемпотентность;
нужна дедупликация.

Exactly-once effect #

at-least-once delivery
+
transactional inbox / idempotency
+
transactional outbox

Плюс:

повторная доставка не создаёт повторного бизнес-эффекта.

Минус:

существенно сложнее;
требует хранения идентификаторов;
требует транзакционной дисциплины;
не всегда возможно для внешних систем.

Что выбирать #

Для большинства бизнес-систем разумная цель:

at-least-once доставка
+
идемпотентные Consumer-ы
+
Outbox у Producer-а
+
Inbox/deduplication у Consumer-а

Итоговая схема:

Producer DB transaction
├── business data
└── outbox event
        ↓
Outbox Publisher
        ↓ publisher confirm
RabbitMQ durable/quorum queue
        ↓ manual delivery
Consumer transaction
├── processed_messages
└── business changes
        ↓
ACK

Это не означает, что пакет физически никогда не будет отправлен повторно. Это означает, что повторная доставка не изменит бизнес-результат второй раз.

Главное правило #

At-most-once:
можно потерять, нельзя повторить.

At-least-once:
нельзя терять, можно повторить.

Exactly-once:
обычно достигается не брокером,
а идемпотентной и транзакционной обработкой
поверх at-least-once доставки.

7. Гарантии доставки в RabbitMQ #

Гарантии доставки в RabbitMQ #

RabbitMQ не предоставляет одну глобальную настройку вида «гарантировать доставку». Надёжность складывается из трёх отдельных участков:

Producer
   │ 1. публикация
   ▼
RabbitMQ
   │ 2. хранение и маршрутизация
   ▼
Consumer
   │ 3. обработка и подтверждение
   ▼
БД / внешний сервис

Для каждого участка применяются собственные механизмы:

Producer → RabbitMQ:
publisher confirms

Хранение в RabbitMQ:
durable/quorum queue + persistence

RabbitMQ → Consumer:
manual acknowledgements

Бизнес-обработка:
идемпотентность и дедупликация

At-most-once #

At-most-once означает:

Сообщение будет обработано:
0 или 1 раз

Потеря возможна, повторная доставка не предполагается.

Например, Consumer использует автоматическое подтверждение:

RabbitMQ отправил сообщение
        ↓
считает его подтверждённым
        ↓
Consumer упал до обработки
        ↓
сообщение потеряно

В RabbitMQ отсутствие ручных подтверждений приближает доставку к at-most-once: после передачи сообщения Consumer-у Broker не знает, действительно ли бизнес-операция завершилась. Официальное руководство RabbitMQ отмечает, что без acknowledgements возможна потеря сообщений.

Такой режим подходит только там, где потеря допустима:

некритичная телеметрия;
часть метрик;
временные события интерфейса;
обновления, которые скоро будут заменены новыми.

At-least-once #

At-least-once означает:

Сообщение будет обработано:
1 или несколько раз

RabbitMQ стремится не потерять сообщение, но повторная доставка возможна.

Типичный случай:

1. Consumer получил сообщение
2. Изменил данные в PostgreSQL
3. Упал до отправки ACK
4. RabbitMQ не получил подтверждение
5. Сообщение вернулось в очередь
6. Другой Consumer получил его повторно

При закрытии канала или соединения неподтверждённые сообщения автоматически возвращаются в очередь. Поэтому ручные acknowledgements обеспечивают at-least-once, но требуют готовности к повторной обработке.

Схема Consumer-а:

def handle(message) -> None:
    try:
        process_message(message)
    except TemporaryError:
        channel.basic_nack(
            delivery_tag=message.delivery_tag,
            requeue=True,
        )
    else:
        channel.basic_ack(
            delivery_tag=message.delivery_tag,
        )

Подтверждать сообщение следует только после завершения необходимой работы:

получить сообщение
        ↓
изменить БД
        ↓
зафиксировать транзакцию
        ↓
отправить ACK

После ACK RabbitMQ получает право удалить сообщение из очереди.

Exactly-once RabbitMQ не гарантирует #

RabbitMQ не обеспечивает сквозное выполнение бизнес-операции строго один раз.

Рассмотрим Consumer:

Consumer обновил БД
        ↓
Consumer должен отправить ACK

Если Consumer упадёт между этими действиями:

БД уже изменена
ACK не дошёл
RabbitMQ повторно доставляет сообщение

Если отправлять ACK до транзакции:

ACK отправлен
Consumer упал
БД не обновлена
сообщение уже удалено

RabbitMQ не может атомарно объединить:

COMMIT в PostgreSQL
+
ACK сообщения

Поэтому обычно строят:

at-least-once delivery
+
идемпотентный Consumer
=
бизнес-эффект, близкий к exactly-once

Publisher confirms #

Publisher должен убедиться, что RabbitMQ действительно принял сообщение.

Без подтверждений:

Producer отправил сообщение
        ↓
соединение оборвалось
        ↓
неизвестно, дошло ли сообщение

Для этого RabbitMQ предоставляет publisher confirms:

Producer ──publish──> RabbitMQ
Producer <──confirm── RabbitMQ

После включения confirm mode каждое опубликованное сообщение будет либо подтверждено, либо отклонено Broker-ом. Для persistent-сообщений, направленных в durable-очереди, confirm отправляется после сохранения; для quorum queue — после принятия сообщения кворумом реплик.

Условный пример с pika:

import pika

connection = pika.BlockingConnection(
    pika.ConnectionParameters("localhost")
)
channel = connection.channel()

channel.confirm_delivery()

channel.queue_declare(
    queue="orders",
    durable=True,
)

confirmed = channel.basic_publish(
    exchange="",
    routing_key="orders",
    body=b'{"order_id": 42}',
    properties=pika.BasicProperties(
        delivery_mode=pika.DeliveryMode.Persistent,
        message_id="event-123",
    ),
    mandatory=True,
)

if not confirmed:
    raise RuntimeError("RabbitMQ did not confirm the message")

Publisher confirm означает, что RabbitMQ принял ответственность за сообщение. Он не означает, что Consumer уже обработал его.

publisher confirm:
RabbitMQ принял сообщение

consumer ACK:
Consumer завершил обработку

Это независимые механизмы.

Потерянный confirm и дубликаты #

Возможна ситуация:

1. RabbitMQ принял сообщение
2. Отправил confirm
3. Соединение оборвалось
4. Producer confirm не получил
5. Producer публикует сообщение повторно

В RabbitMQ уже может находиться первая копия, поэтому повторная публикация создаст дубликат. Официальное руководство RabbitMQ прямо предупреждает, что повторная отправка неподтверждённых публикаций может приводить к дубликатам.

Следовательно, publisher confirms не отменяют необходимость использовать:

message_id;
event_id;
idempotency_key;
дедупликацию на стороне Consumer-а.

Durable queue и persistent message #

Для переживания перезапуска Broker-а нужно различать очередь и сообщение.

Durable queue #

channel.queue_declare(
    queue="orders",
    durable=True,
)

durable=True означает, что определение очереди восстанавливается после перезапуска RabbitMQ.

Persistent message #

properties=pika.BasicProperties(
    delivery_mode=pika.DeliveryMode.Persistent,
)

Persistent-сообщение предназначено для сохранения при перезапуске Broker-а.

Для надёжного варианта нужны вместе:

durable queue
+
persistent message
+
publisher confirms

Одного durable=True недостаточно: transient-сообщение не становится persistent только потому, что попало в durable-очередь. RabbitMQ рассматривает долговечность очереди и режим сохранения сообщения как отдельные свойства.

Quorum queues #

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

Quorum queue:

          ┌── replica 1
Producer ─┼── replica 2
          └── replica 3

Она использует Raft и хранит реплицированное состояние очереди. RabbitMQ рекомендует quorum queues для долгоживущих критичных очередей, где важны сохранность данных и высокая доступность.

Публикация считается подтверждённой, когда сообщение принято кворумом реплик:

3 реплики
   ↓
минимум 2 приняли сообщение
   ↓
Producer получает confirm

Успешно подтверждённое сообщение не должно быть потеряно, пока большинство узлов quorum queue не стало постоянно недоступно. При этом для неподтверждённых публикаций гарантий сохранности нет.

Практическая комбинация для критичной очереди:

quorum queue
+
publisher confirms
+
manual consumer acknowledgements
+
идемпотентный Consumer

Unroutable messages #

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

Например:

Producer → Exchange
              │
              └── нет подходящего binding

RabbitMQ может подтвердить сообщение после того, как Exchange определил, что оно не маршрутизируется ни в одну очередь. Чтобы Producer узнал об этом, при публикации используется флаг mandatory; Broker вернёт сообщение через basic.return.

channel.basic_publish(
    exchange="orders",
    routing_key="unknown.key",
    body=payload,
    mandatory=True,
)

Поэтому надёжный Producer должен обрабатывать:

publisher confirm;
publisher nack;
basic.return для unroutable message;
обрыв соединения;
тайм-аут ожидания confirm.

ACK, NACK и reject #

После получения сообщения Consumer может выбрать одно из действий.

ACK #

Операция выполнена успешно
→ удалить сообщение

channel.basic_ack(delivery_tag)

NACK с requeue #

Ошибка временная
→ вернуть сообщение в очередь

channel.basic_nack(
    delivery_tag,
    requeue=True,
)

NACK без requeue #

Сообщение невозможно обработать
→ удалить или направить в DLX

channel.basic_nack(
    delivery_tag,
    requeue=False,
)

RabbitMQ поддерживает basic.reject и basic.nack; последний также позволяет отрицательно подтвердить несколько сообщений. Сообщение с requeue=false может быть направлено в Dead Letter Exchange, если он настроен.

Почему нельзя бесконечно делать requeue #

Опасная схема:

Consumer получил сообщение
        ↓
ошибка
        ↓
NACK requeue=True
        ↓
Consumer снова получил сообщение
        ↓
ошибка
        ↓
...

Это создаёт бесконечный цикл, загружая Broker и Consumer-ы.

Обычно используют:

основная очередь
      ↓ временная ошибка
retry queue
      ↓ задержка
основная очередь
      ↓ превышен лимит
DLQ

Сообщение может быть dead-lettered после reject/nack с requeue=false, истечения TTL, превышения длины очереди либо превышения delivery limit для quorum queue.

DLX тоже имеет собственные гарантии #

Обычная dead-lettering-передача не обязательно является надёжной at-least-once операцией.

Для quorum queue RabbitMQ поддерживает специальный режим at-least-once dead-lettering:

Source quorum queue
        ↓ internal consumer
        ↓ publish + publisher confirms
Target queue
        ↓
ACK исходному сообщению

При этом исходное сообщение удаляется только после подтверждения публикации в целевую очередь. Однако повторные попытки всё равно могут создать дубликаты в целевой очереди. По умолчанию стратегия dead-lettering для quorum queues остаётся at-most-once.

Идемпотентный Consumer #

Поскольку at-least-once допускает дубликаты, Consumer должен уметь распознавать уже обработанные сообщения.

Например:

CREATE TABLE processed_messages (
    consumer_name TEXT NOT NULL,
    message_id UUID NOT NULL,
    processed_at TIMESTAMPTZ NOT NULL DEFAULT now(),

    PRIMARY KEY (consumer_name, message_id)
);

Обработка в одной транзакции:

def handle_message(message: Message) -> None:
    with database.transaction():
        inserted = processed_messages.try_insert(
            consumer_name="order_consumer",
            message_id=message.id,
        )

        if not inserted:
            return

        order_service.apply_event(message)

Важно, чтобы запись о дедупликации и бизнес-операция фиксировались атомарно:

одна транзакция
├── INSERT processed_messages
└── UPDATE business_data

После успешного COMMIT Consumer отправляет ACK.

COMMIT
  ↓
ACK

При падении до ACK сообщение придёт повторно, но уникальный message_id не позволит повторить бизнес-эффект.

Transactional Outbox для Producer-а #

Publisher confirms защищают участок:

приложение → RabbitMQ

Но не решают проблему:

save_order(order)
publish_event(order)

Если приложение упадёт после сохранения заказа, но до публикации события:

Order сохранён
OrderCreated не отправлен

Обычно применяют Transactional Outbox:

Одна транзакция PostgreSQL
├── orders
└── outbox_events

Отдельный Publisher читает outbox и отправляет события в RabbitMQ с publisher confirms.

Business transaction
        ↓
Outbox table
        ↓
Outbox Publisher
        ↓
RabbitMQ

Outbox Publisher может отправить событие повторно при потере confirm, поэтому Consumer всё равно должен выполнять дедупликацию.

Практические уровни RabbitMQ #

Минимальная надёжность #

auto acknowledgement
transient queue/message
без publisher confirms

Семантика близка к:

at-most-once

Потеря сообщений допустима.

Обычная надёжная обработка #

durable queue
persistent messages
publisher confirms
manual acknowledgements
идемпотентный Consumer

Семантика:

at-least-once

Сообщение может прийти повторно.

Критичная кластерная очередь #

quorum queue
publisher confirms
manual acknowledgements
идемпотентный Consumer
Transactional Outbox
Inbox / таблица дедупликации
DLQ и ограниченные retries

Результат:

at-least-once доставка
+
защита от повторного бизнес-эффекта

Итог #

RabbitMQ практически поддерживает две основные модели:

Без acknowledgements:
at-most-once
сообщение может потеряться

С acknowledgements и повторной доставкой:
at-least-once
сообщение может быть обработано повторно

Exactly-once RabbitMQ сам по себе не предоставляет. Для критичных операций обычно применяют:

Publisher:
Transactional Outbox
+ publisher confirms

RabbitMQ:
quorum queue
+ репликация

Consumer:
manual ACK
+ идемпотентность
+ Inbox/deduplication

Главная схема:

Producer DB transaction
├── бизнес-данные
└── outbox
        ↓
publisher confirms
        ↓
RabbitMQ quorum queue
        ↓
manual delivery
        ↓
Consumer DB transaction
├── processed_messages
└── бизнес-изменение
        ↓
ACK

Такое построение не исключает физические дубликаты сообщений, но предотвращает повторный бизнес-эффект.

8. Из каких основных компонентов состоит брокер сообщений? #

Основные компоненты брокера сообщений #

В типовой системе обмена сообщениями участвуют:

Producer
   │ публикует Message
   ▼
Broker
   │ маршрутизирует и хранит
   ▼
Queue / Topic
   │
   ▼
Consumer

Сам брокер находится между отправителями и получателями: принимает сообщения от Producer-ов, определяет место назначения, сохраняет их и доставляет Consumer-ам. В RabbitMQ приложения обычно публикуют сообщения в exchange, после чего они маршрутизируются в очереди по правилам binding.

1. Producer #

Producer, или Publisher, — приложение, которое создаёт и отправляет сообщение.

Например:

Order Service
    │
    │ OrderCreated
    ▼
RabbitMQ

Producer обычно определяет:

• тело сообщения;

• тип события или команды;

• идентификатор сообщения;

• routing key;

• заголовки;

• параметры сохранения;

• срок жизни сообщения.

Пример сообщения:

{
  "message_id": "evt-125",
  "event_type": "OrderCreated",
  "order_id": 42,
  "created_at": "2026-07-27T10:00:00Z"
}

2. Message #

Message — единица передаваемых данных.

Обычно оно состоит из двух частей:

Message
├── Headers / Properties
└── Payload

Пример:

Headers:
message_id = evt-125
content_type = application/json
routing_key = order.created

Payload:
{
  "order_id": 42
}

Broker обычно не обязан понимать бизнес-смысл содержимого сообщения. Для него важнее адрес, routing key, заголовки и параметры доставки.

3. Exchange или маршрутизатор #

В RabbitMQ Producer обычно отправляет сообщение в exchange, а не напрямую Consumer-у.

Producer
   │ routing_key=payment.succeeded
   ▼
Exchange
   ├──> accounting_queue
   ├──> notification_queue
   └──> audit_queue

Exchange принимает сообщение и определяет, в какие очереди его направить. Алгоритм маршрутизации зависит от типа exchange и настроенных bindings. RabbitMQ поддерживает основные типы direct, fanout, topic и headers.

Direct exchange #

Маршрутизация по точному совпадению ключа:

routing_key = payment.succeeded
        ↓
binding_key = payment.succeeded

Fanout exchange #

Отправляет копию сообщения во все связанные очереди:

                 ┌──> analytics
Event → fanout ──┼──> notifications
                 └──> audit

Topic exchange #

Маршрутизирует по шаблону:

События:
order.created
order.cancelled
payment.succeeded

Binding:
order.*

Headers exchange #

Использует значения заголовков вместо routing key.

Exchange — особенность модели AMQP/RabbitMQ. В других брокерах аналогичную роль могут выполнять topics, subscriptions или другие механизмы маршрутизации.

4. Binding #

Binding — правило связи между exchange и очередью.

Exchange ── binding ──> Queue

Например:

Exchange: domain_events
Queue: notification_queue
Binding key: order.*

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

Binding может содержать binding key, который служит фильтром для определённых типов exchange. У одного exchange может быть несколько bindings, а одна очередь может быть связана с несколькими exchange.

5. Queue #

Queue — хранилище сообщений, ожидающих обработки.

Queue
├── Message 1
├── Message 2
├── Message 3
└── Message 4

Очередь разделяет скорость Producer-а и Consumer-а:

Producer: создаёт 1000 сообщений/с
Consumer: обрабатывает 300 сообщений/с
                    ↓
         остаток временно хранится
              в очереди

В RabbitMQ очередь представляет собой упорядоченную коллекцию сообщений, но фактический порядок обработки может изменяться при нескольких конкурирующих Consumer-ах, повторной доставке и приоритетах.

Для очереди обычно настраивают:

• имя;

• долговечность;

• максимальную длину;

• TTL;

• тип очереди;

• Dead Letter Exchange;

• количество Consumer-ов;

• политику репликации.

6. Consumer #

Consumer — приложение, которое подписывается на очередь и обрабатывает сообщения.

Queue
   ├──> Consumer 1
   ├──> Consumer 2
   └──> Consumer 3

Например:

def handle_order_created(message: dict) -> None:
    send_confirmation_email(
        order_id=message["order_id"],
    )

Если несколько Consumer-ов читают одну очередь, они обычно конкурируют за сообщения:

Message 1 → Consumer 1
Message 2 → Consumer 2
Message 3 → Consumer 1

Одно сообщение рабочей очереди обычно передаётся одному Consumer-у, а не всем сразу. Для доставки одного события нескольким системам создают несколько очередей, связанных с одним exchange.

7. Acknowledgement #

Acknowledgement, или ACK, — подтверждение от Consumer-а, что сообщение успешно обработано.

Broker → Consumer: Message
Consumer → БД: COMMIT
Consumer → Broker: ACK

После ACK брокер может удалить сообщение из очереди.

При ошибке Consumer может отправить отрицательное подтверждение:

NACK + requeue=true
→ вернуть сообщение в очередь

NACK + requeue=false
→ удалить или направить в DLQ

RabbitMQ использует Consumer acknowledgements отдельно от Publisher confirms: первые относятся к обработке сообщения Consumer-ом, вторые — к принятию публикации Broker-ом.

8. Publisher confirm #

Publisher confirm — подтверждение Producer-у, что Broker принял публикацию.

Producer ── Message ──> Broker
Producer <── Confirm ── Broker

Это не означает, что Consumer уже выполнил задачу:

Publisher confirm:
сообщение принято RabbitMQ

Consumer ACK:
сообщение обработано Consumer-ом

Эти подтверждения защищают разные участки доставки.

9. Dead Letter Queue #

Dead Letter Queue, или DLQ, хранит сообщения, которые не удалось нормально обработать.

Main Queue
    │ ошибка после нескольких retries
    ▼
Dead Letter Exchange
    ▼
Dead Letter Queue

Туда могут попадать сообщения:

• отклонённые без requeue;

• с истёкшим TTL;

• превысившие лимит доставок;

• вытесненные из-за ограничения длины очереди.

DLQ позволяет не создавать бесконечный цикл обработки неисправного сообщения и сохраняет его для анализа или ручного повторного запуска.

10. Retry queue #

Retry queue используется для отложенной повторной обработки:

Main Queue
    │ временная ошибка
    ▼
Retry Queue
    │ TTL: 30 секунд
    ▼
Main Queue

Retry и DLQ не всегда являются отдельными встроенными сущностями брокера. Часто это обычные очереди, которым задают специальные TTL, bindings и dead-letter policies.

11. Connection #

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

Application ── TCP/TLS connection ──> RabbitMQ

Соединение обычно является долгоживущим. Создавать новое TCP-соединение для каждого сообщения слишком дорого. RabbitMQ поддерживает несколько протоколов, а AMQP-соединения обычно сохраняются и переиспользуются приложением.

12. Channel #

В AMQP 0-9-1 внутри одного TCP-соединения могут существовать несколько логических channels:

TCP Connection
├── Channel 1 → publishing
├── Channel 2 → consuming
└── Channel 3 → topology operations

Channel — облегчённая логическая связь, разделяющая одно физическое TCP-соединение. Через channels приложение объявляет exchange и очереди, публикует сообщения и запускает Consumer-ов.

13. Virtual host #

В RabbitMQ virtual host логически изолирует ресурсы:

RabbitMQ cluster
├── vhost: production
│   ├── exchanges
│   ├── queues
│   └── bindings
│
└── vhost: testing
    ├── exchanges
    ├── queues
    └── bindings

Внутри vhost находятся:

• exchanges;

• queues;

• bindings;

• permissions;

• policies.

Virtual hosts обеспечивают логическое разделение, но не являются физически независимыми RabbitMQ-кластерами. Права пользователей также назначаются в контексте конкретного vhost.

14. Управление доступом #

Брокер также включает инфраструктуру безопасности:

Authentication:
кто подключается

Authorization:
что этому клиенту разрешено

TLS:
защита сетевого трафика

Для Producer-а можно разрешить только публикацию:

write → domain_events

Для Consumer-а — только чтение конкретной очереди:

read → notification_queue

Это ограничивает последствия компрометации одного приложения.

Полная схема RabbitMQ #

Producer
   │
   │ Message + routing_key
   ▼
Connection / Channel
   │
   ▼
Exchange
   │
   │ Binding
   ▼
Queue
   │
   ├── retry / DLQ
   │
   ▼
Consumer
   │
   │ ACK / NACK
   ▼
Business operation

Обязательные и дополнительные компоненты #

Минимальная концептуальная схема любого брокера:

Producer
Message
Broker
Queue или Topic
Consumer

В RabbitMQ она расширяется:

Producer
   ↓
Exchange
   ↓
Binding
   ↓
Queue
   ↓
Consumer

Инфраструктурные компоненты:

Connections и channels
Acknowledgements и confirms
Retry и DLQ
Virtual hosts
Authentication и permissions
Persistence и replication
Monitoring

Итог #

Главные роли компонентов:

В упрощённом виде:

Producer создаёт сообщение.
Exchange решает, куда его направить.
Queue хранит сообщение.
Consumer обрабатывает его.
ACK подтверждает успешную обработку.

9. Механизм доставки сообщений в RabbitMQ #

Общая схема доставки #

В RabbitMQ сообщение проходит следующий путь:

Producer
   │ publish(message, exchange, routing_key)
   ▼
Exchange
   │ проверяет bindings
   ▼
Queue
   │ хранит сообщение
   ▼
Consumer
   │ обрабатывает
   ▼
ACK / NACK

Producer обычно публикует сообщение в exchange. Exchange маршрутизирует его в одну или несколько очередей, после чего RabbitMQ доставляет сообщение подписанному Consumer-у.

1. Подключение Producer-а #

Producer устанавливает TCP-соединение с RabbitMQ и открывает внутри него AMQP-channel:

Producer
   │
   └── TCP connection
          └── Channel

Через channel выполняются:

объявление exchange и очередей;
создание bindings;
публикация сообщений;
получение publisher confirms.

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

2. Публикация сообщения #

Producer указывает:

exchange;
routing_key;
payload;
properties;
headers.

Например:

channel.basic_publish(
    exchange="domain_events",
    routing_key="order.created",
    body=b'{"order_id": 42}',
    properties=properties,
)

Сообщение может содержать свойства:

message_id;
content_type;
delivery_mode;
timestamp;
type;
headers.

delivery_mode=2 обозначает persistent-сообщение, а delivery_mode=1 — transient.

3. Publisher confirm #

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

Producer ── message ──> RabbitMQ
Producer <── confirm ── RabbitMQ

После перевода channel в confirm mode RabbitMQ подтверждает каждую публикацию через basic.ack либо отклоняет её через basic.nack. Publisher confirms относятся только к участку Producer → RabbitMQ и не означают, что Consumer уже обработал сообщение.

Для persistent-сообщения, направленного в durable-очередь, подтверждение обычно отправляется после сохранения сообщения. Для quorum queue — после принятия сообщения кворумом реплик.

4. Маршрутизация через Exchange #

Exchange не хранит сообщения как обычная очередь. Он определяет, куда их направить.

Producer
   │ routing_key = order.created
   ▼
Exchange: domain_events
   ├──> notification_queue
   ├──> analytics_queue
   └──> audit_queue

Решение принимается на основе:

типа exchange;
routing key сообщения;
bindings;
binding keys;
заголовков.

Binding связывает exchange с очередью:

Exchange ── binding ──> Queue

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

Типы Exchange #

Direct #

Требует точного совпадения routing_key и binding_key:

routing_key: payment.succeeded
binding_key: payment.succeeded

Topic #

Поддерживает шаблоны:

order.created
order.cancelled
order.completed

Binding:

order.*

Fanout #

Отправляет сообщение во все связанные очереди, игнорируя routing key:

                  ┌──> queue_a
Producer → fanout ┼──> queue_b
                  └──> queue_c

Headers #

Маршрутизирует по заголовкам сообщения.

5. Что происходит с немаршрутизируемым сообщением #

Если ни один binding не подходит, сообщение не попадает ни в одну очередь.

Producer
   ↓
Exchange
   ↓
подходящих bindings нет
   ↓
сообщение не маршрутизировано

Publisher confirm при этом сам по себе не гарантирует попадание в очередь: RabbitMQ может подтвердить, что exchange обработал публикацию, даже если подходящая очередь не найдена.

Для обнаружения такого случая используется mandatory=True. Тогда RabbitMQ возвращает сообщение Producer-у через basic.return до отправки publisher confirm.

channel.basic_publish(
    exchange="domain_events",
    routing_key="unknown.event",
    body=payload,
    mandatory=True,
)

6. Попадание сообщения в очередь #

После маршрутизации сообщение помещается в подходящие очереди:

Queue
├── message 1
├── message 2
├── message 3
└── message 4

Очередь хранит сообщение, пока:

Consumer не подтвердит обработку;
не истечёт TTL;
сообщение не будет удалено политикой;
оно не попадёт в Dead Letter Exchange;
очередь не будет удалена.

Для переживания перезапуска RabbitMQ обычно сочетают:

durable queue;
persistent message;
publisher confirms.

Для реплицируемых критичных очередей могут использоваться quorum queues; RabbitMQ рекомендует сочетать их с publisher confirms и ручными Consumer acknowledgements.

7. Подписка Consumer-а #

Consumer регистрируется на определённой очереди:

Consumer ── basic.consume ──> Queue

После подписки RabbitMQ обычно push-ит сообщения Consumer-у асинхронно.

Queue ── basic.deliver ──> Consumer

AMQP 0-9-1 также поддерживает одиночное получение через basic.get, но обычная длительная обработка строится через Consumer subscription.

При доставке RabbitMQ передаёт:

payload;
properties;
routing key;
exchange;
delivery tag;
redelivered flag.

delivery_tag идентифицирует конкретную доставку внутри channel. Подтвердить сообщение необходимо на том же channel, на котором оно было получено.

8. Распределение между несколькими Consumer-ами #

Если одну очередь читают несколько Consumer-ов:

              ┌──> Consumer 1
Queue ────────┼──> Consumer 2
              └──> Consumer 3

одно сообщение передаётся одному подходящему Consumer-у. RabbitMQ распределяет сообщения между Consumer-ами с учётом их доступности и количества уже выданных, но ещё не подтверждённых сообщений.

message 1 → Consumer 1
message 2 → Consumer 2
message 3 → Consumer 3

При нескольких Consumer-ах, повторной доставке и разном времени обработки фактический порядок завершения сообщений уже не обязан соответствовать исходному порядку очереди.

9. Prefetch #

RabbitMQ может отправить Consumer-у несколько сообщений, не дожидаясь подтверждения каждого:

Consumer
├── message 1 — processing
├── message 2 — waiting
├── message 3 — waiting
└── message 4 — waiting

Количество неподтверждённых сообщений ограничивается через prefetch:

channel.basic_qos(prefetch_count=10)

При prefetch_count=10 RabbitMQ перестаёт отправлять новые сообщения этому Consumer-у после накопления десяти неподтверждённых доставок. Доставка продолжится после получения хотя бы одного ACK. Значение 0 означает отсутствие такого ограничения.

Prefetch нужен для:

защиты Consumer-а от перегрузки;
справедливого распределения задач;
ограничения потребления памяти;
управления количеством параллельной работы.

10. Обработка и ACK #

При ручных подтверждениях нормальная последовательность выглядит так:

RabbitMQ доставляет сообщение
        ↓
Consumer проверяет сообщение
        ↓
Consumer выполняет операцию
        ↓
COMMIT в БД
        ↓
Consumer отправляет ACK
        ↓
RabbitMQ удаляет сообщение

def handle(channel, method, properties, body):
    process_message(body)

    channel.basic_ack(
        delivery_tag=method.delivery_tag,
    )

RabbitMQ рекомендует отправлять ACK только после выполнения всех действий, за которые отвечает Consumer. После ACK брокер может удалить доставку.

11. Автоматическое подтверждение #

В автоматическом режиме RabbitMQ считает сообщение доставленным сразу после записи в сетевой сокет Consumer-а:

RabbitMQ отправил сообщение
        ↓
считает его обработанным
        ↓
Consumer упал
        ↓
сообщение может быть потеряно

Automatic acknowledgement даёт меньшие накладные расходы, но предоставляет наименьшие гарантии при сбоях. Официальная документация рекомендует в общем случае сначала рассматривать manual acknowledgement.

12. Ошибка обработки: NACK и Reject #

При ошибке Consumer может отрицательно подтвердить доставку.

Вернуть сообщение в очередь #

channel.basic_nack(
    delivery_tag=method.delivery_tag,
    requeue=True,
)

временная ошибка
      ↓
NACK + requeue
      ↓
сообщение снова доступно Consumer-ам

Не возвращать в очередь #

channel.basic_nack(
    delivery_tag=method.delivery_tag,
    requeue=False,
)

Тогда сообщение:

удаляется;
или направляется в Dead Letter Exchange,
если DLX настроен.

Бесконечный requeue=True опасен:

получение → ошибка → requeue
    ↑                     │
    └─────────────────────┘

Для временных ошибок обычно создают ограниченный retry-механизм, а окончательно неисправные сообщения отправляют в DLQ.

13. Падение Consumer-а #

При manual acknowledgement, если Consumer, channel или соединение закрывается до ACK, RabbitMQ автоматически возвращает неподтверждённое сообщение в очередь:

Consumer получил сообщение
        ↓
Consumer упал до ACK
        ↓
RabbitMQ выполняет requeue
        ↓
сообщение получает другой Consumer

При повторной доставке поле redelivered устанавливается в true. Consumer должен быть готов к дубликатам и проектироваться идемпотентно.

14. Почему возможны дубликаты #

Рассмотрим обработку:

1. Consumer обновил БД
2. Consumer должен отправить ACK
3. Consumer упал до ACK
4. RabbitMQ доставил сообщение повторно

RabbitMQ не знает, был ли выполнен COMMIT, поэтому дубликат является нормальной частью at-least-once доставки.

Consumer должен проверять message_id:

CREATE TABLE processed_messages (
    consumer_name TEXT NOT NULL,
    message_id UUID NOT NULL,
    processed_at TIMESTAMPTZ NOT NULL DEFAULT now(),

    PRIMARY KEY (consumer_name, message_id)
);

Первая доставка:
message_id отсутствует
→ выполнить операцию

Повторная доставка:
message_id уже существует
→ не повторять операцию

RabbitMQ прямо указывает, что acknowledgements обеспечивают модель at-least-once, а без них возможна потеря сообщений и семантика становится близкой к at-most-once.

Полный механизм доставки #

1. Producer открывает connection и channel

2. Producer публикует:
   exchange + routing_key + payload

3. Exchange проверяет bindings

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

5. Очередь сохраняет сообщение

6. RabbitMQ выбирает подходящего Consumer-а

7. RabbitMQ отправляет сообщение
   вместе с delivery_tag

8. Consumer выполняет бизнес-операцию

9. Consumer отправляет:
   ACK
   или NACK/reject

10. RabbitMQ:
    ACK            → удаляет сообщение
    NACK + requeue → возвращает в очередь
    NACK без requeue → удаляет или отправляет в DLQ

Два независимых подтверждения #

Важно не путать:

Publisher confirm
Producer <── RabbitMQ

и:

Consumer ACK
RabbitMQ <── Consumer

Publisher confirms и Consumer acknowledgements решают разные задачи и не связаны напрямую.

Итог #

Механизм RabbitMQ строится вокруг пяти операций:

Публикация
    ↓
Маршрутизация
    ↓
Хранение в очереди
    ↓
Доставка Consumer-у
    ↓
Подтверждение или повторная доставка

Для надёжной схемы обычно используют:

Producer:
publisher confirms
+ mandatory publishing

RabbitMQ:
durable или quorum queue
+ persistent messages

Consumer:
manual ACK
+ prefetch
+ ограниченные retries
+ DLQ
+ идемпотентность

Это даёт практическую семантику at-least-once: сообщение стараются не потерять, но оно может быть доставлено и обработано повторно.

10. Что делать, если очередь сообщений не гарантирует порядок? #

Сначала определить, какой порядок действительно нужен #

Глобальный порядок всех сообщений обычно не требуется. Чаще нужен порядок в рамках одной бизнес-сущности:

order_id=42:

OrderCreated
    ↓
OrderPaid
    ↓
OrderShipped

При этом события разных заказов могут обрабатываться параллельно:

order_id=42: Created → Paid → Shipped
order_id=91: Created → Cancelled

Это позволяет сохранить корректность без полной остановки параллелизма.

Почему порядок может нарушаться #

RabbitMQ хранит сообщения очереди упорядоченно, но наблюдаемый порядок обработки может измениться из-за:

• нескольких конкурирующих Consumer-ов;

• параллельной обработки внутри Consumer-а;

• разных скоростей выполнения;

• NACK и повторной постановки в очередь;

• падения Consumer-а и redelivery;

• нескольких каналов публикации;

• приоритетных очередей.

Начальные доставки одному Consumer-у выполняются в порядке постановки в очередь, но конкурентные Consumer-ы и повторная доставка меняют фактический порядок завершения обработки.

Queue:
A → B → C

Consumer 1 получает A — обрабатывает 10 секунд
Consumer 2 получает B — обрабатывает 1 секунду
Consumer 3 получает C — обрабатывает 2 секунды

Порядок завершения:
B → C → A

1. Использовать Single Active Consumer #

Когда необходим строгий порядок обработки одной очереди, можно включить Single Active Consumer, или SAC:

Queue
   ├── Consumer 1 — ACTIVE
   ├── Consumer 2 — standby
   └── Consumer 3 — standby

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

При объявлении очереди:

channel.queue_declare(
    queue="order_events",
    durable=True,
    arguments={
        "x-single-active-consumer": True,
    },
)

При этом нужно учитывать внутреннюю параллельность приложения. Даже если RabbitMQ использует одного Consumer-а, тот может передавать сообщения нескольким потокам или процессам:

RabbitMQ SAC
      ↓
один Consumer
      ↓
thread/process pool
├── task A
├── task B
└── task C

Тогда порядок завершения снова нарушится. Для строгой последовательности обработка должна быть однопоточной:

одна очередь
+
один активный Consumer
+
одна выполняемая операция одновременно

Например, для отдельного Celery Worker:

celery -A app worker \
    -Q ordered_orders \
    --concurrency=1

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

2. Разделить сообщения по ключу #

Обычно лучший вариант — не создавать одну глобально последовательную очередь, а партиционировать сообщения по ключу сущности:

hash(order_id) % 4

order_id=42 → Queue 2
order_id=91 → Queue 3
order_id=17 → Queue 1
order_id=66 → Queue 2

Все события одного заказа всегда попадают в одну очередь:

Queue 2:
OrderCreated(42)
OrderPaid(42)
OrderShipped(42)

Каждая очередь имеет одного активного Consumer-а:

             ┌── Queue 0 → SAC Consumer
Producer ────┼── Queue 1 → SAC Consumer
             ├── Queue 2 → SAC Consumer
             └── Queue 3 → SAC Consumer

RabbitMQ предоставляет x-modulus-hash exchange, который стабильно маршрутизирует одинаковый routing key в одну очередь. В документации этот механизм прямо предлагается сочетать с SAC для сохранения порядка по конкретному ключу при параллельной обработке разных ключей.

routing_key = order_id

Преимущества:

• события одной сущности последовательны;

• разные сущности обрабатываются параллельно;

• можно масштабироваться числом партиций;

• нет глобальной блокировки системы.

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

3. Добавлять номер версии или последовательности #

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

{
  "event_id": "9548309c-2162-4d1c-999f-fc15cedb0807",
  "aggregate_id": "order-42",
  "aggregate_version": 7,
  "event_type": "OrderPaid"
}

Следующее событие:

{
  "aggregate_id": "order-42",
  "aggregate_version": 8,
  "event_type": "OrderShipped"
}

Consumer хранит последнюю обработанную версию:

order-42:
last_processed_version = 7

Правила обработки:

version == last_version + 1
→ обработать

version <= last_version
→ дубликат или устаревшее сообщение, пропустить

version > last_version + 1
→ обнаружен пропуск

Пример:

def handle_event(event: OrderEvent) -> None:
    with database.transaction():
        state = repository.lock(event.aggregate_id)

        if event.version <= state.last_event_version:
            return

        if event.version != state.last_event_version + 1:
            raise EventSequenceGap(
                expected=state.last_event_version + 1,
                received=event.version,
            )

        apply_event(state, event)
        state.last_event_version = event.version

        repository.save(state)

Это не заставляет брокер доставлять сообщения правильно, но позволяет Consumer-у:

• обнаруживать нарушение порядка;

• не применять устаревшее событие;

• распознавать дубликаты;

• не повреждать состояние.

4. Сделать операции устойчивыми к устаревшим сообщениям #

Иногда последовательность можно защитить условным обновлением:

UPDATE orders
SET
    status = 'shipped',
    version = 8
WHERE id = 42
  AND version = 7;

Если обновлена одна строка, событие применено:

version 7 → version 8

Если обновлено ноль строк:

состояние уже новее;
предыдущее событие отсутствует;
событие доставлено повторно.

Такой optimistic concurrency control защищает от применения события к неправильной версии агрегата.

Для простого обновления снимка можно принимать только более новую версию:

UPDATE product_read_model
SET
    name = :name,
    price = :price,
    source_version = :version
WHERE product_id = :product_id
  AND source_version < :version;

Тогда поздно пришедшее старое сообщение не перезапишет новые данные.

5. Использовать буфер переупорядочивания #

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

Получено:
version 5

Ожидалось:
version 4

Consumer помещает version=5 в буфер:

pending_events:
order-42:
    version 5

После получения version=4:

обработать 4
обработать сохранённое 5

Упрощённая логика:

def handle(event: Event) -> None:
    expected = get_expected_version(event.aggregate_id)

    if event.version < expected:
        return

    if event.version > expected:
        save_to_reorder_buffer(event)
        schedule_gap_check(event.aggregate_id)
        return

    process_in_sequence(event)
    process_buffered_contiguous_events(event.aggregate_id)

Обязательно нужны:

• ограниченный размер буфера;

• TTL;

• тайм-аут ожидания пропущенного события;

• DLQ или reconciliation при незакрытом разрыве;

• метрики количества gaps.

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

6. Не возвращать сообщение бесконечно в основную очередь #

При строгом порядке возникает проблема:

A → B → C

A завершилось ошибкой:

A → NACK requeue
B → успешно
C → успешно
A → повторная доставка

Теперь порядок нарушен.

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

Если порядок критичен, есть несколько вариантов.

Остановить обработку очереди до исправления A #

A failed
   ↓
Consumer прекращает брать новые сообщения
   ↓
A retry
   ↓
после успеха продолжаются B и C

Это сохраняет строгий порядок, но одно проблемное сообщение блокирует очередь.

Заблокировать только конкретный ключ #

order-42 заблокирован из-за ошибки
order-91 продолжает обрабатываться
order-15 продолжает обрабатываться

Для этого нужна партиционированная обработка или диспетчеризация по aggregate_id.

Переместить проблемное сообщение в DLQ #

A failed → DLQ
B и C продолжаются

Это повышает доступность, но строгий порядок уже потерян. Подходит только если B и C умеют обнаружить отсутствующую версию и не применять её автоматически.

7. Проектировать события так, чтобы порядок был менее важен #

Плохое событие:

{
  "product_id": 42,
  "operation": "increase_price",
  "amount": 100
}