آموزش Celery — صف کار (Task Queue) با پایتون

Celery در یک نگاه — خلاصه کلیدی

• هدف اصلی: اجرای کارهای زمان‌بر یا پرمصرف در پس‌زمینه، خارج از جریان اصلی درخواست‌ها، برای افزایش سرعت و پاسخگویی برنامه.
• چرا Celery؟ حل مشکل Block کردن رابط کاربری، جلوگیری از Time Out شدن درخواست‌ها، پردازش داده‌های بزرگ و زمان‌بندی دقیق وظایف.
• اجزای کلیدی: Client (برنامه‌ای که وظایف را ارسال می‌کند)، Broker (صف پیام، مانند Redis یا RabbitMQ)، Worker (پردازش‌گر وظایف) و Results Backend (برای ذخیره نتایج).
• کاربردهای رایج: ارسال ایمیل، تولید گزارشات پیچیده، پردازش تصاویر/ویدیوها، همگام‌سازی داده‌ها، عملیات مرتبط با APIهای خارجی.
• نصب و راه‌اندازی: آسان با `pip install celery` و انتخاب یک Broker (مثلاً Redis).
• مفاهیم پیشرفته: زمان‌بندی وظایف با Celery Beat، مدیریت خطا، Tranactional tasks و مانیتورینگ با Flower.

Celery ابزاری قدرتمند برای ساخت برنامه‌های مقیاس‌پذیر و واکنش‌گرا در پایتون است.

فهرست مطالب

در دنیای پر سرعت امروز، برنامه‌های وب و نرم‌افزارها باید واکنش‌گرا، سریع و قابل اعتماد باشند. اما چه اتفاقی می‌افتد وقتی برنامه شما نیاز به انجام کارهایی دارد که زمان‌بر هستند؟ مثلاً ارسال هزاران ایمیل، تولید گزارشات پیچیده، پردازش تصاویر سنگین یا همگام‌سازی داده‌ها با یک API کند؟ اجرای این وظایف به صورت همزمان با درخواست کاربر (Synchronous)، باعث کندی برنامه، تجربه کاربری ناخوشایند و حتی Time Out شدن درخواست می‌شود.

اینجاست که مفهوم «صف کار» یا Task Queue وارد میدان می‌شود. Celery، یک فریمورک قدرتمند و انعطاف‌پذیر در پایتون، دقیقاً برای حل همین مشکل طراحی شده است. Celery به شما این امکان را می‌دهد که وظایف زمان‌بر را به صورت ناهمزمان (Asynchronous) و در پس‌زمینه اجرا کنید، بدون اینکه تجربه کاربری تحت تأثیر قرار گیرد.

Celery چیست و چرا به آن نیاز داریم؟

Celery یک سیستم صف کار توزیع شده (Distributed Task Queue) است که به زبان پایتون نوشته شده و به شما کمک می‌کند وظایف را به صورت ناهمزمان اجرا کنید. تصور کنید یک کاربر دکمه‌ای را در وب‌سایت شما کلیک می‌کند و این عمل باعث شروع یک فرآیند طولانی می‌شود، مثلاً تولید یک فایل PDF با ۱۰۰ صفحه. اگر این فرآیند به صورت مستقیم توسط سرور وب شما انجام شود، کاربر باید تا پایان آن منتظر بماند. اما با Celery، شما می‌توانید وظیفه تولید PDF را به Celery واگذار کنید و بلافاصله به کاربر اعلام کنید که درخواست او در حال پردازش است. سپس Celery این وظیفه را در پس‌زمینه و بدون ایجاد اختلال در عملکرد اصلی برنامه، انجام می‌دهد.

موارد استفاده رایج از Celery:

