本文主要介绍django和celery结合使用的案例。
celery 是一个异步任务的调度工具，可以完成一些异步任务和定时任务。
本文使用djcelery来完成django和celery的结合使用。
该案例在github中django_celery_demo (opens new window)
任务发布者(Producer)将任务丢到消息队列(Broker)中，任务消费者(worker)从消息代理中获取任务执行，然后将保存存储结果(backend)。
celery 有个定时功能，通过定时去将task丢到broker中，然后worker去执行任务。但是有个确定是，该定时任务必须硬编写到代码中，不可在程序运行中动态增加任务。使用djcelery可以将定时任务写入到数据库中，然后通过操作数据库操作定时任务。
访问接口，异步调用程序中task
安装**djcelery**
pip install django_celery
在settings中设置celery配置
代码: django_celery_demo/settings.py
import djcelery
djcelery.setup_loader() # 加载djcelery
# 允许的格式
CELERY_ACCEPT_CONTENT = ['pickle', 'json', 'yaml']
BROKER_URL = 'redis://localhost:6379/1' # redis作为中间件
BROKER_TRANSPORT = 'redis'
CELERYBEAT_SCHEDULER = 'djcelery.schedulers.DatabaseScheduler' # 定时任务使用数据库来操作
CELERY_RESULT_BACKEND = 'djcelery.backends.database:DatabaseBackend' # 结果存储到数据库中
# worker 并发数
CELERY_CONCURRENCY = 2
# 指定导入task任务
CELERY_IMPORTS = {
'tasks.tasks'
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
celery app 配置
代码: django_celery_demo/celery.py
import os
import django
from celery import Celery, shared_task
from celery.schedules import crontab
from celery.signals import task_success
from django.conf import settings
from django.utils import timezone
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'django_celery_demo.settings')
django.setup()
app = Celery('django_celery_demo')
app.config_from_object('django.conf:settings') # celery app 加载 settings中的配置
app.now = timezone.now # 设置时间时区和django一样
# 加载每个django app下的tasks.py中的task任务
app.autodiscover_tasks(lambda: settings.INSTALLED_APPS)
# 这个一个task
@app.task(bind=True)
def debug_task(self):
print('Request: {0!r}'.format(self.request))
# 异步执行这个task
debug_task.delay()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
创建djcelery中的表
会自动创建djcelery中的表。里面有保存定时记录、结果记录等等表。
python manage.py migrate
在app中创建**add**task
代码: demo/tasks.py
from celery import shared_task
@shared_task(name="add")
def add(a, b):
return int(a) + int(b)
1
2
3
4
5
创建view去异步执行该task
代码: demo/views.py
from django.http import HttpResponse
from demo.tasks import add as add_task
def add(request):
a = request.GET["a"]
b = request.GET["b"]
add_task.delay(a, b)
return HttpResponse("success")
1
2
3
4
5
6
7
8
9
url中配置view
from demo.views import add
urlpatterns = [
path('add', add),
]
1
2
3
4
5
运行celery worker
celery -A django_celery_demo worker -l info
运行项目
python manage.py runserver 0:8888
访问接口
http://127.0.0.1:8888/add?a=1&b=2 (opens new window)
结果: 返回success，在worker中可以看到add任务被调用，并且结果是3
定时调用异步任务
有两种定时任务方式，这里使用的是常见的crontab，与linux一毛一样，方便很多。
配置和案例1中一样。
硬编码中创建定时任务
每分钟调用一次add task
代码: django_celery_demo/celery.py
# 这个是硬编码的定时任务
app.conf.beat_schedule = {
'aa': {
'task': 'add',
'schedule': crontab(minute="*/1"),
'args': (2, 4)
},
}
1
2
3
4
5
6
7
8
开启celery beat
celery beat -A django_celery_demo -l info
这个服务会将数据库中的定时任务丢到broker 中
将不同的任务放到不同的队列中，放到不同的worker中。
图: 消息分发与任务调度的实现机制
default = Exchange('default', type="direct")
frequent = Exchange('frequent', type="direct")
CELERY_QUEUES = {
Queue('default', default, routing_key="default"),
Queue('frequent', frequent, routing_key="frequent")
}
app.conf.task_default_queue = 'default'
task_routes = {
'apps.periodic.tasks.oozie_workflow_task': {'queue': 'default'},
'apps.periodic.tasks.oozie_workflow_status': {'queue': 'custom'}
}
app.conf.beat_schedule = {
# 每分钟检查oozie运行中的任务状态
'oozie_workflow_status': {
'task': 'oozie_workflow_status',
'schedule': crontab(),
'args': (2, 1)
}
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
使用supervisor来管理worker进程。
安装
pip install supervisor
生成默认配置文件
echo_supervisord_conf > /etc/supervisor/supervisord.conf
supervisorctl
| 命令 | 描述 |
|---|---|
| status | |
| reread | 读取配置文件 |
| update | 加载最新的进程 |
| stop 进程名 | |
| start 进程名 | |
| reload | 重新加载配置 |
在supervisord.conf中添加下面的配置。
[include]
; files = relative/directory/*.ini
files = /home/jim/conf/supervisor/supervisord.conf.d/*.conf
1
2
3
创建配置文件/home/jim/conf/supervisor/supervisord.conf.d/celeryd_worker.conf，添加下面配置
[program:celeryworker]
command=celery -A datahub_poster worker -l info
directory=/home/hadoop/jim/projs/datahub_poster
stdout_logfile=/yun/jim/log/supervisor/celeryworker.log
;stderr_logfile=/yun/jim/log/supervisor/celeryworker_err.log
redirect_stderr=true
autorestart=true
autostart=true
numprocs=1
startsecs=10
stopwaitsecs = 600
priority=15
[program:celerybeat]
command=celery -A datahub_poster beat -l info
directory=/home/hadoop/jim/projs/datahub_poster
stdout_logfile=/yun/jim/log/supervisor/celerybeat.log
;stderr_logfile=/yun/jim/log/supervisor/celerybeat_err.log
redirect_stderr=true
autorestart=true
autostart=true
numprocs=1
startsecs=10
stopwaitsecs = 600
priority=15
[program:celery_flower]
command=celery -A datahub_poster flower --port=5555
directory=/home/hadoop/jim/projs/datahub_poster
stdout_logfile=/yun/jim/log/supervisor/celery_flower.log
;stderr_logfile=/yun/jim/log/supervisor/celery_flower_err.log
redirect_stderr=true
autorestart=true
autostart=true
numprocs=1
startsecs=10
stopwaitsecs = 600
priority=15
[inet_http_server]
port=127.0.0.1:9001
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
使用配置文件启动supervisor
supervisord -c /etc/supervisor/supervisord.conf
http://localhost:9001 refused connection错误。解决办法：
在配置文件supervisord.conf中添加
[inet_http_server]
port=127.0.0.1:9001
1
2
然后再update或reload以下。
可以通过flower监控celery中的worker、task等等。
安装flower
pip install flower
运行
celery flower --broker=redis://localhost:6379/0
问题: 每次重启flower之后发现，以前的task运行记录清空。
解决: 启动flower时添加 --persistent=True，可以持久化task
flower会读取celery的时区配置，在项目中配置下面参数即可。
TIME_ZONE = 'Asia/Shanghai'
CELERY_TIMEZONE = TIME_ZONE
1
2
此内容由惯性聚合(RSS阅读器)自动聚合整理，仅供阅读参考。 原文来自 — 版权归原作者所有。