Django集成Celery实现异步任务处理实战
本文通过一个加法计算的小项目，详细演示如何在Django框架中集成Celery以解决长时间后台处理阻塞HTTP请求的问题。内容涵盖环境配置、项目结构设计、Broker与Backend选择、代码编写及Worker启动步骤，旨在提升Web应用的用户体验。
前文已介绍Celery的基础知识，本文将通过一个具体小项目，详细阐述如何在Django框架中集成Celery进行开发。
本系列文章所基于的开发环境如下：
window 7 + python2.7 + pycharm5 + celery3.1.25 + django1.9.4
在Web应用中，当用户触发某个操作时，后台处理程序可能需要较长时间才能返回结果。为了避免阻塞HTTP请求并提升用户体验，我们需要解决这一等待问题。本例的设计方案是：提供两个网页界面，一个用于提交加法计算任务至后台执行；另一个用于查看各项计算任务的详细信息，包括任务ID和当前状态等。
效果如下图所示，访问地址为：http://127.0.0.1:8000/index/
点击“查看任务”链接，即可浏览所有任务的执行状态：其中“success”表示执行成功，“started”表示任务已开始执行。
二、项目结构
消息代理（Broker）选用RabbitMQ，结果存储后端（Backend）选用django-celery。
三、开始编程
1、安装django-celery：
$ pip install django-celery
2、新建Django工程项目test_celery，目录结构如下：
【说明】：tools为Django应用包，与test_celery同级，主要用于存放工具类函数及已定义的任务；
celery.py：包含Celery的配置信息；
3、编辑setting.py，添加Celery的Broker和Backend配置；
#celery setting
import djcelery
djcelery.setup_loader() #加载djcelery
BROKER_URL = 'pyamqp://guest@localhost//' #配置broker
BROKER_POOL_LIMIT = 0
CELERY_RESULT_BACKEND='djcelery.backends.database:DatabaseBackend' #配置backend
INSTALLED_APPS = [
'django.contrib.admin',
'django.contrib.auth',
'django.contrib.contenttypes',
'django.contrib.sessions',
'django.contrib.messages',
'django.contrib.staticfiles',
'djcelery', #django-celery必须添加
'test_celery',
]
DATABASES = { 'default': { 'ENGINE': 'django.db.backends.mysql', 'NAME': 'celery', 'USER':'username', 'PASSWORD':'password', 'HOST':'ip', 'PORT':3306, }}
from __future__ import absolute_import
import os
from celery import Celery
# set the default Django settings module for the 'celery' program.
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'test_celery.settings')
from django.conf import settings # noqa
app = Celery('test_celery')
# Using a string here means the worker will not have to
# pickle the object when using Windows.
app.config_from_object('django.conf:settings')
app.autodiscover_tasks(lambda: settings.INSTALLED_APPS)
@app.task(bind=True)
def debug_task(self):
print('Request: {0!r}'.format(self.request))
from __future__ import absolute_import from celery import shared_task import time @shared_task(track_started=True) def add(x, y): time.sleep(30) #模拟长时间执行 return x + y
from django.db import models #在数据库中生成test_celery_add表 class Add(models.Model): task_id = models.CharField(max_length=128) #存储taskid first = models.IntegerField() #存储第一个加数 second = models.IntegerField() #存储第二个加数 log_date = models.DateTimeField() #存储开始时间
python manage.py makemigrations test_celery python manage.py migrate
完成上述配置后，数据库中会自动生成以下数据表：
8、接下来编写前端HTML文件以及View视图文件，具体代码可自行编写或下载源码运行。
9、代码编写完成后，启动Worker，即可在网页上进行模拟操作测试。
进入test_celery项目目录（即包含manage.py文件的层级），执行以下命令以启动Worker：
celery worker -A test_celery -l info
项目源代码下载
免责声明：本文内容仅供参考，不代表本站观点。转载或引用请注明出处。