• ارسال ایمیل: ارسال ایمیل‌های تأیید ثبت‌نام، خبرنامه‌ها یا اعلان‌ها به صورت دسته‌ای.
• تولید گزارشات: ساخت گزارشات پیچیده مالی، آماری یا تحلیلی که نیاز به پردازش حجم زیادی از داده‌ها دارند.
• پردازش فایل‌ها: فشرده‌سازی، تغییر اندازه یا واترمارک کردن تصاویر و ویدیوها.
• همگام‌سازی داده‌ها: به‌روزرسانی اطلاعات از APIهای خارجی یا دیتابیس‌های دیگر.
• کارهای زمان‌بندی شده: اجرای وظایف به صورت دوره‌ای (مثلاً هر شب بک‌آپ گرفتن از دیتابیس) با Celery Beat.

Celery چطور کار می‌کند؟ اجزای اصلی

Celery یک معماری ساده اما قدرتمند دارد که بر پایه چهار جزء اصلی بنا شده است:

• Client (مشتری): این قسمت از برنامه شماست (مثلاً یک برنامه وب با جنگو یا فلاسک) که یک وظیفه را تعریف کرده و آن را برای اجرا به Celery می‌فرستد.
• Broker (واسطه/کارگزار): قلب سیستم Celery. Broker یک صف پیام (Message Queue) است که وظایف ارسال شده توسط Client را دریافت کرده و آن‌ها را در یک صف نگه می‌دارد تا Workerها آن‌ها را پردازش کنند. محبوب‌ترین Brokerها Redis و RabbitMQ هستند.
• Worker (پردازش‌گر): Workerها فرآیندهای مجزایی هستند که به صورت مداوم به Broker گوش می‌دهند. به محض اینکه یک وظیفه جدید در صف قرار می‌گیرد، Worker آن را برمی‌دارد، اجرا می‌کند و نتیجه را (در صورت نیاز) به Results Backend ارسال می‌کند.
• Results Backend (پایگاه نتایج): اختیاری است، اما اگر بخواهید نتیجه یک وظیفه Celery را پس از اجرا پیگیری و دریافت کنید، به آن نیاز دارید. می‌توان از دیتابیس، Redis، یا S3 به عنوان Results Backend استفاده کرد.

جریان کار به این صورت است: Client یک وظیفه را به Broker ارسال می‌کند. Broker آن را در صف نگه می‌دارد. Workerها به صورت موازی به Broker متصل هستند و وظایف را از صف برمی‌دارند و اجرا می‌کنند. پس از اتمام کار، Worker نتیجه را (اگر پیکربندی شده باشد) به Results Backend می‌فرستد. Client می‌تواند در زمان‌های بعدی از Results Backend برای گرفتن نتیجه وظیفه استفاده کند.

نصب و راه‌اندازی Celery

نصب Celery نسبتاً ساده است. قبل از هر چیز، مطمئن شوید که پایتون و `pip` روی سیستم شما نصب هستند. همچنین، به یک Broker نیاز دارید. در این آموزش از Redis استفاده می‌کنیم که نصب و استفاده از آن بسیار آسان است.

گام ۱: نصب Redis (به عنوان Broker)

Redis یک دیتابیس درون حافظه‌ای سریع است که به عنوان Broker و Results Backend در Celery عالی عمل می‌کند. برای نصب Redis در لینوکس (مثلاً Ubuntu):

sudo apt update
sudo apt install redis-server
sudo systemctl enable redis-server
sudo systemctl start redis-server

برای ویندوز یا macOS، می‌توانید از Docker استفاده کنید یا نصاب‌های مربوطه را دانلود کنید.

گام ۲: ایجاد Virtual Environment

همیشه توصیه می‌شود که برای پروژه‌های پایتون از Virtual Environment استفاده کنید:

mkdir celery_project
cd celery_project
python3 -m venv venv
source venv/bin/activate  # برای لینوکس/macOS
# venvScriptsactivate   # برای ویندوز

گام ۳: نصب Celery و کتابخانه‌های مورد نیاز

پس از فعال‌سازی محیط مجازی، Celery و کتابخانه پایتون برای Redis را نصب کنید:

pip install celery redis

اولین صف کار با Celery: گام به گام

حالا که Celery نصب شده، بیایید اولین وظیفه خود را تعریف و اجرا کنیم.

۱. ساخت فایل `celery_app.py`

در این فایل، نمونه Celery (Application) را ایجاد و Broker و Results Backend را پیکربندی می‌کنیم. این فایل در واقع نقطه ورود Celery به برنامه شماست.

