下面是“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。
本站文章如无特殊说明,均为本站原创,如若转载,请注明出处:Django+Django-Celery+Celery的整合实战 - Python技术站