当前位置：首页 > news >正文

Django 使用Apscheduler执行定时任务

news 来源：原创 2024/9/20 8:15:13

APScheduler是一个轻量级的Python库，用于调度作业（运行定时任务）。在Django中使用APScheduler，你可以设置定时任务，比如每天定时发送报告，或者定时清理无效数据。

以下是一个简单的例子，展示如何在Django项目中使用APScheduler：

1.首先，安装APScheduler：

pip install apscheduler

2.在Django应用中创建一个任务文件，例如jobs.py：

from apscheduler.schedulers.background import BackgroundScheduler
import datetimedef my_job():print(f"任务执行：{datetime.datetime.now()}")# 创建调度器
scheduler = BackgroundScheduler()
# 添加任务
scheduler.add_job(my_job, 'interval', minutes=1)
# 启动调度器
scheduler.start()

3.在Django的apps.py中启动调度器，确保在Django启动时任务能被加载：

from django.apps import AppConfig
from .jobs import schedulerclass MyAppConfig(AppConfig):name = 'my_app'def ready(self):# 当Django应用就绪时，启动调度器scheduler.start()

4.在init.py中设置AppConfig：

default_app_config = 'my_app.apps.MyAppConfig'

确保你的Django应用的名字替换成你实际的应用名。

以上代码将每分钟执行一次my_job函数。你可以根据需要调整任务的触发方式（例如，改变为每天或每小时一次）。记得在生产环境中部署时，确保调度器的启动逻辑只在一个进程中执行，避免多个进程执行相同任务的问题。

5.代码结构

在这里插入图片描述

6.代码优化：重复执行任务

INSTALLED_APPS = (# ..."django_apscheduler",
)

django_apscheduler_djangojob 表保存注册的任务以及下次执行的时间django_apscheduler_djangojobexecution 保存每次任务执行的时间和结果和任务状态

在这里插入图片描述

# views.py
from django_apscheduler.jobstores import DjangoJobStore, register_jobfrom apscheduler.schedulers.blocking import BlockingScheduler
from datetime import datetimescheduler = BlockingScheduler()  # 创建调度器
scheduler.add_jobstore(DjangoJobStore(), "default")# 添加定时任务方式一
@register_job(scheduler, "interval", seconds=5, id="func", replace_existing=True, misfire_grace_time=120)
def job():print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"))# 添加定时任务方式二
def job1():print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"))scheduler.add_job(job1,'interval',seconds=5,id='my_job',  # 任务的唯一标识replace_existing=True,
)scheduler.start()