آموزش جامع Celery؛ مدیریت صف وظایف در پایتون

در بحث توسعه نرم افزار٬ سرعت پاسخگویی به درخواست ها صرفا یک ویژگی فنی نیست که بتوان از آن چشم پوشی کرد یا نسبت به آن بی تفاوت بود! بلکه بخش مهمی از تجربه کاربری به شمار می رود.

کاربران انتظار دارند عملیات هایی مانند ثبت سفارش٬ ارسال فرم٬ بارگذاری فایل یا درخواست گزارش در سریع ترین زمان ممکن انجام شود و زمان زیادی را صرف انتظار برای به نتیجه رسیدن درخواست خود نداشته باشند.

با این حال پشت بسیاری از درخواست هایی که به ظاهر ساده هستند پردازش هایی وجود دارند که ممکن است چند ثانیه یا حتی چند دقیقه زمان ببرند.

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

راهکار مناسب این است که درخواست های کاربران در سریع ترین زمان ممکن انجام شود و کارهای زمان بر و یا غیرضروری و کارهایی که امکان به تعویق انداختن آن ها وجود دارد به نوبت در پس زمینه اجرا شوند تا درخواست های اصلی به نتیجه برسد و کاربر نیاز نباشد برای پاسخ به درخواست خود مدت زمان طولانی منتظر باشد.

در این شرایط برنامه می تواند بدون اینکه کاربر را منتظر نگه دارد وظایف سنگین را به پردازش های دیگری بسپارد و نتیجه درخواست اصلی را سریع تر نمایش دهد.

اگر بخواهیم در پروژه های پایتونی چنین ساختاری را پیدا کنیم ابزارهای مختلف وجود دارد اما مهم ترین و کاربردی ترین ابزاری که بین توسعه دهندگان پایتون شناخته شده است در واقع Celery می باشد.

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

Celery چیست؟

اگر بخواهیم به تعریف ساده تر Celery بپردازیم باید بگوییم این ابزار در واقع یک Distributed Task Queue یا همان صف وظایف توزیع شده برای پایتون می باشد که به برنامه ها اجازه می دهد عملیات های سنگین٬ زمان بر و یا قابل تعویق انداختن را خارج از جریان اصلی برنامه انجام دهند.

در حالت عادی٬ وقتی که یک تابع در پروژه فراخوانی می شود همان تابع در همان پردازش اجرا می شود و برنامه منتظر می ماند این درخواست به پایان برسد. این رفتار برای عملیاتی های کوتاه و سبک که سریع به اتمام می رسد هیچ مشکلی ایجاد نمی کند و شیوه کار هم دقیقا به همین شکل می باشد اما اگر اجرای تابع قرار باشد چند ثانیه یا چند دقیقه طول بکشد قاعده بازی تغییر می کند! اینجا دیگر نمی توان انتظار داشت کاربر منتظر به پایان رسیدن عملیات بماند و علاوه بر اینکه تجربه کاربری بدی را بوجود می آورد باعث می شود کاربر عملیات را نیمه کاره رها کند.

دقیقا Celery برای حل همین مشکل توسعه داده شد تا دیگر کاربران نیاز نباشد در پروژه های سنگین برای عملیات های زمان بر منتظر بمانند!

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

پس از این مرحله یک یا چند پردازش مستقل به نام Worker پیام های موجود در صف را دریافت می کند و Task مربوط به هر پیام را به نوبت اجرا می کند و به این شکل برنامه اصلی دیگر منتظر نمی ماند تا عملیات ها به پایان برسد بلکه به اجرا خود ادامه می شود و پاسخ مناسب را سریع به کاربر برمیگرداند.

در ساده‌ ترین حالت، جریان اجرای یک Task در Celery به شکل زیر است:

کاربر

↓

برنامه وب یا Producer

↓

Message Broker

↓

Celery Worker

↓

اجرای Task

↓

Result Backend، در صورت نیاز

اگر بخواهیم با یک مثال ساده این فرآیند را توضیح دهیم باید بگوییم فرض کنید کاربری در یک سایت درخواست تولید گزارش را ثبت می کند طبیعی هست که ساخت گزارش ممکن است چند دقیقه زمان نیاز داشته باشد اینجا دیگر نیاز نیست درخواست کاربر به صورت باز باقی ماند تا عملیات به پایان برسد بلکه با پیغامی مانند درخواست شما ثبت شده است را دریافت کند و این عملیات را به Celery ارسال می کند و سپس یکی از Worker ها گزارش را در پس زمینه تولید می کند و در کارتابل یا صفحه مربوط به گزارش گیری قرار می دهد.

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

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

نکته خوب ماجرا اینجاست که Worker ها حتی می توانند روی سرور های متفاوتی وجود داشته باشند و به همین دلیل Celery در واقع یک سیستم توزیع شده محسوب می شود و امکان مقیاس پذیری را فراهم می کند.

برای نمونه، ممکن است یک Worker مسئول پردازش ایمیل‌ ها باشد، Worker دیگری گزارش‌ ها را تولید کند و چند Worker دیگر وظایف محاسباتی را انجام دهند. همه این پردازش‌ ها می‌ توانند مستقل از برنامه وب و حتی مستقل از یکدیگر اجرا شوند.

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

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

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

Celery چه مشکلی را حل می‌ کند؟

اگر بخواهیم وظیفه ابزاری مانند Celery را دقیق تر بررسی کنیم باید بگوییم در واقع Celery کمک می کند هر پردازشی نیاز نباشد تا به اتمام رسیدن تمام بخش های مختلف باز بماند.

فرض کنید شما یک سایت فروشگاهی دارید که کاربران شما در آن خرید اینترنتی ثبت می کنند و پس از ثبت سفارش هم ممکن است نیاز باشد کارهایی نظیر موارد زیر انجام شود:

سفارش را در پایگاه داده ثبت کند.

ایمیل تایید سفارش ارسال کند.

پیامک اطلاع‌ رسانی بفرستد.

فاکتور PDF تولید کند.

موجودی انبار را بروزرسانی کند.

اطلاعات سفارش را برای سیستم حسابداری ارسال کند.

برای مدیر فروش یا واحد پشتیبانی اعلان ایجاد کند.

اطلاعات سفارش را در یک سرویس خارجی ثبت کند.

از میان تمام این علمیات هایی که گفته شد فقط ثبت صحیح سفارش در دیتابیس برای پاسخ اولیه ضروری هست و کارهایی مانند ارسال ایمیل٬ ساختار فاکتور یا ارتباط با سرویس های خارجی معمولا می تواند بدون مشکل در مدت زمان کوتاهی بعد درخواست انجام شود.

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

روند اجرای مستقیم ممکن است به این شکل باشد:

دریافت درخواست ثبت سفارش

↓

ثبت سفارش در پایگاه داده

↓

ارسال ایمیل

↓

ارسال پیامک

↓

ساخت فاکتور

↓

ارتباط با سیستم حسابداری

↓

ارسال پاسخ به کاربر

در این ساختار کاملا مشخص هست که هر کدام از این سرویس های جانبی با اختلال یا کندی همراه شوند کل درخواست تحت تاثیر قرار می گیرد! برای مثال فرض کنید موقتا سرویس ایمیل پاسخ ندهد ممکن است درخواست کاربر طولانی شود و یا خطای تایم اوت رخ دهد و در این شرایط با وجود اینکه به نظر می رسد درخواست ناموفق بوده ولی سفارش در دیتابیس ثبت شده است.

Celery این مشکل را با جدا کردن عملیات زمان‌ بر از مسیر اصلی درخواست حل می‌ کند. برنامه فقط کار ضروری را انجام می‌ دهد و سایر عملیات را برای اجرای پس‌ زمینه به صف می‌ فرستد.

برای مثال:

order = create_order(request.data)

send_order_email.delay(order.id)
generate_invoice.delay(order.id)
sync_with_accounting.delay(order.id)

return {
    "message": "Order created successfully",
    "order_id": order.id,
}

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

در نتیجه، جریان اجرا به شکل زیر تغییر می‌ کند:

دریافت درخواست ثبت سفارش

↓

ثبت سفارش در پایگاه داده

↓

ارسال Task ها به صف

↓

ارسال سریع پاسخ به کاربر

در پس‌ زمینه:

ارسال ایمیل٬ تولید فاکتور٬ هماهنگ‌ سازی با سیستم حسابداری

متد delay() یک درخواست برای اجرای Task ایجاد می‌ کند و آن را برای پردازش به Celery می‌ فرستد. این متد منتظر پایان اجرای Task نمی‌ ماند؛ بنابراین برنامه می‌ تواند بلافاصله اجرای بخش بعدی کد را ادامه دهد.

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

استفاده از Celery در چنین شرایطی چند مزیت مهم ایجاد می‌ کند:

پاسخ برنامه سریع‌ تر به کاربر برگردانده می‌ شود.

عملیات سنگین، پردازش اصلی برنامه وب را درگیر نمی‌ کنند.

