← назад к разделу

Асинхронный сервис живёт сетью: база, кеш, брокер, соседние сервисы. Клиенты к ним это объекты с пулами соединений, фоновыми задачами и своими таймаутами, и большинство странностей «иногда 502 от соседа», «кончились соединения к базе», «утечка файловых дескрипторов» начинаются с неправильного обращения с клиентом. Разберём три самых частых клиента и общие правила, которые к ним применимы.

Обязательно

Правило одно: клиент на процесс

Клиент создают один раз при старте приложения и закрывают при остановке. Создание клиента на каждый запрос означает новый пул, новое TCP-рукопожатие и TLS-переговоры на каждый вызов, а незакрытый клиент оставляет открытые сокеты. Во FastAPI место для этого lifespan:

import httpx, asyncpg
from redis.asyncio import Redis

@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.http = httpx.AsyncClient(
        base_url=settings.catalog_url,
        timeout=httpx.Timeout(5.0, connect=2.0),
        limits=httpx.Limits(max_connections=100, max_keepalive_connections=20),
    )
    app.state.pg = await asyncpg.create_pool(settings.pg_dsn, min_size=5, max_size=20, command_timeout=10)
    app.state.redis = Redis.from_url(settings.redis_url, socket_timeout=1, socket_connect_timeout=1)
    try:
        yield
    finally:
        await app.state.http.aclose()
        await app.state.pg.close()
        await app.state.redis.aclose()

Обработчики получают клиентов через зависимости, которые читают app.state, или через контейнер зависимостей, как показано в статье про Dependency Injection во FastAPI. Порядок закрытия обратный порядку создания, и закрытие идёт после того, как обработчики в полёте завершились, иначе последние запросы получат закрытый клиент.

httpx: лимиты и таймауты

httpx.AsyncClient держит пул соединений с keep-alive. Умолчания проверены на httpx 0.28: до 100 одновременных соединений, из них до 20 живут между запросами, таймаут 5 секунд на каждую фазу (соединение, чтение, запись, ожидание из пула). Пять секунд это много для вызова внутри обработчика с SLA в секунду, поэтому таймаут задают явно под каждого соседа, а не один на всех: httpx.Timeout(timeout=2.0, connect=0.5).

Когда пул исчерпан, следующий запрос ждёт свободного соединения до pool таймаута и падает с PoolTimeout. Это полезный сигнал: сосед отвечает медленно, и вместо роста очереди запросов сервис быстро отвечает ошибкой. Размер пула это и есть лимит параллелизма на соседа, о чём статья про таймауты и обратное давление.

Один клиент на соседа, а не один на всё: у разных соседей разные таймауты, базовые адреса, заголовки аутентификации и лимиты, и общий клиент заставляет медленного соседа занимать соединения быстрого. Повторы при сетевых ошибках делают поверх клиента (транспорт с retries для соединения, собственная обёртка для идемпотентных запросов), не в каждом вызове.

asyncpg: пул и его границы

asyncpg.create_pool по умолчанию держит от 10 до 10 соединений, то есть фиксированный пул на десять; под сервис его задают явно: min_size для прогрева, max_size от max_connections базы, делённого на число подов и воркеров. command_timeout ограничивает каждый запрос; без него зависший запрос держит соединение бесконечно.

async def get_order(pool: asyncpg.Pool, order_id: int) -> Order | None:
    async with pool.acquire() as conn:
        row = await conn.fetchrow("SELECT id, status, total FROM orders WHERE id = $1", order_id)
        return Order(**row) if row else None

Соединение берут на время одной единицы работы и возвращают через async with: acquire без release при исключении это утечка, которая исчерпает пул за минуту под нагрузкой. Транзакция живёт внутри async with conn.transaction(): и не должна содержать внешних вызовов. Соединение нельзя делить между задачами: два await conn.fetch из разных задач на одном соединении дадут ошибку another operation is in progress; каждая задача берёт своё соединение из пула. Если сервис работает через SQLAlchemy, то же самое делает AsyncSession поверх asyncpg, и об этом статья про асинхронный SQLAlchemy.

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

redis.asyncio: таймауты зависят от способа создания

