TaskIQ: современная альтернатива Celery для асинхронных задач в Python

Что такое TaskIQ

TaskIQ — это современный Python task-queue, ориентированный на async/await. Он изначально заточен под asyncio и под фреймворки вроде FastAPI, поддерживает несколько брокеров сообщений (Redis, RabbitMQ, NATS, ZeroMQ) и в целом настраивается проще, чем Celery.

Основные возможности:

• async/await — API рассчитан на асинхронный код; sync-задачи тоже поддерживаются (thread/process pool).
   
• Несколько брокеров — Redis, RabbitMQ, NATS, ZeroMQ; community: PostgreSQL, SQS, YDB, Kafka.
   
• scheduler — встроенный планировщик (cron, interval schedules, dynamic scheduling через Redis).
   
• Dependency Injection — TaskiqDepends() (как в FastAPI), переиспользование зависимостей между HTTP и задачами.
   
• retry — SimpleRetryMiddleware или SmartRetryMiddleware (middleware, подключается явно).
   
• Мониторинг — Prometheus (taskiq[metrics]) и OpenTelemetry (taskiq[opentelemetry], 0.12+).
   
• acknowledgements — --ack-type (when_saved по умолчанию), per-task ack_type, manual через Context.ack().
   
• type casts — автоматический парсинг аргументов по type hints (Pydantic, dataclasses).

Как устроен TaskIQ

Схема выполнения задачи выглядит так:

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

Какие задачи лучше подходят для TaskIQ

TaskIQ лучше всего подходит для IO-bound и сетевых задач. Благодаря асинхронной архитектуре воркер не простаивает во время ожидания ответа от API, базы данных или другого внешнего сервиса. Пока одна задача ожидает завершения операции ввода-вывода, event loop может переключиться на выполнение других задач, что позволяет эффективнее использовать ресурсы.

Memory-heavy задачи TaskIQ не «лечит» сам по себе: async не снижает потребление RAM. Такие задачи имеет смысл выносить в отдельные воркеры или process pool и контролировать лимиты памяти на уровне инфраструктуры.

С CPU-bound задачами ситуация иная. Вычисления упираются в GIL и могут блокировать event loop. Для sync CPU-задач используйте --use-process-pool; для async — отдельные воркеры/очереди, чтобы тяжёлые задачи не мешали IO-bound.

Основные возможности TaskIQ

TaskIQ предоставляет все основные механизмы для построения системы фоновых задач: работу с разными брокерами сообщений, декларативное описание задач, управление воркерами и запуск задач по расписанию.
 

Поддержка нескольких брокеров

TaskIQ не привязывается к одному брокеру сообщений. Можно выбрать подходящий вариант в зависимости от требований проекта: скорости, надежности, архитектуры и инфраструктуры.

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

Декларация задач

В TaskIQ задачи описываются декоратором @broker.task — подход похож на Celery. Декоратор регистрирует sync или async функцию как задачу. Sync выполняется в thread pool (IO) или process pool (CPU, флаг --use-process-pool).
 

```python
from taskiq_redis import ListQueueBroker, RedisAsyncResultBackend

broker = ListQueueBroker(url="redis://localhost:6379/0").with_result_backend(
RedisAsyncResultBackend(redis_url="redis://localhost:6379/1")
)

@broker.task
async def send_notification(user_id: int) -> None:
...

# await send_notification.kiq(user_id)
# CLI: taskiq worker myapp.broker:broker myapp.tasks --workers 2 --max-async-tasks 10
```

Интеграция с FastAPI

TaskIQ интегрируется с FastAPI через пакет taskiq-fastapi. Вызов taskiq_fastapi.init(broker, "myapp.main:app") подключает брокер к приложению.

Брокер нужно явно стартовать и останавливать. Предпочтительный способ — lifespan FastAPI:
 

@asynccontextmanager
async def lifespan(app: FastAPI):
    await broker.startup()
    yield
    await broker.shutdown()
app = FastAPI(lifespan=lifespan)

Дополнительные возможности TaskIQ

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

Dependency Injection

DI через TaskiqDepends() — по аналогии с FastAPI. Зависимости можно переиспользовать между HTTP-handlers и задачами (с учётом ограничений: Request в задаче — mock, не тот же объект, что в handler).

Например, таким способом можно подключать настройки приложения, клиентов внешних API или сервисы для работы с базой данных.
 

State и Context

TaskiqState — shared state воркера: инициализируется в @broker.on_event(WORKER_STARTUP) (connection pools, clients). Из задачи доступен через Context (TaskiqDepends()).

Context также даёт requeue() и reject() — вернуть задачу в очередь или отбросить без повторного выполнения. Manual ack — через await context.ack() (требует AckableMessage у broker).
 

Smart Retry

Smart Retry — middleware (SmartRetryMiddleware), подключается явно к брокеру. Автоматически возвращает неуспешные задачи в очередь с настраиваемым backoff и jitter. Не включён по умолчанию.

В TaskIQ можно настроить:

• количество попыток;
• задержку между повторными запусками;
• случайное смещение времени ожидания (jitter).

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