اختلال موقت یک سرویس جانبی لزوما درخواست اصلی را ناموفق نمی‌ کند.

چند Task می‌ توانند به‌ صورت مستقل یا هم‌ زمان پردازش شوند.

با افزایش حجم کار، می‌ توان Worker های بیشتری برای پردازش صف‌ ها اجرا کرد.

مدیریت عملیات زمان‌ بر از منطق اصلی درخواست جدا می‌ شود.

برنامه در زمان افزایش تعداد کاربران، کنترل‌ پذیرتر و مقیاس‌ پذیرتر خواهد بود.

برای مثال اگر در یک سایت فروشگاهی پرترافیک که جشنواره تخفیفی هم دارد٬ سفارش های زیادی در لحظه ثبت شود٬ اجرای مستقیم ایمیل٬ فاکتور و هماهنگ سازی حسابداری برای هر درخواست قرار باشد همان لحظه انجام شود فشار زیادی به منابع سرور وارد می کند و احتمالا سرور بعد از ثبت چندین درخواست دیگر قادر به پاسخگویی به شکل سابق نخواهد بود.

با استفاده از Celery درخواست ها سریع تر ثبت می شوند و عملیات های جانبی به تدریج و براساس ظرفیت Worker ها پردازش خواهند شد.

اینطور تصور نکنید که Celery فقط محدود به فروشگاه های اینترنتی می باشد بلکه همین عملکرد در بسیاری از پروژه های دیگر هم کاربرد دارد برای مثال پردازش فایل پس از بارگذاری٬ ساخت گزارش های مدیریتی٬ ارسال اعلان به کاربران٬ تبدیل ویدیو٬ اجرای محاسبات سنگین یا دریافت داده از سرویس های خارجی!

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

اصطلاحات اصلی در Celery:

برای استفاده و کار با Celery بهتر است ابتدا با اجزای اصلی آن آشنا شویم که هر کدام از این اجزا در واقع چه نقشی را در فرآيند ارسال٬ نگهداری و اجرای Task ها دارند و در کنار یکدیگر ساختار کلی Celery را تشکیل می دهند.

Task:

Task واحد اصلی کار در Celery است. به زبان ساده، هر عملیاتی که قرار است توسط Celery اجرا شود، به‌ صورت یک Task تعریف می‌ شود.

برای تعریف Task معمولا از دکوراتور @app.task استفاده می‌ کنیم:

@app.task
def add(x, y):
    return x + y

در این مثال، تابع add یک Task محسوب می‌ شود و می‌ تواند مانند یک تابع معمولی یا از طریق صف Celery اجرا شود.

وقتی که یک تابع به عنوان تسک ثبت می شود Celery برای آن یک نام مشخص در نظر میگیرد و Worker ها از این نام استفاده می کنند تا هنگام دریافت پیام تشخیص دهند کدام تابع باید اجرا شود.

این اسامی معمولا براساس مسیر ماژول و نام تابع الهام گرفته می شود برای مثال اگر تابع add در فایل tasks.py قرار داشته باشد، نام آن ممکن است به شکل زیر باشد:

tasks.add

امکان تعیین نام دستی نیز وجود دارد:

@app.task(name="math.add")
def add(x, y):
    return x + y

ورودی‌ های Task باید همراه پیام ارسال شوند؛ بنابراین بهتر است از داده‌ های ساده و قابل‌ سریال‌ سازی مانند عدد، رشته، فهرست، دیکشنری و شناسه رکورد های پایگاه داده استفاده شود.

Task می‌تواند مقدار بازگشتی نیز داشته باشد. این مقدار در صورت کانفیگ Result Backend قابل ذخیره و بازیابی است.

Producer یا Client:

Producer بخشی از سیستم است که درخواست اجرای Task را ایجاد و آن را به صف ارسال می‌ کند. گاهی از واژه Client نیز برای اشاره به این بخش استفاده می‌ شود.

برای نمونه، دستور زیر یک Task را برای اجرا ارسال می‌ کند:

add.delay(10, 20)

در اینجا، برنامه‌ای که متد delay() را فراخوانی کرده است، نقش Producer را دارد.

Producer لزوما یک برنامه وب نیست. بخش‌ های مختلفی می‌توانند این نقش را بر عهده داشته باشند، از جمله:

برنامه Django

برنامه FastAPI

برنامه Flask

یک سرویس مستقل پایتون

یک اسکریپت خط فرمان

یک Task دیگر

یک زمان‌ بند مانند Celery Beat

وظیفه Producer اجرای مستقیم Task نیست. Producer فقط پیام مربوط به اجرای Task را ایجاد کرده و آن را برای Broker ارسال می‌ کند.

این پیام معمولاً شامل اطلاعاتی مانند موارد زیر است:

نام Task

آرگومان‌ های ورودی

شناسه Task

زمان اجرای احتمالی

نام صف مقصد

برخی تنظیمات مربوط به اجرای Task

پس از ارسال پیام، Producer می‌تواند بدون انتظار برای پایان Task به اجرای خود ادامه دهد.

Broker:

Broker واسط میان Producer و Worker است. پیام‌ های مربوط به Task ها ابتدا توسط Producer به Broker ارسال می‌شوند و تا زمان دریافت توسط Worker در آن نگهداری خواهند شد.

جریان ساده ارتباط به این شکل است:

Producer

↓

Broker

↓

Worker

Broker باعث می‌ شود Producer و Worker مجبور نباشند مستقیما با یکدیگر ارتباط داشته باشند. Producer فقط پیام را ارسال می‌ کند و Worker نیز هر زمان آماده باشد پیام را دریافت خواهد کرد.

این جداسازی اهمیت زیادی دارد زیرا ممکن است در زمان ارسال Task، Worker مشغول اجرای کار دیگری باشد. Broker پیام را نگه می‌ دارد تا Worker امکان پردازش آن را پیدا کند.

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

Brokerهای رایج قابل‌ استفاده با Celery عبارت‌اند از:

Redis

RabbitMQ

Amazon SQS

Google Cloud Pub/Sub

برخی سرویس‌ ها و سیستم‌ های پیام‌ رسان دیگر

برای شروع و یادگیری، Redis معمولا گزینه ساده‌ تری است؛ زیرا نصب، پیکربندی و استفاده از آن پیچیدگی کمتری دارد.

RabbitMQ نیز یک Message Broker تخصصی و قدرتمند است که امکانات پیشرفته‌ تری برای صف‌ بندی، مسیریابی و مدیریت پیام‌ ها ارائه می‌ دهد.

Redis و RabbitMQ از گزینه‌ های اصلی و رایج برای راه‌ اندازی Celery محسوب می‌ شوند.

Worker:

Worker پردازشی است که وظیفه دریافت و اجرای Task ها را بر عهده دارد. Worker به Broker متصل می‌ شود، پیام‌ های مربوط به Task ها را دریافت می‌ کند و تابع مناسب را با آرگومان‌ های ارسال‌ شده اجرا می‌ کند.

برای اجرای Worker می‌ توان از دستور زیر استفاده کرد:

celery -A tasks worker --loglevel=INFO

پس از اجرای این دستور، Worker فعال می‌ ماند و منتظر دریافت Task های جدید می‌ شود.

هنگامی که یک پیام از Broker دریافت می‌شود، Worker مراحل کلی زیر را انجام می‌ دهد:

نام Task را از پیام می‌ خواند.

Task ثبت‌ شده با آن نام را پیدا می‌ کند.

آرگومان‌ های ارسال‌ شده را دریافت می‌ کند.

تابع مربوط به Task را اجرا می‌ کند.

وضعیت اجرا را ثبت می‌ کند.

در صورت نیاز، نتیجه را در Result Backend ذخیره می‌ کند.

Worker از برنامه اصلی مستقل است. برای مثال، برنامه Django می‌تواند در یک پردازش اجرا شود و Worker های Celery در پردازش‌ های جداگانه فعالیت کنند.

امکان اجرای چند Worker نیز وجود دارد. این Worker ها می‌ توانند روی یک سیستم یا چند سرور مختلف قرار داشته باشند:

               ┌── Worker 1
Broker ────────┼── Worker 2
               └── Worker 3

با افزایش تعداد Worker ها، چند Task می‌ توانند به‌صورت هم‌ زمان پردازش شوند. البته تعداد مناسب Worker ها به عواملی مانند نوع Task، منابع سرور و میزان بار سیستم بستگی دارد.

هر Worker باید Task هایی را که قرار است اجرا کند بشناسد. اگر فایل یا ماژول مربوط به یک Task توسط Worker بارگذاری نشده باشد، Worker نمی‌ تواند آن Task را اجرا کند.

Result Backend:

Broker وظیفه نگهداری و انتقال پیام Task را بر عهده دارد، اما معمولا محل ذخیره نتیجه نهایی Task نیست. برای ذخیره وضعیت و خروجی Task از بخشی به نام Result Backend استفاده می‌ شود.

برای مثال، Task زیر یک مقدار عددی برمی‌ گرداند:

@app.task
def add(x, y):
    return x + y

اگر Task به شکل زیر اجرا شود:

result = add.delay(10, 20)

نتیجه محاسبه، یعنی عدد 30، می‌تواند در Result Backend ذخیره شود. سپس برنامه با استفاده از شناسه Task می‌ تواند وضعیت یا نتیجه آن را دریافت کند.

در هنگام ساخت برنامه Celery می‌توان Broker و Result Backend را مشخص کرد:

from celery import Celery

app = Celery(
    "demo",
    broker="redis://localhost:6379/0",
    backend="redis://localhost:6379/1",
)

در این مثال، Redis هم برای Broker و هم برای Result Backend استفاده شده است؛ اما داده‌ های آن‌ ها در دو Database جداگانه Redis قرار می‌ گیرند.

Result Backend می‌تواند یکی از گزینه‌ های زیر باشد:

Redis

پایگاه داده

RPC

Cache

سایر Backend های پشتیبانی‌ شده توسط Celery

استفاده از Result Backend همیشه ضروری نیست. بسیاری از Task ها فقط باید اجرا شوند و برنامه نیازی به دریافت مقدار بازگشتی آن‌ ها ندارد.

برای نمونه، در Task ارسال اعلان معمولا ذخیره نتیجه ضرورت ندارد:

@app.task(ignore_result=True)
def send_notification(user_id):
    ...

غیرفعال کردن نتایج غیرضروری می‌تواند مصرف حافظه، فضای ذخیره‌ سازی و پردازش‌ های اضافی را کاهش دهد.

Celery Beat:

Celery Beat بخش زمان‌ بندی Celery است. از Beat برای اجرای Task ها در زمان‌ های مشخص یا به‌ صورت دوره‌ ای استفاده می‌ شود.

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

هر پنج دقیقه

هر شب ساعت 12

ابتدای هر ماه

هر هفته در روز مشخص

یک‌بار در یک زمان از پیش تعیین‌ شده

Celery Beat برنامه زمان‌ بندی را بررسی می‌ کند و در زمان مناسب، پیام مربوط به Task را به Broker ارسال می‌ کند.

جریان کار Beat به شکل زیر است:

Celery Beat

↓

Broker

↓

Worker

↓

اجرای Task

نکته مهم این است که Beat خودش Task را اجرا نمی‌ کند. Beat فقط تشخیص می‌ دهد چه زمانی باید Task ارسال شود. اجرای واقعی Task همچنان توسط Worker انجام می‌ شود.

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

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

در مجموع، ارتباط اجزای اصلی Celery را می‌توان به این شکل نمایش داد:

Producer

↓

ایجاد پیام Task

↓

Broker

↓

دریافت پیام توسط Worker

↓

اجرای Task

↓

Result Backend

در صورت نیاز Celery Beat نیز می‌ تواند در زمان‌ های مشخص مانند یک Producer پیام Task را به Broker ارسال کند.

هرکدام از این اجزا مسئولیت مشخصی دارند: Task عملیاتی است که باید انجام شود، Producer درخواست اجرای آن را ارسال می‌ کند، Broker پیام را نگه می‌ دارد، Worker عملیات را اجرا می‌ کند، Result Backend نتیجه را ذخیره می‌کند و Celery Beat اجرای زمان‌ بندی‌ شده Task ها را مدیریت می‌ کند.

تعریف و اجرای اولین Task در Celery:

تا اینجا مقاله بیشتر سعی شد در خصوص ساختار کلی Celery و اجزای آن آشنا شویم ولی الان زمان آن رسیده تا یک نمونه ساده را به صورت عملی پیاده سازی کنیم و ببینیم یک Task چگونه تعریف٬ به صف ارسال و توسط Worker اجرا می شود.

در این مثال یک Task ساده برای جمع دو عدد می نویسیم! در پروژه های واقعی این عملیات بسیار سبک هست و نیاز نیست به Celery سپرده شود ولی به دلیل سادگی٬ گزینه مناسبی برای مشاهده جریان کامل ارسال و اجرای Task می باشد.

برای اجرای این مثال به اجزای زیر نیاز داریم:

پایتون

کتابخانه Celery

یک Message Broker

یک یا چند Celery Worker

در این آموزش از Redis به‌عنوان Broker استفاده می‌ کنیم. Redis پیام Task را از برنامه دریافت می‌ کند و تا زمان دریافت آن توسط Worker در صف نگه می‌ دارد.

ساخت محیط مجازی:

ابتدا پیش از نصب کتابخانه ها باید یک محیط مجازی برای پروژه ایجاد کنیم تا بتوانیم وابستگی های آن را از سایر پروژه های پایتون جدا کنیم.

ابتدا یک پوشه برای پروژه ایجاد می کنیم و وارد آن می شویم:

mkdir celery_demo
cd celery_demo

سپس محیط مجازی را ایجاد می‌ کنیم:

python -m venv .venv

در سیستم‌ عامل‌ های Linux و macOS می‌ توان محیط مجازی را با دستور زیر فعال کرد:

source .venv/bin/activate

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

نصب Celery و وابستگی Redis:

برای استفاده از Redis در کنار Celery باید علاوه بر خود Celery، وابستگی‌ های مربوط به Redis را نیز نصب کنیم. بسته celery[redis] این وابستگی‌ ها را به‌صورت یکجا نصب می‌ کند.

pip install -U "celery[redis]"

برای اطمینان از نصب Celery می‌ توان دستور زیر را اجرا کرد:

celery --version

در صورت نصب صحیح، نسخه Celery در خروجی نمایش داده می‌ شود.

اجرای Redis:

پیش از اجرای Worker باید Redis فعال باشد٬ زیرا برنامه و Worker برای ارسال و دریافت پیام‌ ها به آن متصل می‌ شوند.

اگر Docker روی سیستم نصب شده باشد، می‌توان Redis را با دستور زیر اجرا کرد:

docker run \
    --name celery-redis \
    -p 6379:6379 \
    -d redis:alpine

در این دستور:

گزینه --name نام Container را مشخص می‌ کند.

گزینه -p 6379:6379 پورت Redis را در اختیار سیستم قرار می‌ دهد.

گزینه -d باعث می‌ شود Container در پس‌زمینه اجرا شود.

مقدار redis:alpine تصویر سبک Redis را مشخص می‌ کند.

برای اینکه بررسی کنیم که Container فعال می باشد یا خیر می توانیم از دستور زیر آن را چک کنیم:

docker ps

در صورت نیاز به متوقف‌کردن Redis نیز می‌ توان دستور زیر را اجرا کرد:

docker stop celery-redis

و برای اجرای مجدد همان Container از دستور زیر استفاده کرد:

docker start celery-redis

Redis به‌صورت پیش‌ فرض روی پورت 6379 اجرا می‌ شود. آدرس اتصال معمول آن در Celery نیز به شکل زیر است:

redis://localhost:6379/0

عدد انتهای آدرس، شماره Database مورد استفاده در Redis را مشخص می‌ کند.

ساختار پروژه:

در این مثال ما از یک ساختار پروژه ساده استفاده می کنیم برای مثال ساختار را به شکل زیر در نظر میگیریم:

celery_demo/
├── .venv/
├── celery_app.py
├── tasks.py
└── run_task.py

هرکدام از این فایل‌ ها مسئولیت مشخصی دارند:

فایل celery_app.py برنامه Celery و تنظیمات اتصال به Redis را نگه می‌ دارد.

فایل tasks.py شامل Task های پروژه است.

فایل run_task.py نقش Producer را دارد و درخواست اجرای Task را ارسال می‌ کند.

ساخت برنامه Celery:

ابتدا فایل celery_app.py را ایجاد کرده و کد زیر را در آن قرار می‌ دهیم:

from celery import Celery


app = Celery(
    "celery_demo",
    broker="redis://localhost:6379/0",
    backend="redis://localhost:6379/1",
    include=["tasks"],
)

در این کد ابتدا کلاس Celery را وارد کرده و سپس یک نمونه از آن با نام app ساخته‌ ایم. این نمونه نقطه مرکزی تنظیمات و مدیریت Task های پروژه محسوب می‌ شود.

آرگومان اول:

"celery_demo"

نام برنامه Celery است. این نام برای شناسایی برنامه و تولید بعضی نام‌ های داخلی مورد استفاده قرار می‌ گیرد.

آرگومان broker آدرس Message Broker را مشخص می‌ کند:

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

برنامه پیام‌ های مربوط به اجرای Task را به این آدرس ارسال می‌ کند. Redis نیز پیام‌ ها را نگه می‌ دارد تا یکی از Worker ها آن‌ها را دریافت کند.

آرگومان backend محل ذخیره وضعیت و نتیجه Task را مشخص می‌ کند:

backend="redis://localhost:6379/1"

در این مثال از Redis Database شماره صفر برای Broker و Database شماره یک برای Result Backend استفاده کرده‌ ایم. جدا کردن این دو Database برای مثال آموزشی باعث می‌ شود پیام‌های صف و نتیجه Task ها در فضای منطقی متفاوتی قرار بگیرند.

