Django-celery-beat动态添加周期性任务实现过程解析
前期准备
1.beat插件安装
pip3installdjango-celery-beat
2.注册APP
INSTALLED_APPS=[
....
'django_celery_beat',
]
3.数据库变更
python3manage.pymigratedjango_celery_beat
配置工作
目录结构请参考://www.nhooo.com/article/200659.htm
1.配置celerypro.py
from__future__importabsolute_import importos fromceleryimportCelery fromdjango.confimportsettings fromdjango.utilsimporttimezone #setthedefaultDjangosettingsmoduleforthe'celery'program. #为celery设置环境变量 os.environ.setdefault('DJANGO_SETTINGS_MODULE','voice_quality_assurance_configure.settings') #创建celeryapp app=Celery('voice_quality_assurance_configure') #Usingastringheremeanstheworkerwillnothaveto #pickletheobjectwhenusingWindows. #从单独的配置模块中加载配置 app.config_from_object('voice_quality_assurance_configure.celeryconfig') #设置app自动加载任务 app.autodiscover_tasks(lambda:settings.INSTALLED_APPS) #解决时区问题,定时任务启动就循环输出 app.now=timezone.now
2.配置celeryconfig.py
from__future__importabsolute_import fromkombuimportQueue fromdjango.confimportsettings #设置代理人broker CELERY_BROKER_URL='redis://127.0.0.1:6379/2' #指定Backend CELERY_RESULT_BACKEND='redis://127.0.0.1:6379/1' #指定时区,默认是UTC CELERY_TIMEZONE='Asia/Shanghai' #celery序列化与反序列化配置 CELERY_TASK_SERIALIZER='pickle' CELERY_RESULT_SERIALIZER='pickle' CELERY_ACCEPT_CONTENT=['pickle','json'] CELERY_IGNORE_RESULT=True #celery的启动工作数量设置 CELERY_WORKER_CONCURRENCY=10 #任务预取功能,会尽量多拿n个,以保证获取的通讯成本可以压缩。 CELERYD_PREFETCH_MULTIPLIER=20 #有些情况下可以防止死锁 CELERYD_FORCE_EXECV=True #celery的worker执行多少个任务后进行重启操作 CELERY_WORKER_MAX_TASKS_PER_CHILD=100 #禁用所有速度限制,如果网络资源有限,不建议开足马力。 CELERY_DISABLE_RATE_LIMITS=True #celerybeat配置(周期性任务设置) CELERY_ENABLE_UTC=False CELERY_TIMEZONE=settings.TIME_ZONE DJANGO_CELERY_BEAT_TZ_AWARE=False CELERY_BEAT_SCHEDULER='django_celery_beat.schedulers:DatabaseScheduler'
3.分别启动woker和beta
项目根目录终端执行(voice_quality_assurance_configure为项目名称,简单来说,和manage.py文件同级)
celery-Avoice_quality_assurance_configurebeat-linfo--schedulerdjango_celery_beat.schedulers:DatabaseScheduler#
启动beta调度器使用数据库
celeryworker-Avoice_quality_assurance_configure--loglevel=info-nworker1#启动celeryworker
4.创建周期性任务
fromdatetimeimportdatetime,timedelta importjson importos,django os.environ.setdefault("DJANGO_SETTINGS_MODULE","voice_quality_assurance_configure.settings")#project_name项目名称 django.setup() fromdjango_celery_beat.modelsimportPeriodicTask,IntervalSchedule schedule,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,#wecreatedthisabove. name='test_task',#simplydescribesthisperiodictask. task='app名.tasks.任务函数名',#nameoftask.)
有参数的创建方法:
PeriodicTask.objects.create( interval=schedule,#wecreatedthisabove. name='test'_task',#simplydescribesthisperiodictask. task='app名.tasks.任务函数名',#nameoftask. args=json.dumps(['arg1','arg2']), kwargs=json.dumps({'be_careful':True,}), expires=datetime.utcnow()+timedelta(seconds=30))
classMonitorDeviceTask(object): """ 设备创建,增加周期性任务 """ def__init__(self,device_obj): self.device_obj=device_obj self.periodic_task=PeriodicTask.objects.create( interval=schedule, name='test_task', task='mission.tasks.my_task', args=json.dumps([self.device_obj.ip]) ) defstarttask(self): """ 启动任务 """ self.periodic_task.enabled=True self.periodic_task.save() defstoptask(self): """ 停止任务 """ self.periodic_task.enabled=False self.periodic_task.save() defdeltask(self): """ 删除任务 """ self.periodic_task.delete() self.periodic_task.save()
创建基于crontab的周期性任务
fromdjango_celery_beat.modelsimportCrontabSchedule,PeriodicTask schedule,_=CrontabSchedule.objects.get_or_create( minute='30', hour='*', day_of_week='*', day_of_month='*', month_of_year='*', timezone=pytz.timezone('Canada/Pacific') )
以上就是本文的全部内容,希望对大家的学习有所帮助,也希望大家多多支持毛票票。