参数
- scheduler: 指定调度器
- trigger: 任务执行的方式，共有三种：'date'、'interval'、'cron'。
- 'date' + 'run_date' 的参数组合, 能实现单次任务。 例子：2019-07-07 22:49:00 执行任务
@register_job(scheduler, 'date', id='test', run_date='2019-07-07 22:49:00') 注：在亲测时，执行完任务会报错，原因时执行完任务后会去mysql中删除djangojob表中的任务。但是djangojobexecution表记录着执行结果，有外键关联着djangojob表，所以删除时显示有外键约束错误。但是任务会正常执行，执行之后也会正常删除。 - 'interval' + 'hours' + 'minutes' + ..... 的参数组合，能实现间隔性任务。 例子：每隔3个半小时执行任务 还有seconds,days参数可以选择 注：如果任务需要执行10秒，而间隔设置为1秒，它是不会给你开10个线程同时去执行10个任务的。它会错过其他任务直到当前任务完成。
- @register_job(scheduler, 'interval', id='test', hours=3, minutes=30)
- 'cron' + 'hour' + 'minute'+...的参数组合，能实现cron类的任务。 例子：每天的8点半执行任务 还有day,second,month等参数可以选择。
- @register_job(scheduler, 'cron', id='test', hour=8, minute=30)
- id: 任务的名字，不传的话会自动生成。不过为了之后对任务进行暂停、开启、删除等操作，建议给一个名字。并且是唯一的，如果多个任务取一个名字，之前的任务就会被覆盖。
- args: list类型。执行代码所需要的参数。
- next_run_time：datetime类型。开始执行时间。如果你现在创建一个定时任务，想3天后凌晨三点半自动给你女朋友发微信，那就需要这个参数了。
还有些其他的参数感兴趣的同学可以查看源代码来了解。
def add_job(self, func, trigger=None, args=None, kwargs=None, id=None, name=None,
misfire_grace_time=undefined, coalesce=undefined, max_instances=undefined,
next_run_time=undefined, jobstore='default', executor='default',
replace_existing=False, **trigger_args):
"""
add_job(func, trigger=None, args=None, kwargs=None, id=None, \
name=None, misfire_grace_time=undefined, coalesce=undefined, \
max_instances=undefined, next_run_time=undefined, \
jobstore='default', executor='default', \
replace_existing=False, **trigger_args)
Adds the given job to the job list and wakes up the scheduler if it's already running.
Any option that defaults to ``undefined`` will be replaced with the corresponding default
value when the job is scheduled (which happens when the scheduler is started, or
immediately if the scheduler is already running).
The ``func`` argument can be given either as a callable object or a textual reference in
the ``package.module:some.object`` format, where the first half (separated by ``:``) is an
importable module and the second half is a reference to the callable object, relative to
the module.
The ``trigger`` argument can either be:
#. the alias name of the trigger (e.g. ``date``, ``interval`` or ``cron``), in which case
any extra keyword arguments to this method are passed on to the trigger's constructor
#. an instance of a trigger class
:param func: callable (or a textual reference to one) to run at the given time
:param str|apscheduler.triggers.base.BaseTrigger trigger: trigger that determines when
``func`` is called
:param list|tuple args: list of positional arguments to call func with
:param dict kwargs: dict of keyword arguments to call func with
:param str|unicode id: explicit identifier for the job (for modifying it later)
:param str|unicode name: textual description of the job
:param int misfire_grace_time: seconds after the designated runtime that the job is still
allowed to be run (or ``None`` to allow the job to run no matter how late it is)
:param bool coalesce: run once instead of many times if the scheduler determines that the
job should be run more than once in succession
:param int max_instances: maximum number of concurrently running instances allowed for this
job
:param datetime next_run_time: when to first run the job, regardless of the trigger (pass
``None`` to add the job as paused)
:param str|unicode jobstore: alias of the job store to store the job in
:param str|unicode executor: alias of the executor to run the job with
:param bool replace_existing: ``True`` to replace an existing job with the same ``id``
(but retain the number of runs from the existing one)
:rtype: Job
"""
job_kwargs = {
'trigger': self._create_trigger(trigger, trigger_args),
'executor': executor,
'func': func,
'args': tuple(args) if args is not None else (),
'kwargs': dict(kwargs) if kwargs is not None else {},
'id': id,
'name': name,
'misfire_grace_time': misfire_grace_time,
'coalesce': coalesce,
'max_instances': max_instances,
'next_run_time': next_run_time
}
job_kwargs = dict((key, value) for key, value in six.iteritems(job_kwargs) if
value is not undefined)
job = Job(self, **job_kwargs)
# Don't really add jobs to job stores before the scheduler is up and running
with self._jobstores_lock:
if self.state == STATE_STOPPED:
self._pending_jobs.append((job, jobstore, replace_existing))
self._logger.info('Adding job tentatively -- it will be properly scheduled when '
'the scheduler starts')
else:
self._real_add_job(job, jobstore, replace_existing)
return job
如何页面设定，定时任务时间
通常我们希望页面配置所以一般使用add_job进行定时任务创建和管理；
from apscheduler.schedulers.background import BackgroundScheduler
from django_apscheduler.jobstores import DjangoJobStore, register_job
from apscheduler.triggers.cron import CronTrigger
from apscheduler.triggers.interval import IntervalTrigger
from apscheduler.triggers.date import DateTrigger
scheduler = BackgroundScheduler() # 创建一个调度器对象
scheduler.add_jobstore(DjangoJobStore(), "default") # 添加一个作业
if not scheduler.state:
scheduler.start()
class ScheduleView(CustomViewSet):
lookup_field = 'id'
queryset = ScheduleTrigger.objects.all()
serializer_class = ScheduleSerializer
parser_classes = [JSONParser, FormParser]
def destroy(self, request, *args, **kwargs):
instance = self.get_object()
job_id = str(instance.id)
if scheduler.get_job(job_id=job_id) :
scheduler.remove_job(job_id=job_id)
self.perform_destroy(instance)
return CustomResponse(data=[],status=200,msg="删除ok",success=True)
def create(self, request, *args, **kwargs):
""" cron tigger need rewrite ,default */10"""
serializer = self.get_serializer(data=request.data)
serializer.is_valid(raise_exception=True)
self.perform_create(serializer)
try:
job_id =serializer.data.get('id')
if serializer.data.get("enable"):
if serializer.data.get('schedule_type') == '1':
scheduler.add_job(test,
trigger=DateTrigger(run_date=serializer.data.get('schedule_time')),
id=str(job_id),
max_instances=1, replace_existing=True, args=serializer.data.get('schedule_args'))
if serializer.data.get('schedule_type') == '2':
scheduler.add_job(test,
trigger=IntervalTrigger(seconds=int(serializer.data.get('schedule_time'))),
id=str(job_id),
max_instances=1, replace_existing=True, args=serializer.data.get('schedule_args'))
if serializer.data.get('schedule_type') == '3':
scheduler.add_job(test,
trigger=CronTrigger(second="*/10"),
id=str(job_id),
max_instances=1, replace_existing=True, args=serializer.data.get('schedule_args'))
register_job(scheduler)
except Exception:
pass
headers = self.get_success_headers(serializer.data)
return CustomResponse(data=serializer.data, code=200,
msg='ok', success=True,headers=headers)
def patch(self, request, *args, **kwargs):
""" cron tigger need rewrite ,default */10"""
instance = self.get_object()
job_id = instance.id
enable = request.data.get("enable")
if scheduler.get_job(job_id=str(job_id)):
if not enable:
scheduler.pause_job(job_id=str(job_id))
if enable:
scheduler.resume_job(job_id=str(job_id))
else:
if enable:
if instance.schedule_type =='1':
scheduler.add_job(test,
trigger=DateTrigger(run_date=instance.schedule_time),
id=str(job_id),
max_instances=1, replace_existing=True, args=instance.schedule_args)
if instance.schedule_type == '2':
scheduler.add_job(test,
trigger=IntervalTrigger(seconds=int(instance.schedule_time)),
id=str(job_id),
max_instances=1, replace_existing=True, args=instance.schedule_args)
if instance.schedule_type == '3':
scheduler.add_job(test,
trigger=CronTrigger(second="*/10"),
id=str(job_id),
max_instances=1, replace_existing=True, args=instance.schedule_args)
register_job(scheduler)
kwargs['partial'] = True
response = self.update(request, *args, **kwargs)
return CustomResponse(data=response.data, code=200,
msg='ok', success=True)
其他功能
django-apscheduler框架还提供了很多操作定时任务的函数。比如：
add_job 创建任务
- 删除任务
scheduler.remove_job(job_name) - 暂停任务
scheduler.pause_job(job_name) - 重新开启任务
scheduler.resume_job(job_name) - 修改任务
scheduler.modify_job(job_name) 注：修改任务只能修改参数，如果要修改执行时间的话，就把任务删了重新创建。
可以在页面上做一个这样的表格，再加上简单的前后端交互就可以让用户自行管理定时任务：