Чтобы TaskIQ работал эффективно в production-среде, важно правильно организовать задачи, настроить обработку ошибок и контролировать состояние системы. 

• Разделяйте задачи по типу нагрузки — отдельный брокер для CPU-bound, отдельный для IO-bound (cpu_tasks.py, io_tasks.py).
   

• Оборачивайте задачи в try-except и подключайте SmartRetryMiddleware. Для Redis используйте taskiq-redis (Streams/ListQueue) и настраивайте --ack-type (when_saved, when_executed, manual). Надёжность зависит от брокера и ack-политики, а не только от «Redis vs RabbitMQ».
   

• Используйте async/await везде, где это возможно, а не только в отдельных задачах.
   

• Настройте мониторинг — Prometheus или OpenTelemetry плюс структурированные логи.
   

• Вызывайте await broker.startup() в клиенте и broker.shutdown() при остановке.
   

• Настройте timeouts для долгих задач: @broker.task(timeout=30) или .kicker().with_labels(timeout=30).kiq().
   

• Для sync CPU-bound — --use-process-pool; для sync IO — thread pool (default).
   

• --max-prefetch имеет смысл только с брокерами, поддерживающими ack.
   

• Установите uvloop — TaskIQ подхватит его автоматически, если пакет есть.

Меньше production-кейсов, чем у Celery

TaskIQ активно развивается (документация на taskiq-python.github.io заметно выросла), но community и production-историй всё ещё меньше, чем у Celery (~2.2k vs ~28k stars на GitHub).

• меньше готовых решений для нестандартных edge cases;
• меньше battle-tested примеров из крупных prod-систем;
• часть архитектурных решений придётся валидировать своими нагрузочными тестами.

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

Scheduler не сохраняет состояние между перезапусками

У встроенного планировщика TaskIQ есть ограничение: он не хранит информацию о предыдущих запусках между перезапусками.

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

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

В системах, где важно гарантировать однократное выполнение запланированных задач, это необходимо учитывать. Например, в Celery есть Database Scheduler, который сохраняет состояние расписания в базе данных. В TaskIQ подобного встроенного механизма пока нет.
 

Scheduler работает только в одном экземпляре

TaskIQ Scheduler рассчитан на работу в одном экземпляре. Запустить несколько планировщиков одновременно для повышения отказоустойчивости нельзя без дополнительной архитектуры вокруг него.

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

TaskIQ vs Celery

Celery — один из самых популярных инструментов для выполнения фоновых задач в Python. Проект существует более 15 лет, имеет большое сообщество, множество интеграций и активно используется в production-системах, особенно в проектах на Django.

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

TaskIQ появился позже и изначально создавался с учетом современных асинхронных подходов Python. Его основная идея — нативная работа с async/await и удобная интеграция с асинхронными фреймворками.

Основные отличия TaskIQ и Celery привёл в таблице ниже.
 

Производительность TaskIQ и Celery

Мы прогнали сравнение на внутреннем стенде. Это не официальный benchmark — результат зависит от железа, брокера и настроек воркеров.
 

Наибольшая разница наблюдается в IO-bound сценариях: запросах к API, работе с внешними сервисами и другими операциями ожидания. Это связано с async-first архитектурой TaskIQ — пока одна задача ожидает ответа, воркер может выполнять другие операции.

Для CPU-bound задач преимущество меньше, поскольку такие операции ограничиваются особенностями Python и GIL (Global Interpreter Lock). Асинхронность не ускоряет сами вычисления, поэтому прирост зависит от конкретной архитектуры выполнения.

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

Отдельно стоит учитывать влияние дополнительных компонентов. Например, при использовании TaskIQ Admin производительность может снижаться из-за дополнительной нагрузки от мониторинга и middleware. Поэтому перед внедрением в production рекомендуется проводить собственные нагрузочные тесты.

Когда выбирать Celery и TaskIQ

Выбор между инструментами зависит не только от производительности, но и от архитектуры проекта.
 

Стоит ли мигрировать с Celery на TaskIQ

Если задачи уже вынесены в отдельные функции, технический переход умеренный, но это не «замена декоратора»: нужно сменить .delay() → .kiq(), настроить broker startup/shutdown, Celery Beat → TaskIQ Scheduler, retry через middleware. Миграция оправдана не всегда.

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

Переход на TaskIQ имеет смысл, если текущая архитектура действительно ограничивает развитие проекта. Например:

• приложение построено на FastAPI и активно использует asyncio;
• Celery усложняет работу с асинхронным кодом;
• требуется более удобная интеграция зависимостей;
• система состоит из микросервисов с большим количеством сетевых операций.

Главный аргумент в пользу TaskIQ — современный асинхронный подход и удобная работа с типизированным Python-кодом. Но для команды, которая уже хорошо работает с Celery, этого может быть недостаточно для полноценного перехода.

Во многих проектах используется один и тот же брокер — например Redis или RabbitMQ. Поэтому широкий выбор брокеров TaskIQ не всегда становится решающим преимуществом.