Django搭配Celery进行异步/定时任务（一）初步搭建

转自：http://blog.csdn.net/samed/article/details/50598371

以下需求场景很常见：

1. 用户点击页面按钮，请求后台进行一系列耗时非常高的操作，页面没有响应/一直Loading，用户体验非常不好。

2. 某些数据需要预先处理，每天凌晨的时候进行运算，大约半小时到1小时才能完成。

3. 进行外部系统的接口数据调用，接口要求在10秒内返回应答报文，但是Django获取数据之后要进行一定的处理，而此处理时间超过3分钟。

……

基于以上场景，就需要对站点进行异步任务 / 定时任务的处理了。

因为Django接受到请求之后，会阻塞进程，此过程未处理完毕，就无法响应反馈。

解决方案有很多，这里说说我觉得稍微靠谱的思路：

1. 先说针对定时任务，这种解决方案比较多，可以将程序书写成 manage common的方式进行，然后使用crontab调用命令即可

    或者使用celery，也采用crontab的方式进行调用。

2. 异步任务，目前比较靠谱的就是celery了。

综上，这里就使用celery的方式进行，manage common的方式改天尝试之后再发文。

python组件：

1. python2.7

2. Django1.7.1

3. celery( amqp-1.4.9 anyjson-0.3.3 billiard-3.3.0.22 celery-3.1.20 )    #使用pip install celery会自定把这些都装上，默认安装符合依赖的最新版。

4. django-celery-3.1.17    #还是用pip装，这个是用来支持结果写入数据库的。

步骤：

备注：以下内容，顶行书写是终端/命令行命令； #表示注释，实际运行请不要加入； python语法高亮均为文件中的内容； >>> 开头表示终端/命令行方式进入python执行的命令。

1. 新建django project：

django-admin startproject django_celery

cd django_celery

django-admin startapp demoapp

2. 新建celery配置：（详情看文末的解释）

django_celery/django_celery/celery.py：

from __future__ import absolute_import
import os
from celery import Celery
from django.conf import settings

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

app = Celery('django_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))

3. 设置导入celery实例：

修改django_celery/django_celery/__init__.py：

from __future__ import absolute_import

# 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

4. 新建demo tasks：

django_celery/demoapp/tasks.py：

from __future__ import absolute_import
from celery import shared_task

@shared_task
def add(x, y):
    return x + y

@shared_task
def mul(x, y):
    return x * y

@shared_task
def xsum(numbers):
    return sum(numbers)

5. 修改项目配置：

一、django_celery/django_celery/settings.py：

1）【可选】屏蔽不需要的app：admin、auth、contenttypes、sessions、messages、staticfiles。

2）增加app：'demoapp','djcelery','kombu.transport.django', ：

INSTALLED_APPS = (
    # 'django.contrib.admin',
    # 'django.contrib.auth',
    # 'django.contrib.contenttypes',
    # 'django.contrib.sessions',
    # 'django.contrib.messages',
    # 'django.contrib.staticfiles',
    'demoapp',
    'djcelery',
    'kombu.transport.django',
)

3）增加 broker 配置： 

BROKER_URL = 'django://localhost:8000//'

二、django_celery/django_celery/urls.py：

1）【可选】既然都去掉了admin了，那么就把admin的相关内容注释掉：

from django.conf.urls import patterns, include, url
# from django.contrib import admin

urlpatterns = patterns('',
    # Examples:
    # url(r'^$', 'django_celery.views.home', name='home'),
    # url(r'^blog/', include('blog.urls')),

    # url(r'^admin/', include(admin.site.urls)),
)

6. 准备就绪，检查当前目录结构：

django_celery
│  db.sqlite3
│  manage.py
│  manage.pyc
│
├─demoapp
│  │  admin.py
│  │  models.py
│  │  models.pyc
│  │  tasks.py
│  │  tasks.pyc
│  │  tests.py
│  │  views.py
│  │  __init__.py
│  │  __init__.pyc
│  │
│  └─migrations
│          __init__.py
│          __init__.pyc
│
└─django_celery
        celery.py
        celery.pyc
        settings.py
        settings.pyc
        urls.py
        wsgi.py
        wsgi.pyc
        __init__.py
        __init__.pyc

执行命令进行测试：

python manage.py syncdb                      #同步数据库，仅首次运行

python managepy runserver                    #运行站点，默认端口8000

【新终端/命令行】

python manage.py celery worker -l info        #开启celery worker 进程

【另一个新终端/命令行】

python manage.py shell

>>> from demoapp.tasks import *
>>> dir()
['__builtins__', 'absolute_import', 'add', 'mul', 'shared_task', 'xsum']
>>> mul(5,2)          #可以看到，直接执行命令是没有问题的，直接返回。
10
>>> mul.delay(5,2)       #采用delay方式执行，会发送任务到worker
<AsyncResult: 2922623d-e89b-48c1-a355-bc8d566d10e7>
>>> add.delay(2,3)
<AsyncResult: e07aea9b-86da-4d84-89c7-768aca076b53>

运行结果（在worker终端/命令行中查看）：

[2016-01-27 17:58:56,418: INFO/MainProcess] Received task: demoapp.tasks.mul[2922623d-e89b-48c1-a355-bc8d566d10e7]
[2016-01-27 17:58:56,437: INFO/MainProcess] Task demoapp.tasks.mul[2922623d-e89b-48c1-a355-bc8d566d10e7] succeeded in 0.0169999599457s: 10
[2016-01-27 18:00:46,618: INFO/MainProcess] Received task: demoapp.tasks.add[e07aea9b-86da-4d84-89c7-768aca076b53]
[2016-01-27 18:00:46,630: INFO/MainProcess] Task demoapp.tasks.add[e07aea9b-86da-4d84-89c7-768aca076b53] succeeded in 0.0100002288818s: 5

至此，demo结束。

【此部分内容仅研究使用，时间较忙无需关心】

1. 下面讲解一下celery.py文件的配置内容，为何要这么配置。

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'django_celery.settings')

设置这个环境变量是为了让 celery 命令能找到 Django 项目。这条语句必须出现在 Celery 实例创建之前。

app = Celery('django_celery')

这个 app 就是 Celery 实例。可以有很多 Celery 实例，但是当使用 Django 时，似乎没有必要。

app.config_from_object('django.conf:settings')

可以将 settings 对象作为参数传入，但是更好的方式是使用字符串，因为当使用 Windows 系统或者 execv 时 celery worker 不需要序列化 settings 对象。

app.autodiscover_tasks(lambda: settings.INSTALLED_APPS)

为了重用 Django APP，通常是在单独的 tasks.py 模块中定义所有任务。Celery 会自动发现这些模块，加上这一句后，Celery 会自动发现 Django APP 中定义的任务，前提是遵循如下 tasks.py 约定：

- app1/
    - tasks.py
    - models.py
- app2/
    - tasks.py
    - models.py

2. 关于broker：

这个是个什么东西，我还是不太理解，按照seeting的配置来说，我理解就是承载的站点。

BROKER_URL = 'django://localhost:8000//'

这里要注意我是使用了django自带的broker来作为celery broker，传说可以选的broker有：

RabbitMQ
Redis
database

更多的内容可以参看参考文献2。

--------------------------------------

参考文献：

1. Celery 3.1 中文文档：Django

2. First Steps with Celery