استفاده از Celery برای Task Queue در پایتون
در دنیای توسعه وب مدرن، بسیاری از عملیات ممکن است زمانبر باشند و نباید به طور مستقیم در پاسخ به درخواستهای کاربر انجام شوند. این عملیات میتوانند شامل پردازش تصاویر، ارسال ایمیل، تولید گزارشها، یا هر نوع محاسبات سنگین دیگر باشند. انجام این کارها به صورت همزمان (synchronously) میتواند منجر به کندی و عدم پاسخگویی برنامه شود. برای حل این مشکل، از مفهوم Task Queue استفاده میشود. Celery یک کتابخانه قدرتمند و محبوب در پایتون برای پیادهسازی Task Queue است.
Task Queue چیست؟
Task Queue یک الگوی طراحی است که به شما امکان میدهد وظایف (tasks) را به صورت غیرهمزمان (asynchronously) پردازش کنید. به جای اینکه وظایف مستقیماً در همان فرآیندی که درخواست را دریافت کرده است انجام شوند، در یک صف (queue) قرار میگیرند و توسط یک یا چند کارگر (worker) پردازش میشوند. این کار باعث میشود برنامه اصلی شما پاسخگو بماند و کاربران تجربه بهتری داشته باشند.
چرا Celery؟
Celery به دلایل متعددی انتخاب مناسبی برای پیادهسازی Task Queue در پایتون است:
- سادگی: Celery رابط کاربری ساده و قابل فهمی دارد که یادگیری و استفاده از آن را آسان میکند.
- مقیاسپذیری: Celery به راحتی میتواند برای پردازش حجم زیادی از وظایف مقیاسبندی شود.
- انعطافپذیری: Celery از انواع مختلفی از brokerها (مانند RabbitMQ و Redis) و backendها (مانند Redis و databaseها) پشتیبانی میکند.
- ویژگیهای پیشرفته: Celery ویژگیهای پیشرفتهای مانند زمانبندی وظایف، retryها، و زنجیرهسازی وظایف را ارائه میدهد.
- جامعه فعال: Celery دارای یک جامعه فعال از توسعهدهندگان است که به طور مداوم در حال بهبود و توسعه آن هستند.
پیشنیازها
قبل از شروع، باید پیشنیازهای زیر را نصب کنید:
- پایتون: نسخه 3.6 یا بالاتر
- RabbitMQ یا Redis: به عنوان broker
- Celery: با استفاده از pip:
pip install celery
راهاندازی Celery
برای راهاندازی Celery، مراحل زیر را دنبال کنید:
- ایجاد یک فایل Celery app: این فایل حاوی تنظیمات Celery و تعریف وظایف شما خواهد بود.
- راهاندازی Broker: یک broker (مانند RabbitMQ یا Redis) را نصب و پیکربندی کنید.
- راهاندازی Worker: یک یا چند worker را برای پردازش وظایف راهاندازی کنید.
ایجاد فایل Celery app
یک فایل به نام celery_app.py ایجاد کنید و کد زیر را در آن قرار دهید:
from celery import Celery
app = Celery('my_tasks', broker='redis://localhost:6379/0')
@app.task
def add(x, y):
return x + y
در این کد:
Celery('my_tasks', broker='redis://localhost:6379/0')یک instance از Celery ایجاد میکند.'my_tasks'نام برنامه Celery است.'redis://localhost:6379/0'آدرس broker (در اینجا Redis) است.@app.taskیک دکوراتور است که یک تابع پایتون را به عنوان یک وظیفه Celery تعریف میکند.add(x, y)یک وظیفه ساده است که دو عدد را با هم جمع میکند.
راهاندازی Broker
اگر از Redis به عنوان broker استفاده میکنید، مطمئن شوید که Redis نصب و در حال اجرا است. اگر از RabbitMQ استفاده میکنید، RabbitMQ را نصب و پیکربندی کنید.
راهاندازی Worker
برای راهاندازی worker، دستور زیر را در ترمینال اجرا کنید:
celery -A celery_app worker --loglevel=info
این دستور یک worker را با استفاده از فایل celery_app.py راهاندازی میکند و سطح log را روی info تنظیم میکند.
استفاده از وظایف Celery
برای فراخوانی یک وظیفه Celery، از متد delay() یا apply_async() استفاده کنید.
استفاده از delay()
متد delay() یک راه ساده برای فراخوانی یک وظیفه Celery به صورت غیرهمزمان است.
from celery_app import add result = add.delay(4, 4) print(result.id) # شناسه وظیفه
در این کد، وظیفه add با آرگومانهای 4 و 4 فراخوانی میشود. متد delay() یک instance از AsyncResult را برمیگرداند که میتوانید از آن برای بررسی وضعیت وظیفه و دریافت نتیجه آن استفاده کنید.
استفاده از apply_async()
متد apply_async() انعطافپذیری بیشتری نسبت به delay() ارائه میدهد. میتوانید از آن برای تنظیم پارامترهای مختلفی مانند queue، retryها، و expiration time استفاده کنید.
from celery_app import add result = add.apply_async((4, 4), queue='my_queue') print(result.id)
در این کد، وظیفه add با آرگومانهای 4 و 4 در queue به نام my_queue فراخوانی میشود.
دریافت نتیجه وظیفه
برای دریافت نتیجه یک وظیفه Celery، از متد get() استفاده کنید.
from celery_app import add result = add.delay(4, 4) print(result.get()) # نتیجه وظیفه
متد get() تا زمانی که وظیفه به پایان برسد و نتیجه در دسترس باشد، منتظر میماند. اگر وظیفه با خطا مواجه شود، متد get() یک exception را raise میکند.
زمانبندی وظایف
Celery به شما امکان میدهد وظایف را در زمانهای مشخصی زمانبندی کنید. برای این کار، از کلاس celery.schedules استفاده کنید.
from celery_app import add
from celery.schedules import crontab
@app.task
def periodic_task():
print("Periodic task running...")
app.conf.beat_schedule = {
'add-every-30-seconds': {
'task': 'celery_app.add',
'schedule': 30.0,
'args': (16, 16)
},
'periodic-task-every-minute': {
'task': 'celery_app.periodic_task',
'schedule': crontab(minute='*/1')
}
}
در این کد:
app.conf.beat_scheduleیک dictionary است که وظایف زمانبندی شده را تعریف میکند.'add-every-30-seconds'نام وظیفه زمانبندی شده است.'task': 'celery_app.add'نام وظیفهای است که باید اجرا شود.'schedule': 30.0زمانبندی وظیفه را به ثانیه مشخص میکند.'args': (16, 16)آرگومانهای وظیفه را مشخص میکند.'periodic-task-every-minute'نام وظیفه زمانبندی شده است.'schedule': crontab(minute='*/1')زمانبندی وظیفه را با استفاده از crontab مشخص میکند.
برای فعال کردن beat، دستور زیر را اجرا کنید:
celery -A celery_app beat --loglevel=info
Retryها
Celery به شما امکان میدهد وظایفی را که با خطا مواجه میشوند، به طور خودکار retry کنید.
from celery_app import add
from celery import retry
@app.task(bind=True, max_retries=3)
def add_with_retry(self):
try:
result = add(10, 10)
return result
except Exception as e:
print(f"Error: {e}")
retry(exc=e, countdown=60)
در این کد:
bind=Trueبه وظیفه اجازه میدهد تا به instance خود دسترسی داشته باشد.max_retries=3حداکثر تعداد retryها را مشخص میکند.retry(exc=e, countdown=60)وظیفه را بعد از 60 ثانیه retry میکند.
نتیجهگیری
Celery یک کتابخانه قدرتمند و انعطافپذیر برای پیادهسازی Task Queue در پایتون است. با استفاده از Celery، میتوانید وظایف زمانبر را به صورت غیرهمزمان پردازش کنید و پاسخگویی برنامه خود را بهبود بخشید. این مقاله یک معرفی جامع به Celery ارائه داد و نحوه راهاندازی، استفاده از وظایف، زمانبندی وظایف، و retryها را توضیح داد. با استفاده از این دانش، میتوانید Celery را در پروژههای پایتون خود به کار ببرید و از مزایای آن بهرهمند شوید.
بدون دیدگاه