وظایف ناهمزمان با استفاده از فلاسک، ردیس و کرفس

معرفی

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

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

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

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

Task Queue چیست؟

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

صف‌های وظیفه برای واگذاری کاری مفید هستند که در غیر این صورت برنامه‌ها را در حین انتظار برای پاسخ‌ها کند می‌کند. آنها همچنین می توانند برای رسیدگی به وظایف منابع فشرده در حالی که ماشین اصلی یا process با کاربر تعامل دارد.

به این ترتیب، تعامل با کاربر سازگار، به موقع و تحت تأثیر حجم کار قرار نمی گیرد.

کرفس چیست؟

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

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

به این ترتیب، بار روی ماشین اصلی کاهش می یابد و منابع بیشتری برای رسیدگی به درخواست های کاربر در هنگام ورود آنها در دسترس است.

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

نمونه هایی از چنین کارگزاران پیام عبارتند از ردیس و RabbitMQ.

چرا از کرفس استفاده کنیم؟

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

مزیت دیگر این است که Celery به راحتی در چندین چارچوب وب ادغام می شود و اکثر آنها کتابخانه هایی برای تسهیل ادغام دارند.

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

کرفس همچنین می تواند از انواع واسطه های پیام استفاده کند که به ما انعطاف پذیری می دهد. RabbitMQ توصیه می شود اما می تواند Redis و را نیز پشتیبانی کند ساقه لوبیا.

برنامه آزمایشی

ما یک برنامه Flask ایجاد خواهیم کرد که به کاربران امکان می دهد یادآورهایی را تنظیم کنند که در زمان مشخصی به ایمیل آنها تحویل داده شود.

ما همچنین عملکردی را برای سفارشی کردن مدت زمان قبل از فراخوانی پیام یا یادآوری و ارسال پیام به کاربر ارائه خواهیم داد.

برپایی

مانند هر پروژه دیگری، کار ما در یک محیط مجازی انجام می شود که با استفاده از آن ایجاد و مدیریت می کنیم پیپنف ابزار:

$ pipenv install --three
$ pipenv shell

برای این پروژه، ما باید بسته های Flask و Celery را نصب کنیم تا شروع کنیم:

$ pipenv install flask celery

ساختار فایل برنامه Flask ما به این صورت است:

.
├── Pipfile                    # manage our environment
├── Pipfile.lock
├── README.md
├── __init__.py
├── app.py                     # main Flask application implementation
├── config.py                  # to host the configuration
├── requirements.txt           # store our requirements
└── templates
    └── index.html             # the landing page

1 directory, 8 files

برای پروژه مبتنی بر کرفس، ما از Redis به عنوان واسطه پیام استفاده خواهیم کرد و می‌توانیم دستورالعمل‌های تنظیم آن را پیدا کنیم. روی آنها homepage.

پیاده سازی

بیایید با ایجاد برنامه Flask شروع کنیم که فرمی را ارائه می دهد که به کاربران امکان می دهد جزئیات پیامی را که در آینده ارسال می شود وارد کنند.

موارد زیر را به ما اضافه می کنیم app.py فایل:

from flask import Flask, flash, render_template, request, redirect, url_for

app = Flask(__name__)
app.config.from_object("config")
app.secret_key = app.config('SECRET_KEY')

@app.route('/', methods=('GET', 'POST'))
def index():
    if request.method == 'GET':
        return render_template('index.html')

    elif request.method == 'POST':
        email = request.form('email')
        first_name = request.form('first_name')
        last_name = request.form('last_name')
        message = request.form('message')
        duration = request.form('duration')
        duration_unit = request.form('duration_unit')

        flash(“Message scheduled”)
        return redirect(url_for('index'))


if __name__ == '__main__':
    app.run(debug=True)

این یک برنامه واقعا ساده با یک مسیر واحد برای رسیدگی به یک است GET و POST درخواست فرم پس از ارسال جزئیات، می‌توانیم داده‌ها را به تابعی تحویل دهیم که کار را زمان‌بندی می‌کند.

برای اینکه فایل برنامه اصلی خود را خالی کنیم، متغیرهای پیکربندی را در یک جداگانه قرار می دهیم config.py فایل و کانفیگ را از فایل بارگذاری کنید:

app.config.from_object("config")

ما config.py فایل در همان پوشه ای خواهد بود app.py فایل و حاوی برخی از تنظیمات اولیه است:

SECRET_KEY = 'very_very_secure_and_secret'

در حال حاضر، اجازه دهید فرود را اجرا کنیم page مانند index.html:

{% for message in get_flashed_messages() %}
  <p style="color: red;">{{ message }}</p>
{% endfor %}

<form method="POST">
    First Name: <input id="first_name" name="first_name" type="text">
    Last Name: <input id="last_name" name="last_name" type="text">
    Email: <input id="email" name="email" type="email">
    Message: <textarea id="textarea" name="message"></textarea>
    Duration: <input id="duration" name="duration" placeholder="Enter duration as a number. for example: 3" type="text">

   <select name="duration_unit">
      <option value="" disabled selected>Choose the duration</option>
      <option value="1">Minutes</option>
      <option value="2">Hours</option>
      <option value="3">Days</option>
   </select>

   <button type="submit" name="action">Submit </button>
</form>

استایل و قالب‌بندی برای اختصار کوتاه شده است، در صورت تمایل HTML خود را قالب‌بندی/سبک دهید.

اکنون می توانیم برنامه خود را شروع کنیم:

ارسال ایمیل با استفاده از Flask-Mail

به منظور ارسال ایمیل از برنامه Flask ما، از Flask-Mail کتابخانه ای که به صورت زیر به پروژه خود اضافه می کنیم:

$ pipenv install flask-mail

با برنامه Flask و فرم موجود، اکنون می توانیم Flask-Mail را در خود ادغام کنیم app.py:

from flask_mail import Mail, Message

app = Flask(__name__)
app.config.from_object("config")
app.secret_key = app.config('SECRET_KEY')


mail = Mail(app)

def send_mail(data):
    """ Function to send emails.
    """
    with app.app_context():
        msg = Message("Ping!",
                    sender="admin.ping",
                    recipients=(data('email')))
        msg.body = data('message')
        mail.send(msg)

کارکرد send_main(data) پیام ارسالی و گیرنده ایمیل را دریافت می کند و پس از گذشت زمان مشخص شده برای ارسال ایمیل به کاربر فراخوانی می شود.

همچنین باید متغیرهای زیر را به خود اضافه کنیم config.py برای اینکه Flask-Mail کار کند:

MAIL_SERVER = 'smtp.googlemail.com'
MAIL_PORT = 587
MAIL_USE_TLS = True
MAIL_USERNAME = 'mail-username'
MAIL_PASSWORD = 'mail-password'

ادغام کرفس

با برنامه Flask ما آماده و مجهز به قابلیت ارسال ایمیل، اکنون می‌توانیم Celery را به منظور برنامه‌ریزی ایمیل‌ها برای ارسال در آینده یکپارچه کنیم.

ما app.py دوباره اصلاح خواهد شد:

from celery import Celery




client = Celery(app.name, broker=app.config('CELERY_BROKER_URL'))
client.conf.update(app.config)


@client.task
def send_mail(data):
    

@app.route('/', methods=('GET', 'POST'))
def index():
    if request.method == 'GET':
        return render_template('index.html')

    elif request.method == 'POST':
        data = {}
        data('email') = request.form('email')
        data('first_name') = request.form('first_name')
        data('last_name') = request.form('last_name')
        data('message') = request.form('message')
        duration = int(request.form('duration'))
        duration_unit = request.form('duration_unit')

        if duration_unit == 'minutes':
            duration *= 60
        elif duration_unit == 'hours':
            duration *= 3600
        elif duration_unit == 'days':
            duration *= 86400

        send_mail.apply_async(args=(data), countdown=duration)
        flash(f"Email will be sent to {data('email')} in {request.form('duration')} {duration_unit}")

        return redirect(url_for('index'))

ما import celery و از آن برای مقداردهی اولیه مشتری Celery در برنامه Flask با پیوست کردن URL برای واسطه پیام استفاده کنید. در مورد ما، ما از Redis به عنوان کارگزار استفاده خواهیم کرد، بنابراین موارد زیر را به ما اضافه می کنیم config.py:

CELERY_BROKER_URL = 'redis://localhost:6379/0'
CELERY_RESULT_BACKEND = 'redis://localhost:6379/0'