redis.asyncio.Redis держит пул соединений, в redis-py 8 по умолчанию до 100; при исчерпании новое обращение получает ConnectionError, и это лучше бесконечного роста соединений до maxclients сервера. С таймаутами ловушка: проверено на redis-py 8.1, что Redis() без аргументов ставит socket_timeout и socket_connect_timeout в 5 секунд, а Redis.from_url(url) оставляет их пустыми, и обращение к зависшему Redis через такой клиент блокирует задачу навсегда. В сервисе таймауты задают явно и в доли секунды независимо от способа создания, потому что кеш, который отвечает дольше базы, бесполезен.

Один объект Redis на процесс безопасен для одновременного использования из многих задач: каждая команда берёт соединение из пула и возвращает. Исключение составляют pipeline и pubsub: они держат соединение за собой и не должны делиться между задачами. Подробнее о приёмах работы с Redis из Python в статье про redis-py.

Брокеры: клиент с фоновыми задачами

У aiokafka продюсер и потребитель это не просто пул: после start() они запускают фоновые задачи (отправка пакетов, heartbeat группы), и забытый stop() оставляет их висеть, а при остановке приложения это выглядит как «процесс не завершается». Продюсер один на процесс, стартует в lifespan; потребитель это долгоживущая задача в группе задач lifespan, которая по отмене вызывает stop() и фиксирует смещения. Как устроен такой потребитель, разбирает статья про потребителя Kafka на Python.

Что общего у всех клиентов

Четыре правила, которые повторяются от библиотеки к библиотеке. Создание один раз, закрытие в обратном порядке. Явные таймауты на каждую фазу, потому что умолчания либо слишком щедрые, либо отсутствуют. Явный размер пула, выведенный из лимитов сервера и числа процессов. И отказ делить объект с состоянием (соединение, транзакцию, pipeline, подписку) между задачами: делится клиент с пулом, а не то, что из пула взято.

Пятое, не про код: клиент должен быть виден в метриках. Число занятых соединений в пуле, время ожидания соединения, ошибки таймаутов по соседу: без них исчерпание пула выглядит как «сервис тормозит без причины». Что мерить и как, в статье про метрики на Python.

Дополнительно: при первом чтении можно пропустить

Глубже: закрытие под нагрузкой и клиенты в тестахрасширенное

При остановке пода порядок такой: uvicorn перестаёт принимать новые соединения, ждёт завершения обработчиков в полёте, выполняет код после yield в lifespan. Если клиент закрыть до того, как обработчики закончились, они получат RuntimeError: client has been closed; lifespan выполняется после ожидания обработчиков, так что порядок правильный сам по себе, пока клиентов не закрывают из других мест (сигнальных обработчиков, фоновых задач). Бюджет на закрытие ограничен terminationGracePeriodSeconds пода и таймаутом uvicorn, поэтому aclose клиентов не должен ждать бесконечно: у httpx закрытие быстрое, у asyncpg.Pool.close() оно ждёт возврата всех соединений, и зависшая транзакция задержит остановку, отсюда ещё одна причина command_timeout.

В тестах клиентов не создают заново на каждый тест: фикстура уровня сессии или модуля даёт один клиент, а изоляцию обеспечивают данные, не соединения. Для httpx есть транспорт ASGITransport, который ходит в приложение без сети, и respx, который перехватывает исходящие запросы к соседям; об этом статья про тесты асинхронного кода.

Коротко

  • Клиент создают один раз в lifespan и закрывают в обратном порядке; клиент на запрос это новые TCP и TLS на каждый вызов и утечки сокетов.
  • httpx: по умолчанию 100 соединений, 20 keep-alive, 5 секунд на фазу; таймауты и клиент задают под каждого соседа отдельно; PoolTimeout это сигнал о медленном соседе.
  • asyncpg: пул по умолчанию 10 фиксированных соединений, задавать max_size от лимита базы и command_timeout; соединение берут через async with pool.acquire() и не делят между задачами.
  • redis.asyncio: пул до 100 по умолчанию; from_url создаёт клиент без таймаутов сокета, Redis() с пятисекундными; ставить таймауты в доли секунды явно; pipeline и pubsub не делить.
  • Продюсер и потребитель aiokafka запускают фоновые задачи: забытый stop() мешает остановке процесса.
  • Общие правила: один раз создать, явные таймауты, явный размер пула, не делить объекты с состоянием, пул виден в метриках.
  • При остановке клиенты закрываются после обработчиков в полёте; закрытие пула ждёт возврата соединений, поэтому запросы нуждаются в таймаутах.

Что почитать дальше