Tic Tac Toe, часть 6: Flask и Celery/RabbitMQ

В качестве основы возьмем проект с Flask'ом. Попробуем подключить Celery/RabbitMQ к нашему проекту. Celery займется вычислением случайного числа.

Установка проекта

Клонируем проект на свой компьютер:

git clone https://github.com/nomhoi/tic-tac-toe-part6.git

Запускаем контейнеры:

cd tic-tac-toe-part6
docker-compose up -d

Выполняем сборку веб-приложения:

cd front
npm install
npm run-script build

Открываем броузер по адресу http://localhost.

Docker контейнеры

В сервис flask добавили установку пакета Celery в файл requirements.txt и смонтировали папку с исходником проекта Celery: Сервис nginx остался без изменений.

volumes: - ./flask:/code - ./celery/app/proj:/code/proj

Добавились новые сервисы celery и rabbit.

celery: container_name: celery build: context: celery/ dockerfile: Dockerfile volumes: - ./celery/app:/app depends_on: - rabbit networks: - backend rabbit: container_name: rabbit hostname: rabbit image: rabbitmq:3.7.15-alpine environment: - RABBITMQ_DEFAULT_USER=admin - RABBITMQ_DEFAULT_PASS=CT2gNABH8eJ9yVh ports: - "5672:5672" networks: - backend

Сервис celery

Кто не знаком с Celery, имеет смысл тут-же пройтись по этому туториалу: Сервис celery выполнен на базе этого туториала.

$ docker exec -it celery python
Python 3.7.3 (default, May 11 2019, 02:00:41)
[GCC 8.3.0] on linux
Type "help", "copyright", "credits" or "license" for more information.
>>> from proj.tasks import add
>>> add.delay(2, 2)
<AsyncResult: 43662174-657f-4dd3-ab1a-22f5950c8794>
>>>

В задачи Celery в файл tasks.py добавлена наша задача getnumber: Как видим, наш проект в Celery оформлен в виде пакета proj.

@app.task
def getnumber(): return randrange(9)

Сервис flask

Исходный код этого проекта теперь присутствует в двух сервисах flask и celery. Напомню, что в этот сервис мы добавили пакет Celery и смонтировали папку с проектом proj.

from flask import Flask, jsonify
from proj.tasks import getnumber
from proj.celery import app as celery app = Flask(__name__) @app.route('/number')
def number(): task = getnumber.delay() return task.id @app.route('/result/<task_id>')
def result(task_id): task = getnumber.AsyncResult(task_id) result = task.get(timeout = 3) response = return jsonify(response)

В обработчике result мы получаем результат выполненной задачи по идентификатору и возвращаем ответ в JSON формате фронтенду. В обработчике number мы вызываем задачу getnumber, которая выполняется в воркере celery и возвращаем идентификатор задачи.

Фронтенд

После этого в функции getResult периодически отправляем запрос бэкенду на получение результата по адресу /result/<task_id>. В диспетчере нашей игры по нажатию кнопки Get random number сначала отправляем запрос бэкенду по адресу /number и получаем от него идентификатор задачи Celery.

async function getResult(task_id) { var i = 1; var timerId = setTimeout(async function go() { console.log("Result request: " + i); console.log("Task Id: " + task_id) const res = await fetch(`result/` + task_id); const response = await res.text(); if (res.ok) { let result = JSON.parse(response); console.log(result) if (result.state === 'SUCCESS') { let i = parseInt(result.number); if ($status === 1 || $history.state.squares[i]) { promise_number = result.number + ' - busy'; return; } promise_number = i; history.push(new Command($history.state, i)); return; } } if (i < 5) setTimeout(go, 500); i++; }, 500);
}

Изменили вывод результатов запросов к бэкенду:

{#await promise} <p>...подождите</p>
{:then taskid} <p>Task Id: {taskid}</p>
{:catch error} <p style="color: red">{error.message}</p>
{/await} {#await promise_number} <p>...подождите</p>
{:then number} <p>Number: {number}</p>
{:catch error} <p style="color: red">{error.message}</p>
{/await}

Домашнее задание

Попробуйте нашего интеллектуального агента живущего в celery сделать немного задумчивым, чтобы не сразу выдавал ответ. На самом деле сейчас результат приходит сразу после первого запроса.

TimeoutError: The operation timed out.". Время от времени начинает приходить ошибка от flask'a "500 (INTERNAL SERVER ERROR)", это в celery поднимается исключение "celery.exceptions. Пока не копал в чем дело, пожалуйста, посмотрите. Помогает только перезагрузка сервисов.

Можно добавить обработку ответов с FAILURE и PENDING. В getResult обрабатывается ответ только с состоянием SUCCESS, в остальных случаях выполняется повторный запрос. Формирование ответа в обработчике result также может зависеть от состояния задачи.

Вместо брокера сообщений RabbitMQ можно попробовать подключить Redis.

Можно попробовать выполнять запросы из приложения через брокеры сообщений.

Beast. А также попробовать выполнить это на базе примера с Boost.

Репозиторий на GitHub

https://github.com/nomhoi/tic-tac-toe-part6