Result Backend برای اجرای Task ضروری نیست، اما چون می‌ خواهیم نتیجه عملیات جمع را مشاهده کنیم، آن را فعال کرده‌ ایم. Celery بدون Result Backend نیز می‌ تواند Task ها را ارسال و اجرا کند اما برنامه امکان نگهداری و دریافت نتیجه آن‌ ها را نخواهد داشت.

در نهایت گزینه زیر به Celery اعلام می‌کند که هنگام شروع برنامه، ماژول tasks را نیز بارگذاری کند:

include=["tasks"]

Worker باید ماژولی را که Task در آن تعریف شده است بارگذاری کند٬ در غیر این صورت Task را نمی‌شناسد و نمی‌ تواند آن را اجرا کند.

تعریف اولین Task:

اکنون فایل tasks.py را ایجاد کرده و Task مربوط به جمع دو عدد را در آن می‌ نویسیم:

from celery_app import app

@app.task
def add(x, y):
    return x + y

در این فایل ابتدا نمونه Celery را از فایل celery_app.py وارد کردیم:

from celery_app import app

سپس با استفاده از دکوراتور @app.task تابع add را به‌عنوان یک Task در Celery ثبت کردیم:

@app.task
def add(x, y):
    return x + y

تابع add دو ورودی دریافت می‌ کند و حاصل جمع آن‌ ها را برمی‌ گرداند.

بدون استفاده از دکوراتور @app.task، تابع add فقط یک تابع معمولی پایتون خواهد بود. این دکوراتور باعث می‌شود Celery تابع را شناسایی و در فهرست Taskهای ثبت‌شده قرار دهد.

Celery معمولاً نام Task را با ترکیب مسیر ماژول و نام تابع ایجاد می‌ کند. بنابراین نام Task این مثال معمولا به شکل زیر خواهد بود:

tasks.add

Worker هنگام دریافت پیام از Broker، این نام را بررسی می‌ کند تا تشخیص دهد کدام تابع باید اجرا شود.

اجرای Worker:

پس از تعریف Task باید حداقل یک Worker اجرا کنیم. Worker پردازشی مستقل است که به Broker متصل می‌ شود و منتظر دریافت Task های جدید باقی می‌ ماند.

در مسیر اصلی پروژه دستور زیر را اجرا می‌ کنیم:

celery -A celery_app worker --loglevel=INFO

گزینه -A یا --app مشخص می‌ کند برنامه Celery در کدام ماژول قرار دارد:

-A celery_app

Celery فایل celery_app.py را وارد کرده و نمونه‌ ای با نام app را در آن پیدا می‌ کند.

بخش زیر نیز سطح نمایش Log ها را روی INFO قرار می‌ دهد:

--loglevel=INFO

بعد از اجرای موفق Worker، اطلاعاتی درباره Broker، Result Backend، میزان Concurrency و Task های ثبت‌ شده نمایش داده می‌ شود. در میان Task های ثبت‌ شده باید نام Task خود را نیز مشاهده کنیم:

[tasks]
  . tasks.add

پس از این مرحله Worker فعال باقی می‌ ماند و منتظر دریافت پیام‌ های جدید از Redis می‌ شود.

ترمینالی که Worker در آن اجرا شده است باید باز باقی بماند. برای ارسال Task، یک ترمینال جدید باز کرده و محیط مجازی پروژه را در آن فعال می‌ کنیم.

ارسال Task برای اجرا:

اکنون فایل run_task.py را ایجاد کرده و کد زیر را در آن قرار می‌ دهیم:

from tasks import add

result = add.delay(10, 20)

print(f"Task ID: {result.id}")
print("Task sent successfully.")

در این کد ابتدا Task را وارد کردیم:

from tasks import add

سپس با استفاده از متد delay() درخواست اجرای آن را ارسال کردیم:

result = add.delay(10, 20)

فراخوانی delay() تابع را در پردازش فعلی اجرا نمی‌ کند. این متد یک پیام شامل نام Task و آرگومان‌ های آن ایجاد کرده و برای Broker ارسال می‌ کند. Worker پیام را از Broker دریافت کرده و تابع add را در پردازش خود اجرا می‌ کند. متد delay() میان‌ بری برای apply_async() است، با این تفاوت که تنظیمات پیشرفته اجرای Task را دریافت نمی‌ کند.

در نتیجه جریان اجرای این کد به شکل زیر خواهد بود:

run_task.py

↓

ایجاد پیام tasks.add

↓

ارسال پیام به Redis

↓

دریافت پیام توسط Worker

↓

اجرای add(10, 20)

↓

ذخیره نتیجه در Result Backend

برای اجرای فایل از دستور زیر استفاده می‌ کنیم:

python run_task.py

خروجی برنامه مشابه نمونه زیر خواهد بود:

Task ID: 8f3c6b2a-...
Task sent successfully.

شناسه Task معمولاً یک مقدار یکتا است که برای پیگیری وضعیت و نتیجه آن استفاده می‌ شود.

همزمان در ترمینال Worker نیز Log مربوط به دریافت و اجرای Task نمایش داده خواهد شد. خروجی Worker نشان می‌دهد که Task با موفقیت دریافت شده و مقدار 30 را برگردانده است.

دریافت نتیجه Task:

مقداری که از delay() برگردانده می‌ شود، نتیجه مستقیم تابع نیست٬ بلکه یک شی از نوع AsyncResult است. این شی اطلاعاتی مانند شناسه، وضعیت و نتیجه Task را در اختیار برنامه قرار می‌ دهد.

فایل run_task.py را می‌توانیم به شکل زیر تغییر دهیم:

from tasks import add


result = add.delay(10, 20)

print(f"Task ID: {result.id}")
print(f"Initial status: {result.status}")

task_result = result.get(timeout=10)

print(f"Final status: {result.status}")
print(f"Result: {task_result}")

پس از اجرای فایل، خروجی مشابه زیر خواهد بود:

Task ID: 8f3c6b2a-...
Initial status: PENDING
Final status: SUCCESS
Result: 30

متد get() منتظر پایان Task می‌ ماند و سپس نتیجه آن را برمی‌ گرداند:

result.get(timeout=10)

مقدار timeout=10 مشخص می‌ کند برنامه حداکثر ده ثانیه برای دریافت نتیجه منتظر بماند.

استفاده از get() در این مثال فقط برای مشاهده نتیجه Task است. در مسیر اصلی یک درخواست وب معمولا نباید بلافاصله بعد از ارسال Task از get() استفاده کنیم٬ زیرا این کار برنامه را منتظر پایان Task نگه می‌ دارد و اجرای غیرهمزمان را دوباره به یک جریان همزمان تبدیل می‌ کند.

مستندات Celery نیز اشاره می‌ کنند که انتظار مستقیم برای نتیجه معمولا کاربرد محدودی دارد، چون مزیت اجرای غیرهمزمان را کاهش می‌ دهد.

در یک پروژه واقعی معمولاً شناسه Task ذخیره می‌شود و وضعیت آن در درخواست دیگری بررسی خواهد شد:

result = add.delay(10, 20)

task_id = result.id

برای مثال، یک API می‌ تواند بلافاصله شناسه Task را به کاربر برگرداند:

return {
    "message": "Task submitted successfully",
    "task_id": result.id,
}

سپس کاربر یا رابط کاربری می‌ تواند با استفاده از این شناسه، وضعیت Task را از یک Endpoint دیگر دریافت کند.

تفاوت اجرای مستقیم و اجرای Celery:

از آنجا که Task های Celery همچنان اشیای قابل‌ فراخوانی پایتون هستند، می‌ توان تابع add را به شکل مستقیم نیز اجرا کرد:

result = add(10, 20)

print(result)

در این حالت خروجی مستقیماً برابر با 30 خواهد بود. تابع در همان پردازش فعلی اجرا می‌ شود و هیچ پیامی به Broker ارسال نخواهد شد.

اما در حالت زیر:

result = add.delay(10, 20)

تابع در پردازش فعلی اجرا نمی‌ شود. یک پیام برای Broker ارسال می‌ شود و Worker مسئول اجرای آن خواهد بود. مستندات Celery این تفاوت را به این صورت تعریف می‌ کنند که فراخوانی مستقیم Task در پردازش جاری انجام می‌ شود، در حالی که delay() یا apply_async() پیام اجرای Task را ارسال می‌ کنند.

اگر Worker فعال نباشد چه اتفاقی می‌ افتد؟

اگر Redis فعال باشد اما Worker اجرا نشده باشد، Producer همچنان می‌ تواند پیام Task را به Broker ارسال کند.

پیام در صف باقی می‌ ماند و زمانی که Worker اجرا شود، آن را دریافت و پردازش می‌ کند:

Producer

↓

Redis

↓

پیام در صف باقی می‌ ماند

↓

Worker اجرا می‌شود

↓

Task پردازش می‌شود

