Наш путь активной работы с очередями RabbitMQ начался с классического Celery. Осознав критичность низкоуровневого контроля системы, принялись работать с aio-pika. Но и этот уровень слишком местами сложный (далее расскажу почему), и нашли отличное решение, на текущий момент, в лице FastStream. Сразу оставлю такую пометку, что каждый инструмент подходит для решения своей задачи. Мы больше хотели сделать акцент на удобство и скорость разработки относительно затрачиваемого времени на миграции решений.

N.B.: Код возможно покажется неоптимальным или старым. Это всё наш дорогой Легаси.

Постановка задачи

Наша система построена на основе микросервисов, работающих с RabbitMQ. Внутри - обычный асинхронный код для похода на внешние API и в БД.

Требования:

• Надежный консьюминг - это для нас критично, чтобы сообщение шло по всему флоу и нигде не останавливалось без причин. Если ошибка падает, то это должно отражаться в 3 местах: БД, логи и метрики.

• Ретраи при ошибках обработки.

• Трейсинг - поддержка OpenTelemetry.

• Мониторинг сервиса через healthcheck’и.

• Prometheus метрики.

Решение №1: Celery как консьюмер

Почему Celery

Celery — классический инструмент для фоновых задач, знакомый большинству Python-разработчиков. Из коробки: декларативное описание задач, ретраи с экспоненциальной задержкой, хранение результатов, мониторинг через Flower, интеграции с фреймворками. Логика проста: пишешь @app.task, запускаешь воркер — и сообщения из очереди начинают обрабатываться.

Как мы его использовали

Мы не отправляли задачи из кода в духе my_task.delay(), а настраивали Celery на прослушивание внешней очереди, куда сообщения попадали от других систем. По сути, Celery выступал как consumer: подключался к брокеру, забирал сообщения, десериализовал и передавал в наши обработчики. Настройки вроде max_retries, default_retry_delay, countdown позволяли гибко управлять поведением при сбоях. Важно ещё подсветить, что результат всегда игнорируется с помощью параметра ignore_result=True поскольку все результаты записываются в БД.

Пример инициализации воркера:

def create_app(
    name,
    broker,
    include,
    backend=None,
    task_queues=None,
    liveness_probe=1,
    update_period=60,
    watcher_config={},
):
    # Создание само приложение + наложение дополнительных конфигурации
    app = Celery(name, broker=broker, include=include, backend=backend)
    app.conf.update(
        result_expires=120,
    )
    add_without_heartbeat_argument = Option(
        ("--without-heartbeat",),
        default=True,
    )

    app.user_options["worker"].add(add_without_heartbeat_argument)
    if task_queues is not None:
        app.conf.task_queues = task_queues

    if liveness_probe:
        add_update_period_argument = Option(
            ("--update-period",),
            default=update_period,
        )

        HEARTBEAT_FILE.touch()

        app.user_options["worker"].add(add_update_period_argument)

        # Добавление кастомной livenessProbe для K8s
        app.steps["worker"].add(LivenessProbe)
    
    # инициализация трейсинга
    with_tracing = watcher_config.get("with_tracing")
    if with_tracing:
        tracing_exporters = watcher_config.get("tracing_exporters", ())
        signals.worker_process_init.connect(
            init_celery_tracing(app_name=name, tracing_exporters=tracing_exporters),
            weak=False,
        )
    return app

С чем столкнулись

Celery проектировался как система передачи сообщений между системами. Из-за чего столкнулись со следующими проблемами:

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

• Управление соединениями и heartbeat — Celery скрывает многие детали брокера, из‑за чего при сетевых сбоях восстановление происходило с задержками, а тонкая настройка consumer_timeout, broker_transport_options была сложной и плохо документированной.

• Избыточность — Как можно заметить мы используем минимальные дополнительные конфигурации Celery.

• Падение контейнера на битом сообщений — Будет не совсем справедливо относить это полностью к минусам самого Celery, это больше камень в наш огород. Но опыт есть опыт. Если воркер обрабатывал невалидное сообщение в плане структуры, то падал весь контейнер, при этом сообщение консьюмилось. Что приводило к непониманию “куда делось сообщение и что с ним произошло?”.

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

Решение №2: aio-pika — близко к железу

Идея