برای اینکه ما را داشته باشیم send_mail() تابع اجرا شده به عنوان یک کار پس زمینه، ما اضافه می کنیم @client.task دکوراتور تا مشتری کرفس ما از آن آگاه شود.

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

ابتدا داده های ورودی را بسته بندی می کنیم send_mail() عملکرد در فرهنگ لغت سپس، تابع پستی خود را از طریق Celery Task Calling API با استفاده از تابع فراخوانی می کنیم apply_async، که آرگومان های مورد نیاز تابع ما را می گیرد.

یک اختیاری countdown پارامتر تنظیم شده است و یک تاخیر بین اجرای کد و انجام کار را تعیین می کند.

این مدت زمان بر حسب ثانیه است، به همین دلیل است که ما مدت زمان ارسال شده توسط کاربر را بسته به ثانیه به ثانیه تبدیل می کنیم روی واحد زمانی که انتخاب می کنند

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

آوردن همه چیز با هم

برای اجرای پروژه ما به دو ترمینال نیاز داریم، یکی برای شروع برنامه Flask ما و دیگری برای راه اندازی Celery worker که پیام ها را در پس زمینه ارسال می کند.

ابتدا اپلیکیشن Flask را اجرا کنید terminal:

$ python app.py

در دومی terminal، محیط مجازی را راه اندازی کنید و سپس Celery worker را راه اندازی کنید:

# start the virtualenv
$ pipenv shell
$ celery worker -A app.client --loglevel=info

اگر همه چیز خوب پیش برود، بازخورد زیر را در سایت دریافت خواهیم کرد terminal اجرای مشتری Celery:

حالا اجازه دهید به حرکت کنیم http://localhost:5000 و جزئیات را تکمیل کنید تا ایمیل بعد از 2 دقیقه ارسال شود.

در بالای فرم، پیامی ظاهر می شود که آدرس دریافت ایمیل و مدت زمان ارسال ایمیل را نشان می دهد. در کرفس ما terminal، همچنین می توانیم ورودی گزارش را ببینیم که نشان می دهد ایمیل ما برنامه ریزی شده است:

(2019-10-23 16:27:25,399: INFO/MainProcess) Received task: app.send_mail(d65025c8-a291-40d0-aea2-e816cb40cd78)  ETA:(2019-10-23 13:29:25.170622+00:00)

را ETA بخش ورودی نشان می دهد که ما send_email() تابع فراخوانی می شود و بنابراین زمانی که ایمیل ارسال می شود.

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

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

نظارت بر خوشه کرفس با استفاده از گل

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

در حال نصب Flower به همین راحتی است:

$ pipenv install flower

اوایل روی، ما جزئیات مشتری Celery خود را در ما مشخص کردیم app.py فایل. برای نظارت بر آن، باید آن مشتری را به Flower منتقل کنیم.

برای رسیدن به این هدف باید یک سوم را باز کنیم terminal پنجره، به محیط مجازی ما بروید و ابزار نظارت ما را راه اندازی کنید:

$ pipenv shell
$ flower -A app.client --port=5555

هنگام شروع Flower، مشتری Celery را با عبور از برنامه مشخص می کنیم (-A) آرگومان، و همچنین تعیین پورت مورد استفاده از طریق --port بحث و جدل.

با نظارت ما، اجازه دهید ایمیل دیگری را برای ارسال برنامه ریزی کنیم روی داشبورد، و سپس به http://localhost:5555، که در آن موارد زیر از ما استقبال می کنند:

روی این page، ما می توانیم لیستی از کارگران در خوشه کرفس خود را مشاهده کنیم که در حال حاضر فقط از دستگاه ما تشکیل شده است.

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

در این بخش می بینیم که دو ایمیل را برنامه ریزی کرده بودیم و یکی در زمان مقرر با موفقیت ارسال شده است. ایمیل ها قرار بود به ترتیب پس از 1 دقیقه و 5 دقیقه برای اهداف آزمایشی ارسال شوند.

همچنین می توانیم زمان دریافت متن و زمان اجرای آن را از این قسمت مشاهده کنیم.

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

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

نتیجه

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

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

کد منبع این پروژه، مثل همیشه، در دسترس روی GitHub.