# celery_app.py
from celery import Celery

app = Celery(
    'my_celery_app',
    broker='redis://localhost:6379/0',  # آدرس Broker (Redis)
    backend='redis://localhost:6379/1', # آدرس Results Backend (Redis)
    include=['celery_project.tasks'] # ماژول‌هایی که وظایف در آن‌ها تعریف شده‌اند
)

# تنظیمات اختیاری
app.conf.update(
    task_serializer='json',
    accept_content=['json'],
    result_serializer='json',
    timezone='Asia/Tehran', # یا منطقه زمانی شما
    enable_utc=True,
    broker_connection_retry_on_startup=True
)

۲. ساخت فایل `tasks.py`

وظایف (Tasks) توابع پایتونی هستند که با دکوراتور `@app.task` مشخص می‌شوند. آن‌ها را در فایلی به نام `tasks.py` در همان دایرکتوری `celery_project` ایجاد کنید.

# celery_project/tasks.py
import time
from celery_project.celery_app import app # ایمپورت کردن نمونه Celery

@app.task
def add(x, y):
    time.sleep(5) # شبیه‌سازی یک کار زمان‌بر
    print(f"Executing add task: {x} + {y}")
    return x + y

@app.task
def send_email(recipient, subject, body):
    print(f"Sending email to {recipient} with subject '{subject}'...")
    time.sleep(3) # شبیه‌سازی ارسال ایمیل
    print(f"Email sent to {recipient}!")
    return True

۳. اجرای Worker

Workerها فرآیندهایی هستند که وظایف را از صف برمی‌دارند و اجرا می‌کنند. برای شروع یک Worker، ترمینال را باز کنید (مطمئن شوید که در محیط مجازی و دایرکتوری `celery_project` هستید) و دستور زیر را اجرا کنید:

celery -A celery_project.celery_app worker --loglevel=info

• `celery`: دستور اصلی Celery.
• `-A celery_project.celery_app`: مشخص می‌کند که نمونه Celery در ماژول `celery_app.py` داخل پکیج `celery_project` قرار دارد.
• `worker`: دستور شروع Worker.
• `–loglevel=info`: سطح لاگ‌گیری را روی `info` تنظیم می‌کند تا جزئیات بیشتری را در خروجی ببینید.

پس از اجرای این دستور، Celery Worker شروع به کار می‌کند و به Redis گوش می‌دهد.

۴. فراخوانی وظایف

حالا یک ترمینال جدید باز کنید (همچنان در محیط مجازی فعال) و یک فایل پایتون به نام `caller.py` در همان دایرکتوری بسازید:

# caller.py
from celery_project.tasks import add, send_email
import time

print("Calling Celery tasks...")

# فراخوانی وظیفه 'add' به صورت ناهمزمان
result_add = add.delay(4, 6)
print(f"Task add called. Task ID: {result_add.id}")

# فراخوانی وظیفه 'send_email' به صورت ناهمزمان
result_email = send_email.delay("user@example.com", "Welcome", "Hello there!")
print(f"Task send_email called. Task ID: {result_email.id}")

print("Tasks sent to Celery. Waiting for results...")

# منتظر ماندن برای نتیجه وظیفه add (اختیاری، اما برای دیدن نتیجه لازم است)
# این خط برنامه را تا زمانی که نتیجه آماده شود، block می کند.
# در یک برنامه واقعی، شما معمولاً به صورت غیر Block کننده نتایج را چک می‌کنید.
try:
    final_result = result_add.get(timeout=10) # 10 ثانیه منتظر بمان
    print(f"Result of add task: {final_result}")
except Exception as e:
    print(f"Error getting result: {e}")

print("Program finished.")

این فایل را با `python caller.py` اجرا کنید. شما بلافاصله خروجی “Tasks sent to Celery…” را خواهید دید و Worker در ترمینال دیگر شروع به پردازش وظایف می‌کند. پس از چند ثانیه، نتایج در هر دو ترمینال نمایش داده می‌شود.