Перейти на чистый AMQP, отказавшись от посредника в виде Celery. aio-pika — асинхронная библиотека для RabbitMQ, предоставляющая прямой доступ к каналам, обменникам, очередям. Код становится минимальной прослойкой над протоколом: сами управляем подписками, подтверждениями (ack/nack), префетчем, реконнектами.

Что переписали

Написали небольшой собственный “фреймворк” консьюминга: асинхронный раннер, который при запуске создаёт соединение, открывает канал, объявляет очереди с нужными параметрами durable/exclusive, подписывается на них, а в колбэке вызывает наши обработчики.

Поверх этого появились:

• Ручной retry — Декоратор поверх колбэка, имеющий свой счётчик кол-во повторов при возникновений ошибок.

• LivenessProb — Кастомный на уровне ядра aio-pika представлен чуть ниже.

• Трейсинг — вручную оборачиваем вызовы в OpenTelemetry спаны, передаём trace context в заголовках AMQP при ретраях.

Пример LivenessProb:

class CustomRobustConnection(RobustConnection):
    def __init__(
        self,
        url: URL,
        loop: asyncio.AbstractEventLoop | None = None,
        **kwargs: Any,
    ):
        super().__init__(url=url, loop=loop, **kwargs)
        # Кастомный класс пробы, схож с тем, что выше для Celery
        self._liveness_probe = LivenessProbe()

    async def connect(self, timeout: TimeoutType = None) -> None:
        # Тут держитесь крепче. Я когда впервые это увидел, без литра кофе не смог понять как оно работает
        self._RobustConnection__connect_timeout = timeout

        if self.is_closed:
            raise RuntimeError(f"{self!r} connection closed")

        if self.reconnecting:
            raise RuntimeError(
                (
                    "Connect method called but connection "
                    f"{self!r} is reconnecting right now."
                ),
                self,
            )

        if not self._RobustConnection__reconnection_task:
            self._RobustConnection__reconnection_task = self.loop.create_task(
                self.__connection_factory(),
            )

        await self._RobustConnection__fail_fast_future
        await self.connected.wait()

    async def __connection_factory(self) -> None:
        logger.debug("Starting connection factory for %r", self)
        while not self.is_closed and not self._close_called:
            logger.debug("Waiting for connection close event for %r", self)
            await self._RobustConnection__connection_close_event.wait()

            if self.is_closed or self._close_called:
                return

            try:
                self.transport = None
                self.connected.clear()

                logger.debug("Connection attempt for %r", self)
                await Connection.connect(self, self._RobustConnection__connect_timeout)

                if not self._RobustConnection__fail_fast_future.done():
                    self._RobustConnection__fail_fast_future.set_result(None)

                logger.debug("Connection made on %r", self)
                self._liveness_probe.start()

            except CONNECTION_EXCEPTIONS as e:
                if not self._RobustConnection__fail_fast_future.done():
                    self._RobustConnection__fail_fast_future.set_exception(e)
                    return

                logger.warning(
                    'Connection attempt to "%s" failed: %s. '
                    "Reconnecting after %r seconds.",
                    self,
                    e,
                    self.reconnect_interval,
                )
                self._liveness_probe.stop()
            except Exception:
                logger.exception(
                    "Reconnect attempt failed %s. " "Retrying after %r seconds.",
                    self,
                    self.reconnect_interval,
                )
                self._liveness_probe.stop()

            await asyncio.sleep(self.reconnect_interval)

Что стало лучше

Полный контроль над жизненным циклом соединения, тонкая настройка prefetch, возможность реализовать любую логику подтверждения (например, отложенный ack после завершения цепочки действий). Нет лишних метаданных в теле сообщения — брокер передаёт ровно то, что отправил продюсер. Асинхронность нативная, работает на asyncio без костылей.

Проблемы

Каждая «плюшка» делалась вручную и со временем объём инфраструктурного кода разросся. Десятки строк для декларации очередей, логирование reconnect‑цикла, согласование формата trace‑заголовков между сервисами. Healthcheck, хоть и был создан вручную, требовал аккуратности: нужно было отслеживать состояние не только TCP-соединения, но и открытого канала.

Решение №3: FastStream — золотая середина

Основная идея

FastStream — надстройка над aio-pika (а также над NATS, Kafka), которая даёт декларативный стиль описания consumer’ов, lifespan‑хуки, встроенные механизмы: healthcheck-эндпоинт, OpenTelemetry-интеграция, метрики Prometheus.

