Django — очередь асинхронных задач Celery

Python

задний план

В разработке мы часто сталкиваемся с трудоемкими задачами, например:

Загрузите и проанализируйте файл Excel с 1w фрагментами данных и, наконец, сохраните его в базе данных.

В моей программе эта задача занимает около 6 с.Для пользователей ожидание 6 с — катастрофа.

Лучший способ справиться с этим:

  1. Получить запросы на эту задачу
  2. добавить эту задачу в очередь
  3. Сразу возвращает слова "Операция прошла успешно и обрабатывается в фоновом режиме"
  4. Используйте эту очередь в фоновом режиме и выполните эту задачу

Мы следуем этой идее и реализуем ее с помощью Celery.

выполнить

Среда, используемая в этой статье, выглядит следующим образом:

  • Python 3.6.7
  • RabbitMQ 3.8
  • Celery 4.3

Установите RabbitMQ с Docker

Celery использует серверную часть сообщений с такими опциями, как RabbitMQ, Redis и т. д. В этой статье используется RabbitMQ.

При этом для удобства установки RabbitMQ я устанавливал напрямую с помощью Docker:

docker run -d --name anno-rabbit -p 5672:5672 rabbitmq:3

После успешного запуска можно пройтиamqp://localhostДоступ к очереди сообщений.

Установите и настройте сельдерей

Celery — это инструмент, реализованный на Python, и установку можно выполнить напрямую через Pip:

pip install celery

Также предположим, что моя текущая папка проектаproj, проект называетсяmyproj, приложение называетсяmyapp

После завершения установки вproj/myproj/создать путьcelery.pyфайл для инициализации экземпляра Celery:

proj/myproj/celery.py

from __future__ import absolute_import, unicode_literals
import os
from celery import Celery, platforms

# set the default Django settings module for the 'celery' program.
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'myproj.settings')

app = Celery('myproj',
             broker='amqp://localhost//',
             backend='amqp://localhost//')

# Using a string here means the worker don't have to serialize
# the configuration object to child processes.s
# - namespace='CELERY' means all celery-related configuration keys
#   should have a `CELERY_` prefix.
app.config_from_object('django.conf:settings', namespace='CELERY')

# Load task modules from all registered Django app configs.
app.autodiscover_tasks()

затем вproj/myproj/__init__.pyДобавьте ссылку на объект Celery, чтобы обеспечить возможность инициализации Celery после запуска Django:

proj/myproj/__init__.py

from __future__ import absolute_import, unicode_literals

# This will make sure the app is always imported when
# Django starts so that shared_task will use this app.
from .celery import app as celery_app

__all__ = ('celery_app',)

Если нет других специальных конфигураций, это базовые конфигурации Celery.

Напишите трудоемкую задачу

Чтобы имитировать трудоемкую задачу, мы напрямую создаем метод, который «спит» 10 секунд, и устанавливаем его как задачу Celery:

proj/myapp/tasks.py

import time
from myproj.celery import app as celery_app

@celery_app.task
def waste_time():
    time.sleep(10)
    return "Run function 'waste_time' finished."

Запустить рабочий сельдерей

После того, как Celery настроен и задача успешно создана, мы начинаем сШаблоны для асинхронных задачЗапустите сельдерей:

celery -A myproj worker -l info

Заметьте, я подчеркнулАсинхронный режим, потому что Celery помимо асинхронных задач поддерживает временные задачи, поэтому его нужно указывать при запуске.

Также обратите внимание, что после запуска Celerywaste_time) изменения должны перезапустить Celery, чтобы они вступили в силу.

вызов задачи

В коде логики обработки запросов вызовите созданную выше задачу:

proj/myapp/views.py

from django.http import JsonResponse
from django.views.decorators.http import require_http_methods
from .tasks import waste_time

@require_http_methods(["POST"])
def upload_files(request):
    waste_time.delay()
    # Status code 202: Accepted, 表示异步任务已接受，可能还在处理中
    return JsonResponse({"results": "操作成功，正在上传，请稍候..."}, status=202)

перечислитьwaste_time.delay()После метода,waste_timeБудет добавлен в очередь задач в ожидании бездействующих вызовов Celery Worker.

Эффект

Когда мы отправим запрос, этот интерфейс вернется напрямую{"results": "操作成功，正在上传，请稍候..."}Вместо того, чтобы застревать на десять секунд, пользовательский опыт стал намного лучше.

Суммировать

Использование Celery для обработки таких асинхронных задач — распространенный метод в Python.Хотя фактическое время выполнения остается прежним или даже увеличивается (например, рабочий процесс занят, вызывая задержку обработки), это более приемлемо для пользователя. загрузить большой файл, вы можете продолжить обработку других транзакций, не дожидаясь на странице.

Есть и другие способы использования Celery, которые не рассматриваются в этой статье, и его документация уже очень подробная, и вы можете обратиться к ней напрямую, если вам это нужно.

Ссылаться на

Категории