مفهوم Broker در Celery و انتخاب آن

Broker واسطه‌ای است که وظایف را از Client دریافت کرده و به Workerها تحویل می‌دهد. انتخاب Broker مناسب برای پروژه شما حیاتی است. Celery از چندین Broker مختلف پشتیبانی می‌کند، اما دو مورد محبوب‌تر Redis و RabbitMQ هستند.

۱. Redis

Redis یک دیتابیس Key-Value درون حافظه‌ای است که برای صف‌بندی پیام‌ها بسیار سریع و سبک است. نصب و راه‌اندازی آن آسان است و برای پروژه‌های کوچک تا متوسط بسیار مناسب است. در مثال‌های بالا از Redis استفاده کردیم.

broker='redis://localhost:6379/0'

۲. RabbitMQ

RabbitMQ یک سیستم پیام‌رسان قدرتمند (Message Broker) است که پروتکل AMQP را پیاده‌سازی می‌کند. RabbitMQ قابلیت‌های پیشرفته‌تری مانند صف‌های پایدار (Durable Queues)، مسیریابی پیچیده‌تر پیام‌ها و امکانات مدیریتی بهتر را ارائه می‌دهد. برای پروژه‌های بزرگ و سازمانی که نیاز به پایداری بالا و ویژگی‌های پیشرفته صف پیام دارند، RabbitMQ انتخاب بهتری است.

broker='amqp://guest:guest@localhost:5672//'

Results Backend چیست؟

Results Backend جایی است که Celery نتایج (و وضعیت) وظایف اجرا شده را ذخیره می‌کند. همانطور که اشاره شد، این یک جزء اختیاری است. اگر شما فقط می‌خواهید وظایف در پس‌زمینه اجرا شوند و نیازی به پیگیری نتیجه نهایی آن‌ها ندارید، می‌توانید آن را حذف کنید. اما برای بسیاری از برنامه‌های وب، دانستن اینکه آیا یک وظیفه با موفقیت انجام شده یا شکست خورده و نتیجه آن چه بوده، اهمیت دارد.

می‌توانید از Redis، دیتابیس‌های رابطه‌ای (مانند PostgreSQL یا MySQL) یا حتی Apache Cassandra به عنوان Results Backend استفاده کنید. انتخاب آن بستگی به نیازهای پروژه شما دارد. Redis برای این منظور نیز به دلیل سرعت بالا و سهولت استفاده، گزینه‌ی محبوبی است.

backend='redis://localhost:6379/1' # استفاده از Redis برای ذخیره نتایج

مدیریت وظایف پیشرفته (Celery Beat, Chains, Groups)

Celery فقط برای اجرای وظایف ناهمزمان نیست؛ این فریمورک قابلیت‌های پیشرفته‌ای برای مدیریت پیچیده‌تر وظایف ارائه می‌دهد.

۱. Celery Beat: زمان‌بندی وظایف

Celery Beat یک زمان‌بندی‌کننده (Scheduler) است که وظایف را در زمان‌های مشخص یا با فواصل زمانی منظم اجرا می‌کند. این ابزار دقیقاً مانند Cron Job عمل می‌کند، اما به صورت توزیع شده و با قابلیت‌های Celery. برای استفاده از Celery Beat، یک فایل `celeryconfig.py` یا تنظیمات را مستقیماً در `celery_app.py` اضافه کنید:

# celery_app.py (ادامه تنظیمات)
from celery.schedules import crontab

app.conf.beat_schedule = {
    'add-every-10-seconds': {
        'task': 'celery_project.tasks.add',
        'schedule': 10.0, # هر ۱۰ ثانیه
        'args': (16, 16)
    },
    'send-email-every-minute': {
        'task': 'celery_project.tasks.send_email',
        'schedule': crontab(minute='*/1'), # هر دقیقه
        'args': ('admin@example.com', 'Daily Report', 'Here is your daily report.')
    },
}

برای اجرای Celery Beat، در یک ترمینال جداگانه دستور زیر را اجرا کنید:

celery -A celery_project.celery_app beat --loglevel=info

۲. تأخیر در اجرا و زمان‌بندی دقیق (ETA & Countdown)

