Django Celery異步任務隊列的實現(xiàn)

更新時間：2019年07月24日 09:36:30 作者：嚴北

這篇文章主要介紹了Django Celery異步任務隊列的實現(xiàn)，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧

背景

在開發(fā)中，我們常常會遇到一些耗時任務，舉個例子：

上傳并解析一個 1w 條數(shù)據(jù)的 Excel 文件，最后持久化至數(shù)據(jù)庫。

在我的程序中，這個任務耗時大約 6s，對于用戶來說，6s 的等待已經是個災難了。

比較好的處理方式是：

接收這個任務的請求
將這個任務添加到隊列中
立即返回「操作成功，正在后臺處理」的字樣
后臺消費這個隊列，執(zhí)行這個任務

我們按照這個思路，借助 Celery 進行實現(xiàn)。

實現(xiàn)

本文所使用的環(huán)境如下：

Python 3.6.7
RabbitMQ 3.8
Celery 4.3

使用 Docker 安裝 RabbitMQ

Celery 依賴一個消息后端，可選方案有 RabbitMQ, Redis 等，本文選用 RabbitMQ 。

同時為了安裝方便，RabbitMQ 我直接使用 Docker 安裝：

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

啟動成功后，即可通過 amqp://localhost 訪問該消息隊列。

安裝并配置 Celery

Celery 是 Python 實現(xiàn)的工具，安裝可以直接通過 Pip 完成：

pip install celery

同時假設當前我的項目文件夾為 proj ，項目名為 myproj ，應用名為 myapp

安裝完成后，在 proj/myproj/ 路徑下創(chuàng)建一個 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 對象的引用，確保 Django 啟動后能夠初始化 Celery：

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 的基本配置就是這些。

編寫一個耗時任務

為了模擬一個耗時任務，我們直接創(chuàng)建一個方法，使其「睡」10s ，并將其設置為 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 Worker

Celery 配置完成，并且任務創(chuàng)建成功后，我們以異步任務的模式啟動 Celery ：

celery -A myproj worker -l info

注意到我強調了異步模式，是因為 Celery 除了支持異步任務，還支持定時任務，因此啟動時候要指明。

同時要注意，Celery 一旦啟動，對 Task(此處為 waste_time) 的修改必須重啟 Celery 才會生效。

任務調用

在請求處理的邏輯代碼中，調用上面創(chuàng)建好的任務：

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 調用。

效果

當我們發(fā)送請求時，這個接口會直接返回 {"results": "操作成功，正在上傳，請稍候..."} 的響應內容而非卡住十秒，用戶體驗要好許多。

總結

用 Celery 處理這種異步任務是 Python 常用的方法，雖然實際執(zhí)行成功耗時不變甚至有所增加（如 Worker 繁忙導致處理滯后），但是對于用戶體驗來說更容易接受，點擊上傳大文件后可以繼續(xù)處理其他事務，而不需要在頁面等待。
Celery 還有更多用法本文未介紹到，其文檔已經非常詳盡，有需要可直接參考。

參考

http://docs.celeryproject.org/en/latest/django/first-steps-with-django.html

https://hub.docker.com/_/rabbitmq

以上就是本文的全部內容，希望對大家的學習有所幫助，也希望大家多多支持腳本之家。

您可能感興趣的文章: