下面是“Django+Django-Celery+Celery的整合实战”的完整攻略及示例说明：

1. 安装Django

首先需要安装Django，可以通过pip命令进行安装：

pip install Django

2. 安装Django-Celery

接着需要安装Django-Celery，同样可以通过pip命令进行安装：

pip install django-celery

3. 安装Celery

再安装Celery，同样也可以通过pip进行安装：

pip install Celery

4. 配置settings.py

打开Django项目的settings.py文件，添加如下的Celery配置（具体应该根据实际需求进行配置）：

# Celery配置
from datetime import timedelta

# Broker设置，使用RabbitMQ作为消息队列 （可根据实际需要进行配置）
BROKER_URL = 'amqp://localhost'

# 配置结果存储路径 （可根据实际需要进行配置）
CELERY_RESULT_BACKEND = 'django-db'

# celery任务超时时间 （可根据实际需要进行配置）
CELERY_TASK_RESULT_EXPIRES = timedelta(days=1)

# 导入celery启动模块
CELERY_IMPORTS = ('app.tasks', )

# 设置celery的时区 （可根据实际需要进行配置）
CELERY_TIMEZONE = 'Asia/Shanghai'

5. 定义任务

根据需要，在app目录下创建一个tasks.py文件，定义需要进行的Celery任务：

from celery import shared_task

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

6. 启动Celery

启动Celery的Worker进程，可以使用以下命令：

celery -A myproject worker -l info

其中myproject需要替换成自己的项目名称。

示例说明一

下面是一个简单的任务示例，定义一个Celery任务，计算斐波那契数列中的第n项：

from celery import shared_task
from math import sqrt

@shared_task
def fib(n):
    if n < 2:
        return n
    return fib(n-1) + fib(n-2)

@shared_task
def sqrt_task(n):
    return sqrt(n)

在视图函数中调用：

from app.tasks import fib, sqrt_task

def test_task(request):
    result = fib.delay(10)    # 异步调用fib任务
    sqrt_result = sqrt_task.delay(10)    # 异步调用sqrt_task任务
    context = {
        'result': result.get(),
        'sqrt_result': sqrt_result.get()
    }
    return render(request, 'test_task.html', context)

在test_task.html模板中可以输出结果：

{{ result }}
{{ sqrt_result }}

示例说明二

下面是一个Celery任务的定时调度示例，每10秒钟打印一次Hello World：

from celery.schedules import crontab
from celery.task import periodic_task

@periodic_task(run_every=crontab(minute='*/10'))
def hello_world():
    print('Hello World!')

启动Celery的beat进程，可以使用以下命令：

celery -A proj beat -l info

这即可每10秒打印一次Hello World。

本文链接：http://task.lmcjl.com/news/18718.html