برای شروع آموزش Celery در جنگو کافی است Celery را به پروژه اضافه کنید، یک broker مثل Redis را بالا بیاورید، فایل celery.py را در ریشه پروژه بسازید، وظایف را با shared_task تعریف کنید، سپس worker و beat را اجرا کنید تا کارهای پس زمینه و زمان بندی شده بدون قفل کردن وب سرور انجام شوند.

چه چیزی را با Celery حل می کنیم؟

اگر درخواست های وب شما به خاطر ارسال ایمیل، پردازش تصویر، تولید گزارش یا فراخوانی API کند می شوند، آن کارها باید جدا از چرخه پاسخ HTTP اجرا شوند. Celery صفی از وظایف می سازد تا worker ها آن را غیرهمزمان پردازش کنند و با Celery Beat می توانید اجرای دوره ای وظایف را زمان بندی کنید.

پیش نیاز و انتخاب های اصلی

برای راه اندازی پایه به موارد زیر نیاز دارید:

  • Python و Django نصب و پروژه فعال
  • Celery به عنوان موتور صف
  • یک message broker (Redis یا RabbitMQ) و یک result backend

راهنمای انتخاب سریع:

  • Redis: راه اندازی ساده، مناسب شروع و بسیاری از کاربردهای وب.
  • RabbitMQ: مناسب الگوهای صف پیشرفته، اولویت دهی دقیق و مقیاس بزرگ.
  • Result backend: اگر لازم است نتیجه کار را بخوانید، Redis یا دیتابیس مناسب است. اگر فقط اجرای موفق مهم است، می توانید ذخیره نتیجه را غیرفعال کنید.

نصب و راه اندازی سریع

نصب پکیج ها

pip install celery redis
# اختیاری برای زمان بندی از طریق دیتابیس (پنل مدیریت):
pip install django-celery-beat

تنظیمات Django

در settings.py پیکربندی Celery را اضافه کنید. در این مثال از Redis روی سیستم محلی استفاده می کنیم.

# settings.py
CELERY_BROKER_URL = 'redis://127.0.0.1:6379/0'
CELERY_RESULT_BACKEND = 'redis://127.0.0.1:6379/1'

CELERY_TIMEZONE = 'Asia/Tehran'
CELERY_TASK_SERIALIZER = 'json'
CELERY_RESULT_SERIALIZER = 'json'
CELERY_ACCEPT_CONTENT = ['json']  # فقط فرمت های مجاز

# اگر می خواهید beat بر اساس تنظیمات فایل کار کند:
from celery.schedules import crontab
# مثال زمان بندی را در بخش مربوط اضافه می کنیم

# اختیاری: برای زمان بندی از طریق ادمین
INSTALLED_APPS = [
    # ...
    # 'django_celery_beat',  # در صورت استفاده از زمان بندی دیتابیسی فعال کنید
]

ساخت فایل celery.py در ریشه پروژه

اگر ساختار پروژه شما config/settings.py است، فایل celery.py را کنار settings قرار دهید.

# 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()

و در فایل __init__.py همان پوشه:

# config/__init__.py
from .celery import app as celery_app

__all__ = ('celery_app',)

تعریف اولین task

در هر اپ جنگو (مثلا myapp) یک فایل tasks.py بسازید و از shared_task استفاده کنید. نمونه زیر ایمیل خوش آمد را با تکرارپذیری و محدودیت زمان اجرا ارسال می کند.

# myapp/tasks.py
from celery import shared_task
from django.core.mail import send_mail
from django.contrib.auth import get_user_model
from django.utils import timezone

@shared_task(
    bind=True,
    autoretry_for=(Exception,),
    retry_backoff=True,            # 1s, 2s, 4s, ...
    retry_kwargs={'max_retries': 3},
    time_limit=30,                 # سقف سخت
    soft_time_limit=25             # سقف نرم برای پاک سازی منابع
)
def send_welcome_email(self, user_id):
    User = get_user_model()
    user = User.objects.get(pk=user_id)
    send_mail(
        subject='Welcome',
        message=f'Hi {user.username}, thanks for joining at {timezone.now().date()}',
        from_email='no-reply@example.com',
        recipient_list=[user.email],
        fail_silently=False
    )
    return 'ok'

نکته مهم: وظایف باید idempotent باشند. اگر به هر دلیل دوباره اجرا شدند، داده تکراری تولید نکنند. مثلا قبل از ارسال ایمیل، لاگ ارسال را بررسی کنید.

اجرای worker و تست سریع

ابتدا Redis را اجرا کنید. سپس worker را بالا بیاورید:

celery -A config worker -l info

در یک ترمینال دیگر، از طریق شل جنگو وظیفه را فراخوانی و وضعیت را بررسی کنید:

python manage.py shell

from myapp.tasks import send_welcome_email
res = send_welcome_email.delay(user_id=1)
print(res.id, res.status)      # مثلا PENDING, STARTED, SUCCESS
print(res.get(timeout=10))     # در صورت نیاز به نتیجه

اگر وضعیت همیشه PENDING ماند، معمولا worker متوقف است یا Celery نتوانسته وظیفه را کشف کند (autodiscover). اطمینان بگیرید اپلیکیشن شما در INSTALLED_APPS است و فایل tasks.py در آن تعریف شده است.

زمان بندی وظایف با Celery Beat

برای اجرای دوره ای وظایف، beat را با یکی از دو روش زیر پیکربندی کنید.

روش 1: استفاده از CELERY_BEAT_SCHEDULE در تنظیمات

در settings.py یک برنامه زمان بندی اضافه کنید:

# settings.py
from celery.schedules import crontab