اما اگر Broker در دسترس نباشد، Producer نمی‌ تواند پیام را به صف ارسال کند و معمولا با خطای اتصال رو به‌ رو خواهد شد.

این موضوع تفاوت نقش Broker و Worker را به‌ خوبی نشان می‌ دهد. Broker وظیفه دریافت و نگهداری پیام را بر عهده دارد، در حالی که Worker مسئول اجرای واقعی Task است.

خطاهای رایج در اجرای اولین Task:

اگر هنگام اجرای مثال با خطا رو به‌ رو شدیم، بهتر است موارد زیر را بررسی کنیم.

خطای اتصال به Redis:

اگر خطایی مانند Connection refused نمایش داده شود، احتمالا Redis اجرا نشده یا آدرس Broker صحیح نیست.

ابتدا وضعیت Container را بررسی می‌ کنیم:

docker ps

همچنین باید آدرس Broker در فایل celery_app.py با آدرس Redis هماهنگ باشد:

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

خطای Task ثبت‌ نشده:

اگر Worker پیامی مشابه زیر نمایش دهد:

Received unregistered task of type 'tasks.add'

یعنی Worker ماژول مربوط به Task را بارگذاری نکرده است.

در این مثال وجود تنظیم زیر باعث بارگذاری فایل tasks.py می‌ شود:

include=["tasks"]

پس از هر تغییر در تعریف Task ها نیز باید Worker را متوقف و دوباره اجرا کنیم تا کد جدید بارگذاری شود.

اجرا نشدن دستور Celery:

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

pip show celery

همچنین می‌ توان Worker را از طریق ماژول پایتون اجرا کرد:

python -m celery -A celery_app worker --loglevel=INFO

جمع‌بندی جریان اجرای اولین Task:

در این مثال یک برنامه Celery ایجاد کردیم، Redis را به‌ عنوان Broker و Result Backend تنظیم کردیم، تابع add را به‌عنوان Task ثبت کردیم و یک Worker برای اجرای آن راه‌ اندازی کردیم.

جریان کامل اجرای Task به شکل زیر بود:

add.delay(10, 20)

↓

ایجاد پیام Task

↓

ارسال پیام به Redis

↓

دریافت پیام توسط Worker

↓

اجرای تابع add

↓

ذخیره نتیجه در Redis

↓

دریافت نتیجه از طریق AsyncResult

نکته اصلی این است که متد delay() منتظر پایان عملیات نمی‌ ماند. وظیفه آن ارسال درخواست اجرا به Celery است. اجرای واقعی تابع در یک پردازش مستقل و توسط Worker انجام می‌ شود.

دستورات کاربردی مدیریت Worker:

در ادامه چند دستور کاربردی در خصوص مدیریت Worker را ها برای شما گردآوری کردیم.

بررسی Worker های فعال:

celery -A tasks inspect ping

نمایش Task های در حال اجرا:

celery -A tasks inspect active

نمایش Task های رزرو شده:

celery -A tasks inspect reserved

نمایش Task های زمان‌ بندی‌ شده:

celery -A tasks inspect scheduled

نمایش Task های ثبت‌ شده:

celery -A tasks inspect registered

نمایش وضعیت Worker ها:

celery -A tasks status

پاک کردن صف:

celery -A tasks purge

دستور purge تمام پیام‌ های منتظر را حذف می‌ کند و باید با احتیاط استفاده شود.

لغو Task:

celery -A tasks control revoke TASK_ID

تلاش برای متوقف کردن Task در حال اجرا:

celery -A tasks control revoke TASK_ID \
  --terminate \
  --signal=SIGTERM

متوقف کردن اجباری Task ممکن است منابعی مانند فایل، connection یا transaction را در وضعیت نامناسب باقی بگذارد.

اشتباهات رایج در استفاده از Celery:

همانطور که در ابتدای این مقاله ذکر شد استفاده از Celery می تواند سرعت پاسخگویی٬ مقیاس پذیری و پایداری یک برنامه را افزایش دهد اما به این معنا نیست که صرفا انتقال چند تابع به Worker یعنی یک ساختار مناسب و بهینه طراحی کردیم.

بعضی اشتباهات در Celery ممکن است باعث شود یک Task چند مرتبه اجرا شود و Worker ها برای مدت طولانی درگیر بمانند و باعث طولانی شدن صف و بزرگ شدن آن شود و یا حتی بخشی از عملیات های مهم بدون نتیجه باقی بماند.

در ادامه، تعدادی از اشتباهات رایج در استفاده از Celery و روش جلوگیری از آن‌ها را بررسی می‌ کنیم.

1- استفاده از Celery برای تمام عملیات‌ ها:

در قدم اول برای اینکه وارد گودال اشتباه نشوید این است که هر تابع یا عملیات کوچکی را به Celery منتقل نکنید!

Celery زمانی ارزش ایجاد می‌ کند که یک عملیات حداقل یکی از ویژگی‌ های زیر را داشته باشد:

اجرای آن زمانبر باشد.

برای پاسخ اولیه به کاربر ضروری نباشد.

بتوان آن را با تاخیر اجرا کرد.

نیاز به Retry یا زمان‌ بندی داشته باشد.

بخواهیم آن را مستقل از برنامه اصلی مقیاس‌ دهی کنیم.

اجرای آن به سرویس‌ های خارجی وابسته باشد.

برای مثال، محاسبه حاصل جمع دو عدد عملیات مناسبی برای Celery نیست:

result = add.delay(10, 20)

در یک پروژه واقعی، هزینه ساخت پیام، ارسال آن به Broker، دریافت پیام توسط Worker و ذخیره نتیجه ممکن است بسیار بیشتر از هزینه خود محاسبه باشد.

در مقابل، عملیات‌هایی مانند تولید گزارش، پردازش تصویر، ارسال گروهی ایمیل، ارتباط با سرویس حسابداری یا تبدیل فایل، گزینه‌ های مناسب‌ تری برای اجرای پس‌زمینه هستند.

بنابراین پیش از تعریف هر Task باید از خود بپرسیم:

آیا انجام عملیات خارج از پردازش اصلی می تواند ارزش بیشتری از پیچیده شدن اضافی پروژه داشته باشد؟

2- فراخوانی مستقیم Task به‌جای ارسال آن به صف:

یکی از اشتباهات رایج این است که Task را مانند یک تابع معمولی فراخوانی کنیم و تصور کنیم توسط Worker اجرا خواهد شد.

کد زیر Task را در همان پردازش فعلی اجرا می‌ کند:

result = generate_report(report_id)

در این حالت هیچ پیامی به Broker ارسال نمی‌ شود و Worker نقشی در اجرای تابع ندارد.

برای ارسال Task به Celery باید از delay() یا apply_async() استفاده کنیم:

result = generate_report.delay(report_id)

یا:

result = generate_report.apply_async(
    args=[report_id],
)

متد delay() یک میان‌ بر برای apply_async() است، اما گزینه‌ های پیشرفته‌ ای مانند زمان اجرا، صف مقصد، تاریخ انقضا و اولویت را دریافت نمی‌ کند. فراخوانی مستقیم Task نیز آن را در پردازش جاری اجرا می‌ کند و پیامی برای Worker ارسال نمی‌ شود.

3- منتظر ماندن برای نتیجه بلافاصله پس از ارسال Task:

یکی از اهداف اصلی Celery این است که برنامه اصلی برای پایان عملیات زمانبر منتظر نماند. با این حال گاهی Task را ارسال کرده و بلافاصله متد get() را فراخوانی می‌ کنیم:

result = generate_report.delay(report_id)

report_path = result.get()

return {
    "report_path": report_path,
}

این کد ابتدا Task را به Celery ارسال می‌ کند، اما سپس اجرای برنامه را تا زمان آماده‌ شدن نتیجه متوقف نگه می‌ دارد. در نتیجه بخش زیادی از مزیت اجرای غیرهمزمان از بین می‌ رود.

راهکار بهتر این است که شناسه Task را به کاربر برگردانیم:

result = generate_report.delay(report_id)

return {
    "message": "Report generation started",
    "task_id": result.id,
}

سپس یک Endpoint جداگانه برای بررسی وضعیت Task ایجاد کنیم:

from celery.result import AsyncResult


def get_task_status(task_id):
    result = AsyncResult(task_id)

    return {
        "task_id": task_id,
        "status": result.status,
        "result": result.result if result.successful() else None,
    }

استفاده از get() داخل یک Task برای انتظار نتیجه Task دیگر خطرناک‌ تر است. این کار می‌ تواند Worker ها را بلاک کند و در صورت پرشدن Pool حتی باعث Deadlock شود. برای ایجاد جریان‌ های وابسته بهتر است از قابلیت‌ هایی مانند chain استفاده شود.

نمونه نامناسب:

@app.task
def process_document(document_id):
    text = extract_text.delay(document_id).get()
    summary = summarize_text.delay(text).get()

    return summary

نمونه بهتر:

from celery import chain

workflow = chain(
    extract_text.s(document_id),
    summarize_text.s(),
)

workflow.apply_async()

