前期准备
1.beat插件安装
pip3 install django-celery-beat
2.注册APP
INSTALLED_APPS = [
....
'django_celery_beat',
]
3.数据库变更
python3 manage.py migrate django_celery_beat
配置工作
目录结构请参考://www.jb51.net/article/200659.htm
1.配置celerypro.py
from __future__ import absolute_importimport osfrom celery import Celeryfrom django.conf import settingsfrom django.utils import timezone# set the default Django settings module for the 'celery' program.# 为celery设置环境变量os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'voice_quality_assurance_configure.settings')# 创建celery appapp = Celery('voice_quality_assurance_configure')# Using a string here means the worker will not have to# pickle the object when using Windows.# 从单独的配置模块中加载配置app.config_from_object('voice_quality_assurance_configure.celeryconfig')# 设置app自动加载任务app.autodiscover_tasks(lambda: settings.INSTALLED_APPS)# 解决时区问题,定时任务启动就循环输出app.now = timezone.now2.配置celeryconfig.py
3.分别启动woker和beta
项目根目录终端执行(voice_quality_assurance_configure为项目名称,简单来说,和manage.py文件同级)
celery -A voice_quality_assurance_configure beat -l info --scheduler django_celery_beat.schedulers:DatabaseScheduler #
启动beta 调度器使用数据库
celery worker -A voice_quality_assurance_configure --loglevel=info -n worker1 #启动celery worker
4.创建周期性任务
from datetime import datetime, timedeltaimport jsonimport os,djangoos.environ.setdefault("DJANGO_SETTINGS_MODULE", "voice_quality_assurance_configure.settings")# project_name 项目名称django.setup()from django_celery_beat.models import PeriodicTask, IntervalScheduleschedule, created = IntervalSchedule.objects.get_or_create(every=10,period=IntervalSchedule.SECONDS,)# 带参数的创建方法,如下:PeriodicTask.objects.create( interval=schedule, # 上面创建10秒的间隔 interval 对象 name='test_task', # 设置任务的name值 task='mission.tasks.my_task', # 指定需要周期性执行的任务 args=json.dumps([10, 2, 76]), expires=datetime.utcnow() + timedelta(seconds=30))详解创建周期性任务的方法
创建基于interval的周期性任务
第一步创建间隔对象
schedule, created = IntervalSchedule.objects.get_or_create( every=10, period=IntervalSchedule.SECONDS,)IntervalSchedule.DAYS 固定间隔天数
IntervalSchedule.HOURS 固定间隔小时数
IntervalSchedule.MINUTES 固定间隔分钟数
IntervalSchedule.SECONDS 固定间隔秒数
IntervalSchedule.MICROSECONDS 固定间隔微秒
第二步创建任务
无参数的创建方法:
PeriodicTask.objects.create( interval=schedule, # we created this above. name='test_task', # simply describes this periodic task. task='app名.tasks.任务函数名', # name of task.)有参数的创建方法:
创建基于 crontab 的周期性任务
from django_celery_beat.models import CrontabSchedule, PeriodicTaskschedule, _ = CrontabSchedule.objects.get_or_create( minute='30', hour='*', day_of_week='*', day_of_month='*', month_of_year='*', timezone=pytz.timezone('Canada/Pacific'))以上就是本文的全部内容,希望对大家的学习有所帮助,也希望大家多多支持。