Django-celery-beat动态添加周期性任务实现过程解析
作者:-零 发布时间:2021-07-29 13:09:49
前期准备
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_import
import os
from celery import Celery
from django.conf import settings
from 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 app
app = 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.now
2.配置celeryconfig.py
from __future__ import absolute_import
from kombu import Queue
from django.conf import settings
# 设置代理人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
# celery beat配置(周期性任务设置)
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 -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, timedelta
import json
import os,django
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "voice_quality_assurance_configure.settings")# project_name 项目名称
django.setup()
from django_celery_beat.models import PeriodicTask, 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, # we created this above.
name='test_task', # simply describes this periodic task.
task='app名.tasks.任务函数名', # name of task.)
有参数的创建方法:
PeriodicTask.objects.create(
interval=schedule, # we created this above.
name='test'_task', # simply describes this periodic task.
task='app名.tasks.任务函数名', # name of task.
args=json.dumps(['arg1', 'arg2']),
kwargs=json.dumps({ 'be_careful': True, }),
expires=datetime.utcnow() + timedelta(seconds=30) )
class MonitorDeviceTask(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])
)
def starttask(self):
"""
启动任务
"""
self.periodic_task.enabled = True
self.periodic_task.save()
def stoptask(self):
"""
停止任务
"""
self.periodic_task.enabled = False
self.periodic_task.save()
def deltask(self):
"""
删除任务
"""
self.periodic_task.delete()
self.periodic_task.save()
创建基于 crontab 的周期性任务
from django_celery_beat.models import CrontabSchedule, PeriodicTask
schedule, _ = CrontabSchedule.objects.get_or_create(
minute='30',
hour='*',
day_of_week='*',
day_of_month='*',
month_of_year='*',
timezone=pytz.timezone('Canada/Pacific')
)
来源:https://www.cnblogs.com/-wenli/p/13746509.html
猜你喜欢
- 以下是我做美工的两年生活中一条条总结出来的经验,每一点都是我常用的,虽然不是什么大学问,但我觉得要互相学习才能提高,所以现在拿出来和大家一起
- 在工作中,作为一名开发者的你,也许偶尔需要从事维护数据库的工作。下面我们来介绍一下两个SQL服务器的维护技巧:轻松改变数据库拥有者、整理索引
- 原文链接:https://vien.tech/article/157前言本文将介绍ubuntu系统下如何定时执行shell脚本、python
- 以前我浏览博客的时候记得别人说过,BCELoss与CrossEntropyLoss都是用于分类问题。可以知道,BCELoss是Binary
- 交互设计师主导或发起的项目很难拿到结果——现状这个问题在很多的小公司都不存在。小公司养着、催着设计师,设计师不用去考虑能不能拿到结果,因为你
- 我们在使用ASP 内置的ADO组件进行数据库编程时,通常是在脚本的开头打开一个连接,并在脚本的最后关闭它,但是就较大脚本而言,在多数情况下连
- 滑动验证码介绍本篇涉及到的验证码为滑动验证码,不同于极验证,本验证码难度略低,需要的将滑块拖动到矩形区域右侧即可完成。这类验证码不常见了,官
- 在IE7还不支持counter 和increment 属性之前,我从来没有用过它们,也从来没有使用过:before 伪元素和content
- 大一上学期学习的内容之一,小黑屋比较好玩。1.导入函数库先导入random、time两个函数库的使用来达到随机生成人物、生成人物加载时间的目
- 本文实例讲述了PHP实现网页内容html标签补全和过滤的方法。分享给大家供大家参考,具体如下:如果你的网页内容的html标签显示不全,有些表
- 介绍据官方所述,mango Cache是对Guava Cache基于go的部分实现,同时mangoCache参考了Caffeine以及go-
- 随着信息技术的飞速发展,数据处理不仅在数量上要求越来越大,而且在质量上也要求越来越高。操作系统的稳定对数据库来说是十分紧要的,在数据库可操作
- 重复的数据可能有这样两种情况,第一种: 表中只有某些字段一样,第二种是两行记录完全一样。一、对于部分字段重复数据的删除 1.查询重复的数据
- 译文原文:http://blog.benhuoer.com/2009/04/10-simple-and-impressive-design-
- MySQL是关系型数据库中的明星,MongoDB是文档型数据库中的翘楚。下面通过一个设计实例对比一下二者:假设我们正在维护一个手机产品库,里
- GetObject 函数返回对文件中 Automation 对象的引用。GetObject([pathname] [, class])参数P
- 今天我们来学习一下 异常语法 中的另一个成员 —> finally ; 通过学习 finally ,可以帮助我们更
- python的版本及依赖的库的安装#版本python 3.7.1pip install pywin32==224pip install nu
- 我刚进入5gsns的时候,我真不知道怎么玩,我是通过白鸦的博客过去的,之前也没有怎么去玩过这类的网站。对于sns网站还算是陌生,不过还好网站
- YUI 中的 Grids CSS 主要有三个部分值得大家关注和学习:Yahoo!网站性能最佳体验的34条黄金守则—内容 1、布局的思想:使用