Django Apscheduler详解

XOENA • 2025-01-14 18:55 • 编程

一、什么是Django Apscheduler

Django Apscheduler是Django与Apscheduler进行结合的一个开源库，Apscheduler是Python中的一个定时任务调度器，可以让我们可以方便地进行定时任务的管理。

二、Django Apscheduler的安装及配置

1、安装

pip install django-apscheduler

2、在django的settings.py文件中配置相关信息


INSTALLED_APPS =[
    'django_apscheduler',
    # ...
]

SCHEDULER_AUTOSTART = True

SCHEDULER_RUN_NOW = True

SCHEDULER_TIMEZONE = "Asia/Shanghai"

其中，SCHEDULER_AUTOSTART表示Django启动时，任务调度器是否随之启动；SCHEDULER_RUN_NOW表示Django启动时，任务调度器是否跑一遍；SCHEDULER_TIMEZONE是任务调度器所使用的时区。

三、Django Apscheduler的使用方法

1、创建任务

在Django Apscheduler中，我们可以使用装饰器或继承django_apscheduler.models的BaseJob类来创建我们的任务。

装饰器创建方法：


from django_apscheduler.jobstores import DjangoJobStore, register_job

@register_job(DjangoJobStore(), 'interval', seconds=30)
def task_job():
    print('Django apscheduler task')

继承BaseJob类创建方法：


from django_apscheduler.models import DjangoJob
from django_apscheduler.jobstores import DjangoJobStore

class TaskJob(DjangoJob):
    owner = models.ForeignKey(settings.AUTH_USER_MODEL)

    @classmethod
    def get_owner(klass, id):
        try:
            return klass.objects.get(id).owner
        except klass.DoesNotExist:
            return None

    def run(self):
        print('Django apscheduler task')

job = TaskJob()
job.interval_seconds = 30
job.save()

2、任务调度器的各种配置

除了上述的SCHEDULER_AUTOSTART、SCHEDULER_RUN_NOW和SCHEDULER_TIMEZONE，Django Apscheduler还有各种配置可以设定，下面是部分配置：


SCHEDULER_JOBSTORES = {
    'default': DjangoJobStore(),
}

SCHEDULER_EXECUTORS = {
    'default': {'type': 'threadpool', 'max_workers': 20},
}

SCHEDULER_JOB_DEFAULTS = {
    'coalesce': True,
    'misfire_grace_time': 10 * 60,
    'max_instances': 3,
}

SCHEDULER_API_ENABLED = True

3、任务的执行

我们创建任务后，需要在Django启动后，我们的任务才会开始执行。

启动方式1：


from django_apscheduler.jobstores import register_events

register_events(SCHEDULER)

启动方式2：


from apscheduler.schedulers.background import BackgroundScheduler
from django_apscheduler.jobstores import DjangoJobStore
from django_apscheduler.jobstores import register_events

scheduler = BackgroundScheduler(timezone=get_current_timezone())
scheduler.add_jobstore(DjangoJobStore())
register_events(scheduler)
scheduler.start()

启动方式3：


python manage.py runapscheduler

四、任务的删除与修改

我们可以通过任务的id来删除或修改任务。

删除任务：


from django_apscheduler.models import DjangoJobExecution

DjangoJob.objects.filter(id=id).delete()
DjangoJobExecution.objects.filter(job_id=id).delete()

修改任务：


job = DjangoJob.objects.get(id=id)
job.next_run_time = django_timezone.now()
job.save()

五、小结

在本文中，我们详细介绍了Django Apscheduler的安装、配置、创建任务、任务调度器的配置以及任务的执行和删除等相关信息。希望读者可以通过本文对Django Apscheduler有一个深入的了解，并在实际开发中灵活运用。

原创文章，作者：XOENA，如若转载，请注明出处：https://www.506064.com/n/329836.html