CELERY_BEAT_SCHEDULE = {
    'cleanup-expired-sessions-every-night': {
        'task': 'myapp.tasks.cleanup_expired_sessions',
        'schedule': crontab(hour=3, minute=0),  # هر روز ساعت 03:00
    },
}

و وظیفه دوره ای را بنویسید:

# myapp/tasks.py
from celery import shared_task
from django.contrib.sessions.models import Session
from django.utils import timezone

@shared_task
def cleanup_expired_sessions():
    return Session.objects.filter(expire_date__lt=timezone.now()).delete()

حالا beat را اجرا کنید:

celery -A config beat -l info

روش 2: زمان بندی دیتابیسی با django-celery-beat

اگر می خواهید زمان بندی را از طریق پنل مدیر تغییر دهید:

  1. django-celery-beat را نصب و در INSTALLED_APPS فعال کنید.
  2. مهاجرت دیتابیس را اجرا کنید.
  3. Beat را با زمان بند دیتابیسی اجرا کنید.
python manage.py migrate
celery -A config beat -l info --scheduler django_celery_beat.schedulers:DatabaseScheduler

از طریق پنل ادمین می توانید Periodic Task تعریف یا ویرایش کنید.

کار با صف ها و اولویت ها

برای جداسازی بار کارها (مثلا ایمیل و گزارش)، صف های مختلف بسازید و وظایف را به صف مرتبط مسیریابی کنید.

# settings.py
from kombu import Exchange, Queue

CELERY_TASK_QUEUES = (
    Queue('default', Exchange('default'), routing_key='default'),
    Queue('emails', Exchange('emails'), routing_key='emails'),
)

CELERY_TASK_DEFAULT_QUEUE = 'default'
CELERY_TASK_DEFAULT_ROUTING_KEY = 'default'

CELERY_TASK_ROUTES = {
    'myapp.tasks.send_welcome_email': {'queue': 'emails', 'routing_key': 'emails'},
}

worker های اختصاصی برای هر صف:

celery -A config worker -l info -Q emails -n worker.emails@%h
celery -A config worker -l info -Q default -n worker.default@%h

یک الگوی مهم: اجرا بعد از کامیت دیتابیس

اگر وظیفه شما به رکوردی وابسته است که همین الان ذخیره شده، ممکن است قبل از کامیت تراکنش اجرا شود و به داده دسترسی نداشته باشد. از on_commit استفاده کنید:

from django.db import transaction
from myapp.tasks import send_welcome_email

# بعد از ساخت کاربر
transaction.on_commit(lambda: send_welcome_email.delay(user.id))

خطاهای رایج و رفع آنها

  • اتصال به broker برقرار نیست: آدرس Redis یا RabbitMQ را بررسی کنید و سرویس را اجرا کنید. مقدار CELERY_BROKER_URL باید درست باشد.
  • وظیفه کشف نمی شود: نام اپ در INSTALLED_APPS، وجود فایل tasks.py و app.autodiscover_tasks را بررسی کنید.
  • اجرای دو باره یک کار: وظایف را idempotent طراحی کنید. در صورت نیاز، قبل از اجرای کار، قفل سبک ایجاد کنید.
# قفل ساده با کش برای جلوگیری از اجرای موازی
from django.core.cache import cache
from celery import shared_task

@shared_task
def generate_report(report_id):
    lock_key = f'report-lock-{report_id}'
    if not cache.add(lock_key, '1', timeout=600):
        return 'locked'
    try:
        # تولید گزارش
        return 'ok'
    finally:
        cache.delete(lock_key)
  • وظایف خیلی طولانی: از time_limit و soft_time_limit استفاده کنید و کار را تکه تکه کنید. منابع خارجی را در soft timeout آزاد کنید.
  • مسائل سازگاری تراکنش: برای عملیات وابسته به دیتابیس از on_commit استفاده کنید (بخش قبل).

بررسی نتیجه و مانیتورینگ

بدون ابزار اضافی، می توانید با AsyncResult وضعیت را بخوانید. برای مانیتورینگ زنده و نمودارها، ابزارهایی مانند Flower وجود دارند که می توانید در صورت نیاز نصب کنید. در هر صورت، لاگ های worker و beat بهترین نقطه شروع برای عیب یابی هستند.

محدودیت ها و نکات معماری

  • Celery تضمین تحویل حداقل یکبار می دهد. بنابراین وظیفه باید در برابر اجرای تکراری مقاوم باشد.
  • به محتوای سریال شده و سازگاری نسخه ها توجه کنید (json پایدارتر است).
  • اگر کارها بین سرویس های مستقل پخش شده اند و به هماهنگی تراکنش نیاز دارید، الگوهای سطح بالاتر مانند Saga مطرح می شوند. برای آشنایی با این موضوع ببینید: Saga Pattern در میکروسرویس.

منابع تکمیلی و ادامه مسیر

اگر به زمان بندی ساده در سطح سیستم یا اسکریپت های مدیریت فکر می کنید، مقاله Command اختصاصی در Django و زمان بندی اجرا دید خوبی از کرون و رویکردهای مکمل می دهد.

برای تمرین عملی در یک پروژه واقعی فروشگاه، می توانید از آموزش جنگو رایگان (فروشگاه اینترنتی) + سورس استفاده کنید و ارسال ایمیل ها، تسویه سفارش یا گزارش ها را به Celery بسپارید.

گام بعدی چیست؟

روی سیستم توسعه خود Redis را اجرا کنید، یک وظیفه سبک مثل ارسال ایمیل یا بازسازی thumbnail را به Celery منتقل کنید، سپس یک کار دوره ای با beat بسازید. بعد از پایدار شدن، صف های جداگانه تعریف کنید و محدودیت های زمان اجرا و سیاست های تکرار را دقیق تنظیم کنید.