4- ارسال اشیای پیچیده به‌ عنوان آرگومان Task:

اطلاعات ارسال‌ شده به Task باید در قالب پیام قرار بگیرند و سریال‌ سازی شوند. Serializer پیش‌فرض Celery از نوع JSON است و برای داده‌ های ساده‌ ای مانند رشته، عدد، فهرست و دیکشنری مناسب است.

بنابراین بهتر است اشیایی مانند موارد زیر را مستقیما به Task ارسال نکنیم:

نمونه Model

Connection پایگاه داده

فایل بازشده

Request کاربر

Object های وابسته به Framework

کلاس‌ ها و نمونه‌ های پیچیده پایتون

نمونه نامناسب در Django:

user = User.objects.get(id=user_id)

send_welcome_email.delay(user)

راهکار مناسب‌ تر این است که فقط شناسه رکورد را ارسال کنیم:

send_welcome_email.delay(user.id)

و رکورد را داخل Worker دریافت کنیم:

from celery import shared_task

from users.models import User

@shared_task
def send_welcome_email(user_id):
    user = User.objects.get(id=user_id)

    # Send email

این روش چند مزیت دارد:

حجم پیام کوچک‌ تر می‌شود.

Task به یک Object قدیمی و تغییرکرده وابسته نمی‌ ماند.

مشکلات سریال‌ سازی کاهش پیدا می‌ کنند.

Worker اطلاعات موردنیاز را در زمان اجرا از منبع اصلی دریافت می‌ کند.

البته ممکن است رکورد تا زمان اجرای Task حذف یا تغییر کرده باشد. بنابراین Task باید چنین شرایطی را نیز مدیریت کند:

@shared_task
def send_welcome_email(user_id):
    try:
        user = User.objects.get(id=user_id)
    except User.DoesNotExist:
        return {
            "status": "ignored",
            "reason": "user_not_found",
        }

    # Send email

5- قراردادن داده‌ های بزرگ داخل پیام Task:

Broker برای نگهداری و انتقال پیام طراحی شده است، نه برای انتقال فایل‌ های بزرگ یا حجم زیادی از اطلاعات بنابراین باید به این مورد دقت شود!

نمونه نامناسب:

process_video.delay(video_binary_data)

یا:

generate_report.delay(large_list_of_records)

این کار می‌ تواند باعث افزایش مصرف حافظه Broker، کندی سریال‌ سازی و افزایش زمان انتقال پیام شود.

روش بهتر این است که فایل یا اطلاعات اصلی را در یک محل مناسب ذخیره کنیم و فقط شناسه یا مسیر آن را به Task ارسال کنیم:

process_video.delay(video_id)

سپس Task اطلاعات را از پایگاه داده، Object Storage یا فضای ذخیره‌ سازی دریافت کند:

@shared_task
def process_video(video_id):
    video = Video.objects.get(id=video_id)
    input_path = video.file.path

    # Process video

برای فایل‌ های بزرگ می‌ توان از سرویس‌ هایی مانند Object Storage استفاده کرد و فقط کلید فایل را داخل پیام قرار داد:

process_file.delay(
    bucket_name="uploads",
    object_key="videos/input-123.mp4",
)

6- ارسال اطلاعات حساس در آرگومان‌ های Task:

آرگومان‌ های Task ممکن است در Broker، Log ها، ابزارهای مانیتورینگ یا Result Backend قابل مشاهده باشند. بنابراین نباید اطلاعات حساسی مانند رمز عبور، Token، شماره کامل کارت بانکی یا کلید خصوصی را مستقیما داخل پیام قرار دهیم.

نمونه نامناسب:

charge_card.delay(
    card_number="1234-5678-9012-3456",
    cvv="123",
    amount=500,
)

روش بهتر این است که اطلاعات حساس در یک محل امن ذخیره شوند و تنها یک شناسه محدود و غیرحساس به Task ارسال شود:

charge_card.delay(
    payment_id=payment.id,
)

Celery امکان تغییر نمایش آرگومان‌ ها در Log ها را با argsrepr و kwargsrepr فراهم می‌ کند، اما این تنظیم فقط نمایش اطلاعات را تغییر می‌ دهد و داده اصلی همچنان می‌تواند داخل پیام Broker وجود داشته باشد.

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

7- ارسال Task پیش از Commit شدن Transaction:

این اشتباه بیشتر در پروژه‌ های Django مشاهده می‌ شود.

فرض کنید ابتدا یک سفارش ایجاد کرده و بلافاصله Task مربوط به آن را ارسال می‌ کنیم:

order = Order.objects.create(
    user=request.user,
    total_price=total_price,
)

generate_invoice.delay(order.id)

ممکن است Worker بسیار سریع پیام را دریافت کند، در حالی که Transaction مربوط به ایجاد سفارش هنوز Commit نشده است. در این حالت Worker نمی‌ تواند سفارش را در پایگاه داده پیدا کند.

Celery این مورد را یکی از خطاهای رایج در پروژه‌ های Django می‌ داند، زیرا ممکن است Task پیش از ذخیره نهایی تغییرات اجرا شود.

یکی از راهکارها استفاده از transaction.on_commit() است:

from django.db import transaction

order = Order.objects.create(
    user=request.user,
    total_price=total_price,
)

transaction.on_commit(
    lambda: generate_invoice.delay(order.id)
)

در نسخه‌های جدید Celery می‌ توان برای Task های Django از delay_on_commit() نیز استفاده کرد:

generate_invoice.delay_on_commit(order.id)

در این حالت پیام Task فقط پس از Commit شدن موفق Transaction به Broker ارسال می‌ شود. باید توجه داشت که delay_on_commit() در همان لحظه شناسه Task را برنمی‌ گرداند، زیرا Task هنوز برای Broker ارسال نشده است.

این روش علاوه بر جلوگیری از اجرا شدن زود هنگام Task، یک مزیت دیگر نیز دارد: اگر Transaction با خطا مواجه شود و Rollback شود، Task مربوط به داده نامعتبر نیز ارسال نخواهد شد.

8- فرض‌ کردن اینکه هر Task دقیقا یک بار اجرا می‌ شود:

اینکه تصور کنیم در هر Task قطعا یک بار اجرا می شود هم اشتباه می باشد و در سیستم‌ های توزیع‌ شده این مورد می تواند کاملا خلاف تصور باشد.

ممکن است Worker پس از انجام عملیات، پیش از ثبت تایید پیام متوقف شود یا پیام در شرایط خاص دوباره تحویل داده شود. به همین دلیل بهتر است Task های حساس را Idempotent طراحی کنیم؛ یعنی اجرای چند باره آن‌ ها با ورودی یکسان، اثر ناخواسته جدیدی ایجاد نکند. مستندات Celery نیز توصیه می‌کنند Task ها تا حد امکان Idempotent باشند.

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

@shared_task
def decrease_inventory(product_id, quantity):
    product = Product.objects.get(id=product_id)

    product.stock -= quantity
    product.save()

اگر این Task دوبار اجرا شود، موجودی محصول نیز دوبار کاهش پیدا می‌کند.

برای جلوگیری از این مشکل می‌ توان برای هر عملیات یک شناسه یکتا در نظر گرفت:

@shared_task
def decrease_inventory(order_item_id):
    operation, created = InventoryOperation.objects.get_or_create(
        order_item_id=order_item_id,
        defaults={
            "status": InventoryOperation.Status.PENDING,
        },
    )

    if not created:
        return {
            "status": "ignored",
            "reason": "already_processed",
        }

    order_item = OrderItem.objects.get(id=order_item_id)
    product = order_item.product

    product.stock -= order_item.quantity
    product.save()

    operation.status = InventoryOperation.Status.COMPLETED
    operation.save()

    return {
        "status": "completed",
    }

راهکار دقیق به نوع عملیات بستگی دارد، اما معمولاً می‌ توان از موارد زیر استفاده کرد:

Unique Constraint

شناسه یکتای عملیات

وضعیت پردازش

قفل پایگاه داده

کلید Idempotency

get_or_create()

بررسی نتیجه عملیات قبلی

این موضوع برای عملیات‌ هایی مانند پرداخت، کاهش موجودی، صدور فاکتور، ارسال پیامک و ثبت اطلاعات در سرویس‌ های خارجی اهمیت بیشتری دارد.

9- Retry کردن تمام خطاها بدون محدودیت:

Retry برای خطاهای موقت مفید است، اما نباید هر Exception را بدون بررسی دوباره اجرا کنیم.

نمونه نامناسب:

@app.task(
    autoretry_for=(Exception,),
    retry_kwargs={
        "max_retries": None,
    },
)
def sync_order(order_id):
    ...

در این مثال تقریبا هر خطایی باعث Retry نامحدود می‌ شود. اگر خطا به دلیل داده نامعتبر یا یک باگ برنامه‌ نویسی باشد، اجرای مجدد آن مشکلی را حل نمی‌ کند و فقط صف را شلوغ‌ تر می‌ کند.

بهتر است خطاها را به دو گروه تقسیم کنیم:

خطاهای موقت و قابل Retry:

قطع موقت شبکه

Timeout سرویس خارجی

پاسخ موقت 503

محدودیت نرخ درخواست

در دسترس نبودن موقت پایگاه داده

خطاهای دائمی و غیرقابل Retry:

نامعتبر بودن ورودی

وجود نداشتن رکورد

خطای احراز هویت دائمی

فرمت نادرست فایل

خطای منطقی در کد

نمونه مناسب‌ تر:

from celery import shared_task
from requests import RequestException


@shared_task(
    autoretry_for=(RequestException,),
    retry_backoff=True,
    retry_backoff_max=600,
    retry_jitter=True,
    retry_kwargs={
        "max_retries": 5,
    },
)
def sync_order(order_id):
    ...

Celery امکان Retry خودکار برای Exception های مشخص، محدود کردن تعداد تلاش‌ ها و استفاده از Exponential Backoff را فراهم می‌ کند.

هنگام فعال‌بودن Backoff، Jitter نیز می‌ تواند زمان تلاش‌ های مجدد را پراکنده کند تا تعداد زیادی Task دقیقا در یک لحظه دوباره اجرا نشوند.

همچنین بهتر است خطای اصلی را در Retry نگه داریم:

@app.task(bind=True, max_retries=5)
def call_external_service(self, order_id):
    try:
        ...
    except RequestException as exc:
        raise self.retry(
            exc=exc,
            countdown=60,
        )

استفاده از:

raise self.retry(...)

از ادامه ناخواسته اجرای کد بعد از Retry نیز جلوگیری می‌ کند.

10- ساخت Task های بیش از حد بزرگ:

نباید Task های بیش از حد بزرگ ساخته شود که می تواند مشکل های جدی ایجاد کند.

گاهی تمام مراحل یک عملیات طولانی را داخل یک Task قرار می‌ دهیم:

@app.task
def process_uploaded_video(video_id):
    download_video(video_id)
    validate_video(video_id)
    extract_audio(video_id)
    generate_thumbnail(video_id)
    convert_video(video_id)
    upload_result(video_id)
    notify_user(video_id)

این ساختار چند مشکل ایجاد می‌ کند:

تشخیص مرحله‌ ای که خطا داده دشوار می‌ شود.

Retry باعث اجرای دوباره همه مراحل می‌ شود.

مشاهده میزان پیشرفت سخت‌ تر است.

یک Worker برای مدت طولانی درگیر می‌ ماند.

استفاده مجدد از مراحل مختلف دشوار می‌ شود.

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

from celery import chain

workflow = chain(
    download_video.s(video_id),
    validate_video.s(),
    extract_audio.s(),
    generate_thumbnail.s(),
    convert_video.s(),
    upload_result.s(),
    notify_user.s(),
)

workflow.apply_async()

البته تقسیم بیش از حد نیز مناسب نیست. تبدیل هر خط کوچک کد به یک Task جداگانه، سربار پیام و پیچیدگی Workflow را افزایش می‌ دهد.

مرز مناسب زمانی است که هر Task یک مسئولیت مشخص، قابل Retry و قابل مانیتور داشته باشد.

11- قراردادن Task های کوتاه و طولانی در یک صف:

فرض کنید Task های زیر همگی در صف پیش‌ فرض قرار داشته باشند:

send_sms
send_email
generate_pdf
convert_video
generate_annual_report

اگر چند Task تبدیل ویدئو یا تولید گزارش طولانی صف را اشغال کنند، Task های سریع و مهمی مانند ارسال پیامک نیز مجبور می‌ شوند مدت زیادی منتظر بمانند.

بهتر است Task ها براساس ماهیت و اولویت در صف‌ های جداگانه قرار بگیرند:

app.conf.task_routes = {
    "tasks.send_sms": {
        "queue": "notifications",
    },
    "tasks.send_email": {
        "queue": "notifications",
    },
    "tasks.generate_pdf": {
        "queue": "documents",
    },
    "tasks.convert_video": {
        "queue": "media",
    },
}

سپس برای هر صف Worker مناسب اجرا شود:

celery -A celery_app worker \
    -Q notifications \
    --concurrency=8 \
    --loglevel=INFO

celery -A celery_app worker \
    -Q media \
    --concurrency=2 \
    --loglevel=INFO

Celery امکان مسیریابی Task ها به صف‌ های مختلف را فراهم می‌ کند و توصیه می‌شود Route ها به‌جای قراردادن پراکنده نام Queue در کد، در تنظیمات مرکزی تعریف شوند.

12- افزایش بی‌ حساب Concurrency:

ممکن است تصور کنیم هرچه مقدار Concurrency بیشتر باشد، Task ها سریع‌ تر اجرا می‌ شوند:

celery -A celery_app worker \
    --concurrency=100 \
    --loglevel=INFO

اما تعداد بسیار زیاد پردازش‌ ها یا Thread ها می‌ تواند باعث افزایش مصرف حافظه، فشار روی CPU، افزایش Connection های پایگاه داده و اشباع سرویس‌ های خارجی شود.

مقدار مناسب Concurrency به عوامل مختلفی بستگی دارد:

تعداد هسته‌ های CPU

میزان حافظه

CPU-bound یا I/O-bound بودن Task ها

محدودیت Connection پایگاه داده

ظرفیت سرویس‌ های خارجی

اندازه و مدت اجرای Task ها

نوع Pool مورد استفاده

برای Task های CPU-bound معمولا افزایش تعداد پردازش‌ ها بسیار بیشتر از تعداد هسته‌ های CPU سود زیادی ندارد. در Task های I/O-bound ممکن است Concurrency بیشتر قابل استفاده باشد، اما همچنان باید محدودیت منابع جانبی را در نظر گرفت.

مقدار مناسب باید با اندازه‌ گیری و Load Test مشخص شود، نه براساس یک عدد ثابت برای تمام پروژه‌ ها.

13- ذخیره نتیجه تمام Task ها:

بسیاری از Task ها فقط باید اجرا شوند و نیازی نیست مقدار بازگشتی آن‌ ها ذخیره شود.

برای مثال:

@app.task
def send_notification(user_id):
    notification_service.send(user_id)

    return True

اگر برنامه هیچگاه این مقدار را بررسی نمی‌ کند، ذخیره آن در Result Backend فقط منابع بیشتری مصرف خواهد کرد.

راهکار مناسب:

@app.task(ignore_result=True)
def send_notification(user_id):
    notification_service.send(user_id)

Celery توصیه می‌ کند برای Task هایی که نتیجه آن‌ ها مورد نیاز نیست از ignore_result=True استفاده شود، زیرا ذخیره نتایج غیرضروری زمان و منابع مصرف می‌ کند.

البته باید میان مقدار بازگشتی Task و وضعیت شکست آن تفاوت قائل شد. ممکن است نیازی به خروجی Task نداشته باشیم، اما همچنان لازم باشد خطا ها از طریق Log، سیستم مانیتورینگ یا ابزارهایی مانند Sentry ثبت شوند.

14- در نظر نگرفتن انقضای نتیجه‌ ها:

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

می‌ توان برای نتیجه‌ ها زمان انقضا تعیین کرد:

app.conf.result_expires = 3600

در این مثال نتیجه‌ ها پس از یک ساعت منقضی می‌ شوند.

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

اگر نتیجه Task اهمیت تجاری یا دائمی دارد، بهتر است آن را در مدل‌ های اصلی برنامه ذخیره کنیم و Result Backend را به‌عنوان پایگاه داده دائمی کسب‌ و کار در نظر نگیریم.

برای مثال، وضعیت گزارش بهتر است در جدول Report ذخیره شود:

class Report(models.Model):
    class Status(models.TextChoices):
        PENDING = "pending"
        PROCESSING = "processing"
        COMPLETED = "completed"
        FAILED = "failed"

    status = models.CharField(
        max_length=20,
        choices=Status.choices,
    )

    file = models.FileField(
        upload_to="reports/",
        null=True,
        blank=True,
    )

در این حالت حتی اگر داده‌ های Result Backend پاک شوند، وضعیت اصلی گزارش از بین نمی‌ رود.

15- تنظیم‌ نکردن Timezone در Celery Beat:

اشتباه دیگر این است که زمان‌ بندی را براساس ساعت محلی تنظیم کنیم، اما Timezone برنامه را مشخص نکنیم.

Celery برای برنامه‌ های دوره‌ ای به‌ صورت پیش‌فرض از UTC استفاده می‌ کند، مگر اینکه Timezone دیگری در تنظیمات تعریف شود.

برای مثال:

app.conf.timezone = "Asia/Tehran"
app.conf.enable_utc = True

یا در پروژه Django:

CELERY_TIMEZONE = "Asia/Tehran"
CELERY_ENABLE_UTC = True

بهتر است Timezone برنامه، پایگاه داده و زمان‌ بندی‌ ها به‌صورت آگاهانه تعیین شوند. همچنین هنگام استفاده از eta بهتر است از datetimeهای دارای Timezone استفاده شود.

نمونه مناسب:

from datetime import timedelta

from django.utils import timezone

send_reminder.apply_async(
    args=[reminder_id],
    eta=timezone.now() + timedelta(hours=1),
)

16- استفاده از countdown و eta برای تعداد زیادی Task دور دست:

گزینه‌های countdown و eta برای اجرای Task در آینده مفید هستند:

send_reminder.apply_async(
    args=[reminder_id],
    countdown=300,
)

اما این گزینه‌ ها برای ذخیره تعداد بسیار زیادی Task که قرار است چند روز یا چند ماه بعد اجرا شوند، همیشه انتخاب مناسبی نیستند.

Task های دارای eta یا countdown می‌ توانند توسط Worker دریافت شوند و تا رسیدن زمان اجرا در حافظه Worker باقی بمانند. انباشته‌ شدن تعداد زیادی از این Task ها برای زمان‌ های دور ممکن است مصرف حافظه را افزایش دهد.

برای زمان‌ بندی‌ های بلند مدت یا تعداد زیاد عملیات بهتر است از روش‌ هایی مانند موارد زیر استفاده شود:

Celery Beat

django-celery-beat

جدول زمان‌ بندی در پایگاه داده

یک Scheduler اختصاصی

ایجاد Task نزدیک به زمان اجرا

17- نداشتن Log مناسب داخل Task:

استفاده از print() برای بررسی موقت در محیط توسعه قابل‌ قبول است، اما در پروژه واقعی بهتر است از Logging ساختار یافته استفاده کنیم.

نمونه:

from celery import shared_task
from celery.utils.log import get_task_logger


logger = get_task_logger(__name__)


@shared_task(bind=True)
def generate_report(self, report_id):
    logger.info(
        "Report generation started",
        extra={
            "task_id": self.request.id,
            "report_id": report_id,
        },
    )

    try:
        ...
    except Exception:
        logger.exception(
            "Report generation failed",
            extra={
                "task_id": self.request.id,
                "report_id": report_id,
            },
        )
        raise

Celery یک Logger مخصوص Task ها ارائه می‌ دهد که می‌ تواند اطلاعاتی مانند نام Task و شناسه آن را در Log ها قرار دهد.

البته نباید اطلاعات حساس را داخل Log ثبت کنیم.

18- نداشتن مانیتورینگ برای Worker و صف‌ ها:

اگر فقط Worker را اجرا کنیم و هیچ نظارتی روی آن نداشته باشیم، ممکن است دیر متوجه مشکلات زیر شویم:

Worker متوقف شده است.

صف به‌ شدت بزرگ شده است.

Task ها مرتبا Failure می‌ شوند.

زمان اجرای Task ها افزایش پیدا کرده است.

تعداد Retry ها غیرعادی شده است.

یکی از Worker ها به یک Queue متصل نیست.

Task های زمان‌ بندی‌ شده اجرا نمی‌ شوند.

Celery ابزارهایی برای مشاهده Worker های فعال، Task های ثبت‌ شده، Task های فعال، Task های رزرو شده و وضعیت اجرای آن‌ ها دارد. Flower نیز یک ابزار تحت وب برای مشاهده و مدیریت Worker ها و Taskهاست.

برای نمونه:

celery -A celery_app inspect active

celery -A celery_app inspect registered

celery -A celery_app inspect reserved

اجرای Flower:

celery -A celery_app flower

در محیط Production بهتر است علاوه بر Flower، برای معیارهایی مانند طول صف، مدت اجرای Task، نرخ شکست و تعداد Workerهای فعال Alert تعریف شود.

19- مدیریت‌ نکردن خطای ارسال پیام به Broker:

ممکن است خود Task صحیح باشد، اما Producer نتواند پیام را به Broker ارسال کند:

send_email.delay(user_id)

برای مثال Redis یا RabbitMQ ممکن است موقتا در دسترس نباشد. در این حالت خطا پیش از رسیدن پیام به Worker رخ می‌ دهد.

برای عملیات مهم باید این حالت مدیریت شود:

from kombu.exceptions import OperationalError

try:
    send_email.delay(user_id)
except OperationalError:
    logger.exception(
        "Could not send task to broker",
        extra={
            "user_id": user_id,
        },
    )

    # Register task for a later retry or notify monitoring system

Retry اجرای Task با Retry ارسال پیام متفاوت است. تنظیمات Retry داخل Task زمانی کاربرد دارند که Worker پیام را دریافت کرده و اجرای Task با خطا رو به‌ رو شده باشد. اگر Producer نتواند پیام را به Broker تحویل دهد، Task هنوز وارد صف نشده است.

Celery برای خطا های اتصال هنگام ارسال Task نیز ساز و کار Retry دارد، اما باید سیاست آن متناسب با اهمیت عملیات تنظیم شود.

20- استفاده از تنظیمات یکسان برای تمام Task ها:

یکی دیگر از اشتباهات رایجی که در این زمینه رخ می دهد استفاده از تنظیمات یکسان برای Task های متفاوت هست.

تمام Task ها رفتار یکسانی ندارند. برای مثال:

Task ارسال ایمیل I/O-bound است.

Task تبدیل تصویر ممکن است CPU-bound باشد.

Task پرداخت باید Idempotent و با Retry محدود باشد.

Task اعلان ممکن است به Result Backend نیاز نداشته باشد.

Task گزارش ممکن است Time Limit طولانی‌ تری بخواهد.

بنابراین استفاده از یک تنظیم مشترک برای همه Task ها معمولا بهینه نیست:

app.conf.task_time_limit = 60
app.conf.task_default_retry_delay = 30

می‌ توان تنظیمات را برای هر Task براساس نیاز آن مشخص کرد:

@app.task(
    ignore_result=True,
    autoretry_for=(RequestException,),
    retry_backoff=True,
    retry_kwargs={"max_retries": 5},
    soft_time_limit=30,
    time_limit=45,
)
def send_email(user_id):
    ...

@app.task(
    ignore_result=False,
    soft_time_limit=1800,
    time_limit=1900,
)
def generate_report(report_id):
    ...

در طراحی صحیح، تنظیمات Task باید از ماهیت عملیات، اهمیت آن و رفتار سرویس‌ های وابسته به دست آید.

جمع‌ بندی:

در خیلی از پروژه های نرم افزاری و سازمانی بزرگ و حتی پروژه های کوچک که شاید تصور نشود٬ نیاز نیست تمام پردازش ها و انجام کارها در یک لحظه و مقابل چشم کاربر انجام شود.

بعضی عملیات ها نیاز به زمان بیشتری دارند و برخی دیگر به سرویس های دیگر وابسته هستند و بعضی نیز می توانند بدون اینکه هیچ خللی در عملکرد برنامه ایجاد شود با تاخیر اجرا شوند.

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

به کمک این ابزار برنامه می تواند درخواست کاربر را سریع تر پاسخ دهد و انجام کارهای جانبی را به بخش های دیگر بسپارد. این جداسازی باعث می شود بخش های مختلف سیستم مسئولیت مشخص تری داشته باشند و با افزایش حجم درخواست ها٬ مدیریت و توسعه پروژه ساده تر شود.

البته استفاده از Celery تنها به تعریف چند Task و اجرای یک Worker محدود نمی‌ شود. برای رسیدن به نتیجه مناسب باید Task ها با دقت انتخاب شوند، امکان بروز خطا و اجرای مجدد در نظر گرفته شود و وضعیت پردازش‌ ها نیز قابل بررسی باشد. استفاده نادرست از صف وظایف ممکن است به‌جای حل مشکل، پیچیدگی بیشتری به پروژه اضافه کند.

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

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

اگر قصد دارید Celery را در یک محیط واقعی اجرا کنید، پلتفرم ابری رانفلر امکان راه‌ اندازی پروژه‌ های پایتونی و فریم‌ ورک‌ هایی مانند Django و Flask را در کنار سرویس‌ هایی مانند Redis و RabbitMQ فراهم می‌ کند. به این ترتیب می‌توانید برنامه اصلی، Worker های Celery و سرویس Broker را روی بستر ابری اجرا و مدیریت کنید.

دسترسی به ترمینال، مشاهده لحظه‌ ای Log ها و امکان اتصال پروژه به GitHub و GitLab نیز کمک می‌ کند فرایند استقرار، بررسی و بروزرسانی Task های Celery ساده‌ تر انجام شود. بنابراین پس از پیاده‌ سازی ساختار Celery در محیط توسعه، می‌ توانید از رانفلر برای اجرای آن در یک محیط ابری و مقیاس‌ پذیر استفاده کنید.

مقالات مرتبط:

چرا یادگیری هوش مصنوعی با پایتون یک ضرورت هست!

12 تکنیک حرفه ای برای افزایش سرعت پایتون!

نکات مهم قبل از خرید هاست پایتون

Node.js یا Python ؟ کدام برای پروژه شما مناسب ترند!

بررسی کامل امنیت در پایتون؛ تهدیدها و راهکار مقابله