dajngo-apscheduler 实现定时任务

准备：安装模块

pip install django-apscheduler

首先，创建一个django应用，这里是：应用名称：apscheduler_task

settings.py 注册，例如下：注册模块，注册应用

INSTALLED_APPS = [
    'django.contrib.admin',
    'django.contrib.auth',
    'django.contrib.contenttypes',
    'django.contrib.sessions',
    'django.contrib.messages',
    'django.contrib.staticfiles',
    'rest_framework',
    'django_apscheduler',
    'apscheduler_task',
    'django_filters',
    'corsheaders'
]

views.py 写定时任务逻辑，例如下：

from apscheduler.schedulers.background import BackgroundScheduler
from django_apscheduler.jobstores import DjangoJobStore, register_events, register_job
from rest_framework.response import Response
import datetime
from utils.log_maker import logger

scheduler = BackgroundScheduler()
scheduler.add_jobstore(DjangoJobStore(), "default")


@register_job(scheduler, 'cron', id='materier', day_of_week='mon-sun', hour='11', minute='00',misfire_grace_time=3600)
def task1():
    pass


@register_job(scheduler, 'cron', id='runp', day_of_week='mon-sun', hour='11', minute='00',misfire_grace_time=3600)
def task2():
    # 注意仅仅从django表得到的报错信息是不能确定报错位置的，需要异常处理打印异常。
    try:
        pass
    except Exception as e:
        print(e)


register_events(scheduler)
scheduler.start()

urls.py 导入应用，注意：是导入，而不是其中的路由，例如下：

当多进程启动，会出现重复添加定时任务的报错，所以，采用 lock 文件，使其只能生成一次 任务id 到django定时任务表中。注意：数据迁移时会报错，应该注视掉如下包括导入语句的代码，数据迁移后，打开即可。

lock_path = os.path.join(settings.BASE_DIR,'aps.lock')
print('lock',lock_path)
if not os.path.exists(lock_path):
    from apps.apscheduler_task import views
    fd = open(lock_path, mode="w", encoding="utf-8")
    fd.close()

django自动生成的表如下：

django_apscheduler_djangojob：存储任务id，执行时间等信息。id不能重复。
django_apscheduler_djangojobexecution：记录执行任务信息，是否报错。

掘金文章：https://juejin.cn/post/6844903885400702989