شما می‌توانید وظایف را با تأخیر اجرا کنید یا زمان دقیق اجرای آن‌ها را مشخص کنید:

from datetime import datetime, timedelta
from celery_project.tasks import add

# اجرا با تأخیر 10 ثانیه
add.apply_async((10, 5), countdown=10)

# اجرا در یک زمان مشخص در آینده
future_time = datetime.now() + timedelta(minutes=1)
add.apply_async((20, 10), eta=future_time)

۳. ارسال زنجیره‌ای (Chains) و گروه‌ها (Groups)

Celery امکان ترکیب و توالی وظایف را به شما می‌دهد:

• Chain (زنجیره): برای اجرای وظایف به صورت متوالی، جایی که خروجی یک وظیفه، ورودی وظیفه بعدی است.
• Group (گروه): برای اجرای مجموعه‌ای از وظایف به صورت موازی.

# celery_project/tasks.py
@app.task
def multiply(x, y):
    return x * y

@app.task
def subtract(x, y):
    return x - y

# caller.py (مثال برای chain)
from celery import chain
from celery_project.tasks import add, multiply, subtract

# (4 + 6) * 2 - 3
workflow = chain(add.s(4, 6), multiply.s(2), subtract.s(3))
result = workflow.apply_async()
print(f"Chained task ID: {result.id}")
print(f"Chained result: {result.get()}") # خروجی: 17

# (مثال برای group)
from celery import group
from celery_project.tasks import add

parallel_tasks = group(add.s(1, 2), add.s(3, 4), add.s(5, 6))
group_result = parallel_tasks.apply_async()
print(f"Grouped tasks ID: {group_result.id}")
print(f"Grouped results: {group_result.get()}") # خروجی: [3, 7, 11] (ترتیب ممکن است متفاوت باشد)

نظارت و مانیتورینگ Celery با Flower

هنگامی که سیستم شما در حال اجراست، نیاز دارید که وضعیت وظایف، Workerها و صف‌ها را مشاهده و مدیریت کنید. Flower یک ابزار مانیتورینگ بلادرنگ (Real-time) و رابط کاربری وب برای Celery است.

نصب Flower:

pip install flower

اجرای Flower:

Flower را می‌توانید در یک ترمینال جداگانه اجرا کنید:

celery -A celery_project.celery_app flower --broker=redis://localhost:6379/0

سپس می‌توانید به آدرس `http://localhost:5555` در مرورگر خود بروید تا داشبورد Flower را مشاهده کنید. این داشبورد اطلاعات ارزشمندی در مورد وظایف فعال، موفق، ناموفق، Workerها و میزان مصرف منابع به شما می‌دهد.

بهینه‌سازی و بهترین شیوه‌ها در Celery

برای اینکه از Celery بهترین استفاده را ببرید، رعایت چند نکته ضروری است:

• وظایف کوچک و مستقل: وظایف خود را تا حد امکان کوچک و مستقل (Idempotent) نگه دارید. یعنی هر وظیفه باید بتواند بدون تأثیر جانبی تکرار شود.
• مدیریت خطا (Error Handling): همیشه برای وظایف خود مکانیسم مدیریت خطا (try-except) و قابلیت تلاش مجدد (Retry) را در نظر بگیرید. Celery امکاناتی برای `max_retries` و `retry_backoff` ارائه می‌دهد.
• Task Acknowledge: Celery به صورت پیش‌فرض پس از اجرای موفق وظیفه، آن را از Broker حذف می‌کند (Acknowledge). مطمئن شوید که تنظیمات `task_acks_late` را در صورت نیاز به درستی انجام داده‌اید تا اگر Worker قبل از اتمام کار از کار افتاد، وظیفه دوباره به صف بازگردانده شود.
• سریالایزرها (Serializers): از سریالایزر مناسب استفاده کنید. JSON برای اکثر موارد مناسب است، اما برای داده‌های پیچیده‌تر می‌توانید از Pickle یا YAML استفاده کنید (با احتیاطات امنیتی لازم).
• تعداد Workerها و Concurrency: تعداد Workerها و فرآیندهای همزمان (Concurrency) را با توجه به منابع سرور و ماهیت وظایف خود تنظیم کنید. وظایف I/O-bound (مثل درخواست‌های شبکه) می‌توانند از Concurrency بالا بهره ببرند، در حالی که وظایف CPU-bound باید Concurrency کمتری داشته باشند.
• استفاده از Beat در حالت پایدار: Celery Beat در محیط پروداکشن معمولاً با یک Worker خاص یا در یک فرآیند جداگانه اجرا می‌شود تا از مشکلات احتمالی جلوگیری شود. برای افزایش پایداری، می‌توانید از `celery-beat-scheduler` یا ابزارهای مشابه استفاده کنید.
• جلوگیری از Loopهای بی‌پایان: از فراخوانی وظایف Celery در داخل خود وظایف Celery بدون کنترل دقیق خودداری کنید تا از ایجاد Loopهای بی‌پایان جلوگیری شود.

