快捷導(dǎo)航

Django Celery異步任務(wù)隊(duì)列的實(shí)現(xiàn)

更新時(shí)間：2019年07月24日 09:36:30 作者：嚴(yán)北

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

背景

在開發(fā)中，我們常常會(huì)遇到一些耗時(shí)任務(wù)，舉個(gè)例子：

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

在我的程序中，這個(gè)任務(wù)耗時(shí)大約 6s，對(duì)于用戶來說，6s 的等待已經(jīng)是個(gè)災(zāi)難了。

比較好的處理方式是：

接收這個(gè)任務(wù)的請(qǐng)求
將這個(gè)任務(wù)添加到隊(duì)列中
立即返回「操作成功，正在后臺(tái)處理」的字樣
后臺(tái)消費(fèi)這個(gè)隊(duì)列，執(zhí)行這個(gè)任務(wù)

我們按照這個(gè)思路，借助 Celery 進(jìn)行實(shí)現(xiàn)。

實(shí)現(xiàn)

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

Python 3.6.7
RabbitMQ 3.8
Celery 4.3

使用 Docker 安裝 RabbitMQ

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

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

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

啟動(dòng)成功后，即可通過 amqp://localhost 訪問該消息隊(duì)列。

安裝并配置 Celery

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

pip install celery

同時(shí)假設(shè)當(dāng)前我的項(xiàng)目文件夾為 proj ，項(xiàng)目名為 myproj ，應(yīng)用名為 myapp

安裝完成后，在 proj/myproj/ 路徑下創(chuàng)建一個(gè) celery.py 文件，用來初始化 Celery 實(shí)例：

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 中添加對(duì) Celery 對(duì)象的引用，確保 Django 啟動(dòng)后能夠初始化 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 的基本配置就是這些。

編寫一個(gè)耗時(shí)任務(wù)

為了模擬一個(gè)耗時(shí)任務(wù)，我們直接創(chuàng)建一個(gè)方法，使其「睡」10s ，并將其設(shè)置為 Celery 的任務(wù)：

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."

啟動(dòng) Celery Worker

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

celery -A myproj worker -l info

注意到我強(qiáng)調(diào)了異步模式，是因?yàn)?Celery 除了支持異步任務(wù)，還支持定時(shí)任務(wù)，因此啟動(dòng)時(shí)候要指明。

同時(shí)要注意，Celery 一旦啟動(dòng)，對(duì) Task(此處為 waste_time) 的修改必須重啟 Celery 才會(huì)生效。

任務(wù)調(diào)用

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

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, 表示異步任務(wù)已接受，可能還在處理中
  return JsonResponse({"results": "操作成功，正在上傳，請(qǐng)稍候..."}, status=202)

調(diào)用 waste_time.delay() 方法后， waste_time 會(huì)被加入到任務(wù)隊(duì)列中，等待空閑的 Celery Worker 調(diào)用。

效果

當(dāng)我們發(fā)送請(qǐng)求時(shí)，這個(gè)接口會(huì)直接返回 {"results": "操作成功，正在上傳，請(qǐng)稍候..."} 的響應(yīng)內(nèi)容而非卡住十秒，用戶體驗(yàn)要好許多。

總結(jié)

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

參考

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

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

以上就是本文的全部內(nèi)容，希望對(duì)大家的學(xué)習(xí)有所幫助，也希望大家多多支持腳本之家。

您可能感興趣的文章: