Django 作为后端Web开发框架，有时候我们需要用到定时任务来或者固定频次的任务来执行某段代码，这时我们就要用到Celery了。Django中有一个中间件：Django-celery
环境：
- Python 3.6
- Django为小于1.8版本
- Celery为3.1版本
第一步安装：django-celery
|
1
|
pip install django-celery
|
第二步：配置celery和任务
创建测试django环境：
|
1
2
|
django-admin.py createproject test
django-admin.py startapp demo
|
创建好的项目布局如下：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
|
- proj/
- manage.py
- proj/
- __init__.py
- celery.py
- settings.py
- urls.py
- demo/
- migrations
- __init__.py
- admin.py
- apps.py
- models.py
- tasks.py
- tests.py
- views.py
|
2.1 配置celery.py文件
需要替换的内容，我都在对应的行后提示了，剩下的内容默认就好
创建test/test/celery.py文件，内容如下：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
|
from __future__ import absolute_import, unicode_literals
import os
from celery import Celery
# set the default Django settings module for the 'celery' program.
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'proj.settings') # “proj.settings”替换为你的项目信息：test.settings
app = Celery('proj') # 这里的proj替换为你的项目名称:test
# Using a string here means the worker doesn't have to serialize
# the configuration object to child processes.
# - 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()
@app.task(bind=True)
def debug_task(self):
print('Request: {0!r}'.format(self.request))
|
2.2 配置项目的init.py中配置celery内容
打开test/test/__init_.py文件，添加内容：
|
1
2
3
4
5
6
7
|
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',)
|
2.3 在task.py中添加计划任务
编辑test/demo/task.py文件，添加计划任务，内容如下：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
|
# Create your tasks here
from __future__ import absolute_import, unicode_literals
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)
|
第三步：任务执行
运行django项目： python manage.py runserver
3.1 后台添加计划任务
访问“http://localhost:8000/admin/”，在celery的管理页面里，选择Periodic tasks，进行任务添加。选择对应的任务，设置定时或者周期时间
3.2 启动定时的celery服务
注意：celery依赖redis服务，需要提前运行redis服务：`redis-server`
|
1
2
3
|
# 以下两个命令在不同的shell窗口里执行，需要在django的目录下
python manager.py celery beat -l info #接收定时任务的命令
python manager.py celery worker -l info #执行定时任务的命令，此shell窗口会看到任务的输入信息
|
3.3 启动单次的celery服务
注意：celery依赖redis服务，需要提前运行redis服务：`redis-server`
|
1
2
3
4
5
6
7
|
python manager.py shell # 进到django的shell里
from demo.task import mul, xsum # 导入task任务
a = mul()
b = xsum()
# 执行a， b会输出信息
a(1,2)
b(1)
|
PS：django-crontab实现Django定时任务
django-crontab安装：
|
1
|
pip install django-crontab
|
django-crontab加入：只需要将django-crontab加入到settings.py的INSTALLED_APPS即可。如下代码：
|
1
2
3
4
5
6
7
|
INSTALLED_APPS = (
'django_crontab',
...
)
|
django-crontab配置：settings.py中加入django-crontab的命令即可：
|
1
2
3
4
5
|
CRONJOBS = [
('47 11 * * *', 'django.core.management.call_command', ['closepoll']，{},'>> /var/run.log'),
]
|
格式：
参数1：定时 例如47 11 * * * 表示每天的11时47分执行
参数2：方法的python模块路径，如果执行django-admin命令，则写django.core.management.call_command
参数3：方法的位置参数列表（默认值：[]），如果执行django-admin命令，则填写所需执行的命令，例如我们在polls中已经定义过的closepoll
参数4：方法的关键字参数的dict（默认值：{}）
参数5：执行log存放位置（即重定向到文件，默认：''）
django-crontab任务加载：
django-crontab任务加载比较简单，只需要运行 python manage.py crontab add 即可
查看已经激活的任务使用 python manage.py crontab show
删除已经有的任务使用 python manage.py crontab remove
如果你修改了任务记得一定要使用 python manage.py crontab add 这个会更新定时任务
以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持服务器之家。
原文链接：http://www.bigyoung.cn/1096.html