AI Как и когда использовать Celery — мой опыт на проде

[AI]

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

• 
  что такое Celery,
  
  
• 
  когда он реально нужен,
  
  
• 
  и какие подводные камни есть на проде.
  

Что такое Celery (очень кратко)


Celery — это таск-менеджер. Он позволяет выполнять задачи в фоне, вне основного потока вашего приложения.

Если просто:

• 
  У вас есть задача → она ставится в очередь.
  
  
• 
  Специальный процесс (воркер) берёт её из очереди и выполняет.
  
  
• 
  Можно масштабировать воркеры, следить за статусом и перезапускать упавшие задачи.
  

Кейс: внешняя AI-API и очередь из пользователей


У меня была такая ситуация:

• 
  Пользователь отправляет запрос на генерацию ответа → я отправляю его в внешнюю API (работа с ИИ).
  
  
• 
  Один запрос к API = ~10 секунд.
  
  
• 
  Если 10 человек отправляют запрос одновременно — последний ждёт 100 секунд. Жуть.
  

Попытка №1 — asyncio


Я подумал: "Окей, давай сделаем asyncio":

async def call_api():
response = await aiohttp.request(...)


Но это не решает проблему:
– asyncio не даст настоящей параллельности,
– API — внешняя, я всё равно просто жду ответ.

Вердикт: не подходит.

Попытка №2 — multiprocessing


Да, можно было сделать так:

from multiprocessing import Pool

with Pool(10) as pool:
results = pool.map(call_api, data)


Это уже ближе — multiprocessing создаёт настоящие процессы.
Идеально для I/O-задач, которые не грузят CPU.

Но такой код нужно писать и отлаживать вручную, нет очередей, нет мониторинга, нет ретраев.

Победа — Celery + Redis


Вспомнил про Celery. Подходит идеально:

• 
  Очереди, воркеры, мониторинг
  
  
• 
  Умеет ретраить, отменять, отслеживать результат
  
  
• 
  Поддержка Redis и RabbitMQ как брокеров
  

pip install celery redis

tasks.py:

from celery import Celery

app = Celery('my_app', broker='redis://localhost:6379/0')

@app.task
def call_ai_api(data):
# Вызов внешней API
...

Брокеры: Redis vs RabbitMQ

• 
  Redis — просто, быстро, подойдёт для 80% случаев.
  
  
• 
  RabbitMQ — более надёжен под нагрузкой, гибкие маршруты, подтверждение доставки.
  
  
• 
  Amazon SQS / Kafka — если у вас облако и распределёнка.
  

Я выбрал Redis — он уже стоял.

Мониторинг — Celery Flower


pip install flower
celery -A celery_app flower

Flower — веб-интерфейс. Показывает:
– Список задач,
– Аргументы,
– Время выполнения,
– Ошибки.

⚠️ Но: аргументы и результаты обрезаются на фронте. Ограничение в UI, данные всё равно есть в памяти. Решается кастомизацией или логированием.

Грабли на проде: pool и нагрузка


Когда пошла реальная нагрузка (~10+ запросов в секунду), Celery не справлялся.

Разобрался: Celery использует execution pools — механизм запуска воркеров. Есть три:

• 
  prefork — форкает процессы (по умолчанию, надёжно),
  
  
• 
  eventlet / gevent — зелёные потоки, нужны async-таски,
  
  
• 
  solo — один поток, для дебага.
  

Я попробовал --pool=gevent, но всё падало.
В итоге сделал так:

celery -A celery_app worker --concurrency=30 --pool=prefork

И всё заработало как часы.

Советы напоследок

• 
  Не делайте таски async def — Celery их не любит.
  
  
• 
  Если вызываете async-функцию — используйте asyncio.run(), но следите за loop-ами.
  
  
• 
  Храните логи отдельно или сохраняйте в файлы — Flower может обрезать результаты и после его рестарта все трет.
  
  
• 
  Настройте ретраи, таймауты, acks_late, чтобы не терять задачи.
  

Что дальше?


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

Если вы работаете с внешними API или тяжелыми задачами — Celery реально решает проблему.
Но как и всегда: магии нет, нужно понимать, как оно работает "внутри".

Хочешь — могу нарисовать схемку (как идёт запрос → очередь → воркер → API → ответ) или таблицу сравнения режимов исполнения. Нужно?