Как мы переписывали логику очередей: Celery => aio-pika => FastStream

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

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 расход оперативной памяти быстро рос пропорционально потоку сообщений. При высоком темпе обработки воркер начинал использовать значительно больше ОЗУ, чем требовалось самой бизнес-логике, что вынуждало выделять избыточные ресурсы.
  
• Управление соединениями и heartbeat — Celery скрывает многие детали брокера, из‑за чего при сетевых сбоях восстановление происходило с задержками, а тонкая настройка consumer_timeout, broker_transport_options была сложной и плохо документированной.
  
• Избыточность — Как можно заметить мы используем минимальные дополнительные конфигурации Celery.
  
• Падение контейнера на битом сообщений — Будет не совсем справедливо относить это полностью к минусам самого Celery, это больше камень в наш огород. Но опыт есть опыт. Если воркер обрабатывал невалидное сообщение в плане структуры, то падал весь контейнер, при этом сообщение консьюмилось. Что приводило к непониманию “куда делось сообщение и что с ним произошло?”.

• Ручной retry — Декоратор поверх колбэка, имеющий свой счётчик кол-во повторов при возникновений ошибок.
  
• LivenessProb — Кастомный на уровне ядра aio-pika представлен чуть ниже.
  
• Трейсинг — вручную оборачиваем вызовы в OpenTelemetry спаны, передаём trace context в заголовках AMQP при ретраях.

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)

• 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 = 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:

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