结合django的定时任务
应用场景&＃xff1a;
1、异步执行&＃xff0c;如耗时任务&＃xff1b;
2、定时任务
broker&＃xff1a;redis
执行任务&＃xff1a;worker
celery_pro目录&＃xff1a;
celeryf.py __init__.py config.py tasks1.py tasks2.py
celeryf.py
from __future__ import absolute_import,unicode_literals
from celery import Celeryapp &＃61; Celery(main&＃61;&＃39;mycelery&＃39;, #主模块名broker&＃61;&＃39;redis://:lyb&＃64;localhost:6379/0&＃39;, #消息代理backend&＃61;&＃39;redis://:lyb&＃64;localhost:6379/0&＃39;, #结果存储include&＃61;[&＃39;celery_pro.tasks1&＃39;,&＃39;celery_pro.tasks2&＃39;], #定义worker的文件&＃xff0c;不含.py
)app.config_from_object(&＃39;celery_pro.config&＃39;) #读取配置文件&＃xff0c;可以是导入的模块&＃xff0c;也可以是点分隔的文件字符串表示if __name__ &＃61;&＃61; &＃39;__main__&＃39;:app.start()
__init__.py #空
config.py #http://docs.celeryproject.org/en/latest/userguide/configuration.html
import os
from datetime import timedelta
from celery.schedules import crontabBASE_DIR &＃61; os.path.abspath(&＃39;&＃39;)
# PATH_TEST &＃61; &＃39;path_test&＃39;CELERY_ENABLE_UTC &＃61; False # 不是用UTC
CELERY_TIMEZONE &＃61; &＃39;Asia/Shanghai&＃39;CELERY_TASK_RESULT_EXPIRES &＃61; 60 * 60 * 24 # 任务结果的时效时间
# CELERYD_LOG_FILE &＃61; BASE_DIR &＃43; "/log/celery/celery.log" # log路径
# CELERYBEAT_LOG_FILE &＃61; BASE_DIR &＃43; "/log/celery/beat.log" # beat log路径
# CELERY_ACCEPT_CONTENT &＃61; [&＃39;pickle&＃39;, &＃39;json&＃39;, &＃39;msgpack&＃39;, &＃39;yaml&＃39;] # 允许接受的格式 一般采用msgpack(快)和json(跨平台)
# CELERY_RESULT_SERIALIZER &＃61; &＃39;json&＃39; # 读取任务结果格式
# CELERY_TASK_SERIALIZER &＃61; &＃39;msgpack&＃39; # 任务序列化反序列化采用msgpackCELERYBEAT_SCHEDULE &＃61; {&＃39;add-every-30-seconds&＃39;: {&＃39;task&＃39;: &＃39;celery_pro.tasks2.add_2&＃39;,&＃39;schedule&＃39;: 30.0,&＃39;args&＃39;: (16, 16)},&＃39;add-every-10-seconds&＃39;: {&＃39;task&＃39;: &＃39;celery_pro.tasks1.test&＃39;,&＃39;schedule&＃39;: 10.0,&＃39;args&＃39;: [&＃39;nihao&＃39;]},# Executes every Monday morning at 7:30 a.m.&＃39;add-every-monday-morning&＃39;: {&＃39;task&＃39;: &＃39;celery_pro.tasks2.add_2&＃39;,&＃39;schedule&＃39;: crontab(day_of_week&＃61;&＃39;monday&＃39;),&＃39;args&＃39;: (16, 16),},
}
tasks1.py
from __future__ import absolute_import,unicode_literals
from celery.schedules import crontab
from .celeryf import app#1、第一种形式
# &＃64;app.on_after_configure.connect
# def setup_periodic_tasks(sender, **kwargs):
# # Calls test(&＃39;hello&＃39;) every 10 seconds.
# sender.add_periodic_task(10.0, test.s(&＃39;hello&＃39;), name&＃61;&＃39;add every 10&＃39;)
#
# # Calls test(&＃39;world&＃39;) every 30 seconds
# sender.add_periodic_task(30.0, test.s(&＃39;world&＃39;), expires&＃61;10)
#
# # Executes every Monday morning at 7:30 a.m.
# sender.add_periodic_task(
# crontab(hour&＃61;7, minute&＃61;30, day_of_week&＃61;1),
# test.s(&＃39;Happy Mondays!&＃39;),
# )#2、第二种形式
# app.conf.beat_schedule &＃61; {# &＃39;add-every-30-seconds&＃39;: {# &＃39;task&＃39;: &＃39;celery_pro.tasks2.add_2&＃39;,# &＃39;schedule&＃39;: 30.0,# &＃39;args&＃39;: (16, 16)# },# &＃39;add-every-10-seconds&＃39;: {# &＃39;task&＃39;: &＃39;celery_pro.tasks1.test&＃39;,# &＃39;schedule&＃39;: 10.0,# &＃39;args&＃39;: [&＃39;nihao&＃39;]# },# # Executes every Monday morning at 7:30 a.m.# &＃39;add-every-monday-morning&＃39;: {# &＃39;task&＃39;: &＃39;celery_pro.tasks2.add_2&＃39;,# &＃39;schedule&＃39;: crontab(),# &＃39;args&＃39;: (16, 16),# },
# }&＃64;app.task
def test(arg):print(arg)
tasks2.py
from __future__ import unicode_literals,absolute_import
from .celery import app
import time&＃64;app.task
def add_2(x,y):print(&＃39;x &＃43; y :&＃39;)return x&＃43;y&＃64;app.task
def mul(x,y):time.sleep(20)return time.time()
启动worker&＃xff1a;
(venv) l&＃64;l:~/myfile/pycharm/py3$ python celery_pro/celeryf.py worker -l debug
启动celery beat即调度器&＃xff1a;
celery -A celery_pro.tasks1 beat -l debug
京公网安备 11010802041100号