عیب‌یابی سریع Celery

در اینجا برخی از مشکلات رایج و راه‌حل‌های سریع آن‌ها آورده شده است:

• Worker شروع نمی‌شود یا وظایف را دریافت نمی‌کند:
  • بررسی Broker: مطمئن شوید که Redis یا RabbitMQ در حال اجراست و از آدرس صحیح در `broker` تنظیمات Celery استفاده می‌کنید.
  • نام برنامه Celery: مطمئن شوید که `-A celery_project.celery_app` به درستی به فایل `celery_app.py` شما اشاره دارد.
  • لاگ‌ها: `worker –loglevel=DEBUG` را اجرا کنید تا جزئیات بیشتری از مشکل را مشاهده کنید.
  • فایروال: بررسی کنید که پورت Broker (مثلاً 6379 برای Redis) توسط فایروال مسدود نشده باشد.
• وظایف اجرا می‌شوند اما نتیجه‌ای برگردانده نمی‌شود:
  • Results Backend: مطمئن شوید که `backend` در `celery_app.py` به درستی پیکربندی شده و سرور مربوطه (مثلاً Redis) در حال اجراست.
  • Timeout در `get()`: ممکن است زمان `timeout` برای `result.get()` خیلی کم باشد. آن را افزایش دهید یا از `ready()` برای بررسی وضعیت استفاده کنید.
• خطای Serialization (مثلاً `kombu.exceptions.EncodeError`):
  • نوع سریالایزر: مطمئن شوید که `task_serializer` و `result_serializer` در تنظیمات Celery به درستی تنظیم شده‌اند (مثلاً ‘json’).
  • داده‌های قابل سریالایز: مطمئن شوید که پارامترها و نتایج وظایف شما از انواع داده‌ای هستند که توسط سریالایزر انتخابی قابل پردازش باشند (مثلاً شیءهای پیچیده پایتون با JSON خوب کار نمی‌کنند).
• وظایف زمان‌بندی شده (Beat) اجرا نمی‌شوند:
  • Beat در حال اجراست؟ مطمئن شوید که Celery Beat در یک ترمینال جداگانه با دستور صحیح در حال اجراست.
  • تنظیمات Beat: تنظیمات `app.conf.beat_schedule` را در `celery_app.py` به دقت بررسی کنید.
  • زمان‌بندی: از درستی زمان‌بندی (مثلاً `crontab`) اطمینان حاصل کنید.

پرسش‌های متداول (FAQ)

جمع‌بندی

Celery یک ابزار بی‌نظیر برای مدیریت وظایف ناهمزمان در برنامه‌های پایتون است. با استفاده از آن، می‌توانید عملکرد برنامه‌های خود را به شکل چشمگیری افزایش دهید، تجربه کاربری را بهبود ببخشید و سیستم‌های مقیاس‌پذیر و پایدارتری بسازید. از نصب اولیه و راه‌اندازی گرفته تا مفاهیم پیشرفته مانند زمان‌بندی وظایف و مانیتورینگ، Celery مجموعه‌ای کامل از ابزارها را در اختیار شما قرار می‌دهد.

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