django_apscheduler

定时任务django-apscheduler使用教程

场景

redis持久化存储时，使用APScheduler，使数据同步。
用户下单后使用，规定30min内必须支付，否则取消订单。

优点

可以独立运行，也可以放在程序（如Django、Flask）中。
灵活，可以在程序开始前开启定时任务，也可以执行到某个任务时，立即可开启定时任务。（下单30min后取消）
如果依赖程序的话，会占用程序资源；

定时任务django-apscheduler安装

pip install django-apscheduler
或者
pip install apscheduler

离线安装顺序

复制，tab键补全版本
pip install zipp
## pip install importlib-resources 包名，装下面的下划线
pip install importlib_resources
pip install backports.zoneinfo
pip install tzdata
pip install pytz_deprecation_shim-
## pip install importlib-metadata 包名，装下面的下划线
pip install importlib_metadata
pip install tzlocal
pip install APScheduler
## pip install django-apscheduler 包名，装下面的下划线
pip install django_apscheduler

settings.py 注册

INSTALLED_APPS = [
    ...
    'django_apscheduler',     # 新加入的定时任务插件django-apscheduler
]

python manage.py migrate 生成2个表格

django_apscheduler_djangojob——用于存储任务的表格
django_apscheduler_djangojobexecution——用于存储任务执行状态的表格

代码 只能在view.py写

import time
from apscheduler.schedulers.background import BackgroundScheduler
from django_apscheduler.jobstores import DjangoJobStore, register_job, register_events
print('django-apscheduler')
## 实例化调度器
scheduler = BackgroundScheduler()
## 调度器使用DjangoJobStore()
scheduler.add_jobstore(DjangoJobStore(), "default")
## 添加任务1
## 每隔5s执行这个任务
@register_job(scheduler,"interval", seconds=5,args=['王路'],id='job1', replace_existing = True)
## 每天8点半执行这个任务
## @register_job(scheduler, 'cron', id='test', hour=8, minute=30，args=['test'], replace_existing = True)
def job1(name):
    # 具体要执行的代码
    print('{} 任务运行成功！{}'.format(name,time.strftime("%Y-%m-%d %H:%M:%S")))
## 监控任务——注册定时任务
register_events(scheduler)
## 调度器开始运行
scheduler.start()

replace_existing = True
这个参数解决第二次启动失败的问题
'Job identifier (job1) conflicts with an existing job'

参数

定时任务django-apscheduler启动

python manage.py runserver

Django启动，就启动

定时任务django-apscheduler组件

APScheduler 有四种组件，分别是：调度器(scheduler)，作业存储(job store)，触发器(trigger)，执行器(executor)。

定时任务django-apscheduler运行时间date

## 在 2017-12-13 时刻运行一次 job_func 方法
scheduler .add_job(job_func, 'date', run_date=date(2017, 12, 13), args=['text'])
## 在 2017-12-13 14:00:00 时刻运行一次 job_func 方法
scheduler .add_job(job_func, 'date', run_date=datetime(2017, 12, 13, 14, 0, 0), args=['text'])
## 在 2020-12-13 14:00:01 时刻运行一次 job_func 方法
scheduler.add_job(job3,"date",run_date='2020-12-13 14:00:01',args=['王飞'],id="job3")
## 如果date后面没有参数的话，就是立刻执行代码，一般测试的时候用
app.scheduler.add_job(函数名, "date", args=[app])