Планировщик ритма сельдерея + колба + RabbitMQ

В этом руководстве мы собираемся настроить приложение Flask с планировщиком тактов сельдерея и RabbitMQ в качестве брокера сообщений. Приложение Flask будет увеличивать число на 10 каждые 5 секунд. Я делаю это в подсистеме Windows для Linux, но процесс должен быть почти таким же с другими дистрибутивами Linux.

Важная заметка

Бэкэнд celery amqp, который мы использовали в этом руководстве, был удален в Celery версии 5. На данный момент временное решение - просто установить старую версию сельдерея (pip install celery=4.4.6). Я работаю над редактированием этого руководства для другой серверной части.

Предварительные требования:

Во-первых, давайте сделаем папку, вы можете называть ее как угодно, я ее назвал test-app. В этой папке создайте базовое приложение для фляги. Вы можете использовать этот шаблон: https://github.com/delivey/bask или создать свой собственный.

После того, как мы настроили базовое приложение для фляги, давайте установим сельдерей. Чтобы установить сельдерей, просто введите pip install celery в терминал (если у вас есть Python). Если вы этого не сделаете, установите его отсюда: https://www.python.org/downloads/

Теперь давайте установим RabbitMQ. Это довольно сложно, но возможно. Это хорошее руководство по установке RabbitMQ на Ubuntu: https://www.vultr.com/docs/how-to-install-rabbitmq-on-ubuntu-16-04-47. Вам нужно следовать руководству только до шага 5. Для других дистрибутивов здесь есть руководства: https://www.rabbitmq.com/download.html

После того, как мы установили RabbitMQ, приступим к настройке. Вот полное руководство: https://docs.celeryproject.org/en/stable/getting-started/brokers/rabbitmq.html, но вам нужно только заполнить раздел Настройка RabbitMQ.

Приложение:

После этого мы можем приступить к настройке приложения. Давайте импортируем сельдерей и добавим базовую конфигурацию (import sqlite3 и render_templatearen не нужны, если вы создаете собственное приложение):

from flask import Flask, render_template
from celery import Celery
import sqlite3
app = Flask(__name__)
app.config[“CELERY_BROKER_URL”] = ‘amqp://myuser:mypassword@localhost:5672/myvhost’
app.config[“CELERY_RESULT_BACKEND”] = ‘amqp://myuser:mypassword@localhost:5672/myvhost'

Конечно, переименуйте myuser, mypassword и myvhost в значения, которые вы установили при настройке RabbitMQ.

Теперь давайте добавим планировщик ударов:

celery_beat_schedule = {
    "time_scheduler": {
        "task": "app.number_adding",
        "schedule": 5.0,
     }
}

Вам следует переименовать number_adding в свою собственную функцию, если вы создаете собственное приложение. И измените 5.0 на частоту выполнения вашей функции в секундах.

Давайте добавим базовую конфигурацию сельдерея:

celery = Celery(app.name)
celery.conf.update(
    result_backend=app.config["CELERY_RESULT_BACKEND"],
    broker_url=app.config["CELERY_BROKER_URL"],
    timezone="UTC",
    task_serializer="json",
    accept_content=["json"],
    result_serializer="json",
    beat_schedule=celery_beat_schedule,
)

Теперь предположим, что я хотел добавить функцию, увеличивающую число на 10. Я бы напечатал:

@celery.task
def number_adding():
    conn = sqlite3.connect('database.db')
    db = conn.cursor()
    currentNum = db.execute("SELECT number FROM testtb").fetchone()[0]
    db.execute("UPDATE testtb SET number=number+10")
    conn.commit()

@celery.task используется для функций, которые вы хотите выполнять с сельдереем.

Наконец, давайте добавим какой-нибудь маршрут, на котором будет отображаться номер:

@app.route("/")
def index():
    conn = sqlite3.connect('database.db') # connects to db
    db = conn.cursor() # creates a cursor to the db
    number = db.execute("SELECT number FROM testtb").fetchone()[0]
    return render_template("index.html", number=number)

Запуск приложения:

Примечание. Я рекомендую вводить sudo -i при запуске celery beat и celery worker в терминале, чтобы вам не пришлось устанавливать разрешения для файлов.

Во-первых, давайте запустим брокера RabbitMQ. Чтобы запустить его, просто введите: sudo rabbitmq-server. Если вы получили BOOT FAILED ошибку, это, вероятно, означает, что вы не останавливали сервер раньше. Для этого введите: sudo rabbitmqctl stop. Вот что вы должны увидеть в окне терминала после запуска сервера:

А теперь займемся сельдереем. Откройте другое окно терминала и введите: celery -A app.celery worker --loglevel=INFO --pidfile=''.

После того, как воркер будет запущен, мы можем запустить наш пул битов. Это позволит запланировать задачи для выполнения работником. Введите celery -A app.celery beat --loglevel=INFO --pidfile='' в окно терминала.

Наконец, откройте другое окно терминала и введите: flask run. Это запустит веб-приложение, и вы сможете получить к нему доступ на http://127.0.0.1:5000/

Наше приложение для фляги готово! Все, что осталось сделать, это добавить index.html там, где будет отображаться число. Если вы хотите просмотреть полный код этого приложения с комментариями, вы можете просмотреть его здесь: https://github.com/delivey/flask-celery-rabbitmq-code. Вот так выглядит сайт: