آموزش Celery در Django؛ Worker، زمان‌بندی و Retry امن

Celery کارهای زمان‌بر Django را از چرخه درخواست وب جدا می‌کند و به Worker می‌سپارد. ساخت گزارش، پردازش فایل و فراخوانی دسته‌ای مدل هوش مصنوعی نباید لزوماً کاربر را پشت یک درخواست HTTP طولانی نگه دارد. با این حال، صف تضمین نمی‌کند عملیات خارجی فقط یک بار انجام شود؛ طراحی Retry و ثبت وضعیت همچنان بر عهده برنامه است.

در این آموزش اتصال Django و Celery، یک Task کم‌خطر، زمان‌بندی Beat و نکات انتشار روی پلتفرم ابری پاستا را مرور می‌کنیم. نمونه برای فهم چرخه اجراست و عملیات مالی، ارسال واقعی پیام یا تغییر اطلاعات مشتری انجام نمی‌دهد. برای راه‌اندازی خود برنامه، راهنمای استقرار Django را کنار آن داشته باشید.

وب، Broker، Worker و Beat چه فرقی دارند؟

وب درخواست کاربر را می‌گیرد و پس از کنترل مجوز، کار را ثبت می‌کند. Broker پیام را به مصرف‌کننده می‌رساند. Worker تابع Task را اجرا می‌کند. Beat زمان‌بند است و در موعد مقرر پیام می‌سازد؛ خودش جای Worker را نمی‌گیرد. Result Backend نیز در صورت نیاز، وضعیت یا خروجی اجرای Task را نگه می‌دارد و با دیتابیس اصلی کسب‌وکار یکی نیست.

Redis می‌تواند Broker این نمونه باشد. دیتابیس سفارش یا گزارش همچنان باید منبع اصلی وضعیت محصول بماند. اگر خروجی بزرگ است، فایل را در ذخیره مناسب نگه دارید و در نتیجه Task فقط شناسه یا مسیر مجاز را ثبت کنید. متن کامل فایل، کلید مدل یا اطلاعات شخصی را در پیام صف نگذارید.

۱. وابستگی‌ها و ساختار پروژه

فرض می‌کنیم پروژه Django با بسته تنظیمات config و اپ jobs دارید. Python و نسخه Celery باید با هم و با Django سازگار باشند. پس از نصب، نسخه‌های آزمایش‌شده را در فایل وابستگی پروژه تثبیت کنید؛ نصب نسخه جدید بدون آزمون روی production مناسب نیست.

python -m pip install "celery[redis]"
# Existing Django project:
# manage.py
# config/settings.py
# jobs/tasks.py

اپ jobs باید در INSTALLED_APPS باشد. فایل‌های بعدی را به پروژه موجود اضافه کنید و تنظیمات قبلی را بی‌دلیل جایگزین نکنید. راهنمای رسمی اتصال Celery و Django مرجع قرارداد بارگذاری Taskهاست.

# config/celery.py
import os
from celery import Celery

os.environ.setdefault("DJANGO_SETTINGS_MODULE", "config.settings")
app = Celery("config")
app.config_from_object("django.conf:settings", namespace="CELERY")
app.autodiscover_tasks()

# Add to config/__init__.py
from .celery import app as celery_app
__all__ = ("celery_app",)

۲. اتصال Redis و قرارداد سریال‌سازی

نشانی Redis را از متغیر محیطی بخوانید. برای پیام‌های این نمونه JSON کافی است. فرمت‌هایی که اجرای آبجکت دلخواه را ممکن می‌کنند، برای ورودی غیرقابل اعتماد مناسب نیستند. دسترسی شبکه و Credential Broker باید محدود باشد؛ داشتن نام صف محرمانه، جای کنترل شبکه و هویت را نمی‌گیرد.

# Add to config/settings.py
import os

CELERY_BROKER_URL = os.environ["CELERY_BROKER_URL"]
CELERY_ACCEPT_CONTENT = ["json"]
CELERY_TASK_SERIALIZER = "json"
CELERY_RESULT_SERIALIZER = "json"
CELERY_TIMEZONE = "Asia/Tehran"
CELERY_ENABLE_UTC = True
CELERY_TASK_IGNORE_RESULT = True
CELERY_WORKER_PREFETCH_MULTIPLIER = 1
CELERY_BROKER_CONNECTION_RETRY_ON_STARTUP = True

در این نمونه نتیجه پایدار Task نگه‌داری نمی‌شود؛ برای محصولی که وضعیت گزارش را نشان می‌دهد، رکورد وضعیت در دیتابیس برنامه طراحی کنید یا Result Backend را با سیاست نگهداری مشخص اضافه کنید. انتخاب prefetch پایین شروع محافظه‌کارانه‌ای برای کارهای طولانی است، نه تنظیم بهینه همه پروژه‌ها.

طبق راهنمای Redis در Celery، Visibility Timeout و رفتار تحویل دوباره باید با مدت Task هماهنگ باشد. افزایش این زمان بدون تحلیل، بازیابی کار گم‌شده را هم به تأخیر می‌اندازد. کار بسیار طولانی را در صورت امکان به واحدهای کوچک‌تر با پیشرفت ثبت‌شده تقسیم کنید.

۳. یک Task بی‌خطر بسازید و اجرا کنید

فایل jobs/tasks.py زیر متن نمونه را اعتبارسنجی و تعداد کلمات را برمی‌گرداند. برای مشاهده چرخه، در آزمون محلی می‌توانید نتیجه را از خروجی Worker بررسی کنید. داده واقعی و حساس را برای آزمایش اتصال استفاده نکنید.

# jobs/tasks.py
from celery import shared_task

@shared_task(name="jobs.count_words")
def count_words(text):
    if not isinstance(text, str) or len(text) > 10000:
        raise ValueError("Invalid demo input")
    return {"words": len(text.split())}

@shared_task(name="jobs.heartbeat")
def heartbeat():
    return {"ok": True}

# Terminal 1: use a Linux environment supported by Celery.
celery -A config worker --loglevel=INFO --concurrency=2
# Terminal 2:
python manage.py shell -c "from jobs.tasks import count_words; print(count_words.delay('سلام از صف کار').id)"

شناسه چاپ‌شده، پذیرش پیام را نشان می‌دهد؛ نتیجه نهایی نیست. Worker باید Task را دریافت و اجرا کند. اگر پیام Unregistered Task می‌بینید، نام اپ، autodiscover و نسخه کد Worker را بررسی کنید. برای Windows، محیط Linux یا WSL سازگار انتخاب کنید و محدودیت پلتفرم نسخه Celery را بخوانید.