По сути, это aio-pika, обёрнутая в лучшие практики, которые мы сами реализовывали руками в предыдущем решении. Механизм retry описан как пример использования middleware.

Что пошло так

• Healthcheck из коробки — достаточно указать FastStream объект в ASGI-приложении (через AsgiFastStream), и на /health возвращается статус брокера.

• Lifespan — менеджер контекста управляет запуском и корректной остановкой consumer’ов, повторными соединениями. Не нужно писать свои обработчики сигналов. Это касалось не только RMQ коннектов, но и например коннектов к базам данных.

• Мониторинг и трейсинг — подключение OpenTelemetry сводится к нескольким строчкам: спаны автоматически создаются для каждого обработанного сообщения, propagate context через заголовки.

• Декларативные middleware — проще внедрять кросс‑касательную логику (логирование, валидацию) без захламления бизнес-кода.

• Простота конфигурации — брокер, очереди, обменники описываются через Python-декораторы и типы, нет нужды вручную управлять каналами и ack/nack.

• Интеграция с Pydantic — Сообщение на уровне описание консьюмеров сразу пытается отвалидироваться в Pydantic модель, если это нужно.

• Dependency Injection — Как можно увидеть в примере ниже FastStream предлагает нам возможность подключение DI как в FastAPI.

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

app = AsgiFastStream(
    broker,
    # health + metrics из коробки
    asgi_routes=[
        (
            "/health",
            make_system_ping_asgi(broker, timeout=5.0, include_in_schema=False),
        ),
        ("/metrics", make_asgi_app(registry)),
    ],
    # Кастомный lifespan. Обычно здесь идёт инициализации подключений к БД
    lifespan=lifespan,
)

Было бы ещё не очень справедливо не показать, что из себя представляет broker

broker = RabbitBroker(
    settings.RMQ_URL,
    # Примеры миддлвары
    # RabbitPrometheusMiddleware из коробки для мониторинга сообщений
    # EnrichLogMiddleware и FailCatchComplexMiddleware кастомные для отслеживания логирования и ошибок 
    middlewares=(
        RabbitPrometheusMiddleware(registry=registry),
        EnrichLogMiddleware,
        FailCatchComplexMiddleware(
            ignore_routing_keys=[
                settings.RMQ_FAIL_TABLE_QUEUE,
                settings.RMQ_DASHBOARD_SETTINGS_QUEUE,
                settings.RMQ_TIMEOUT_QUEUE,
            ],
        ),
    ),
    logger=logger,
    # Кастомные парсеры и декодеры для обработки входящих сообщений
    parser=json_parser,
    decoder=decoder,
)

broker.include_router(router)

и пример как инициализируется консьюмер

from faststream import Depends
from faststream.rabbit import (
    ExchangeType,
    RabbitExchange,
    RabbitMessage,
    RabbitQueue,
    RabbitRouter,
)
from faststream.rabbit.annotations import RabbitBroker as ContextRabbitBroker

# Вместо RabbitRouter можно использовать broker.
router = RabbitRouter()
@router.subscriber(
    RabbitQueue(
        name=...,
        durable=True,
        routing_key=...,
    ),
    RabbitExchange(
        name=...,
        durable=True,
    ),
)
async def on_service_hub_message(
    message: RabbitMessage,
    # Имеется общий контекст всей системы
    broker: ContextRabbitBroker,
    # DI
    async_session=Depends(get_db_session),
) -> None:

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

Что пошло не так

• Производительность — из-за дополнительных слоёв абстракции (middleware, автоматическая обвязка спанов, встроенные ретраи) пропускная способность ниже, чем у голого aio-pika. Это для нас лишь только в теории, поскольку поток не достигает больших значений 30-40 RPS. Справедливости ради, это не относится к минусам, поскольку бенчмарки я не проводил, и фраза “поверь мне брат” меня убедила.

• Сложность отслеживания кодовой базы — Тут опять из-за дополнительных слоёв абстракции иногда заходишь внутрь посмотреть что там, и можно с лёгкостью потеряться.

Итоговое сравнение

Заключение

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

Да, возможно я выступаю в качестве адвоката этого инструмента, но что могу поделать, когда он так понравился? это вы ещё не увидели как стараюсь переписать всё на Rust.

Отойдя от жаргона и шуток повторю первую свою мысль: У каждой задачи свой инструмент. Мой — это FastStream.

Никто еще не голосовал. Воздержавшихся нет.