1.APScheduler

之前在Django使用crontab来完成定时任务，还有一个更灵活的python工具APScheduler也可以完成

特点
1）可以动态添加任务
2）不依赖Linux的crontab系统定时
3）可以对添加的定时任务做持久保存

2.简单使用

pip install apscheduler

from apscheduler.schedulers.background import BackgroundScheduler

scheduler = BackgroundScheduler()  # 创建调度器对象

def my_job(param1, param2):  # 任务
    print(param1, param2)

scheduler.add_job(my_job, 'date', args=[100, 'python'])  # 添加任务

scheduler.start()  # 启动调度器

3.调度器、执行器、触发器

3.1 Scheduler 调度器

调度器是管理定时任务的

当使用BlockingScheduler时作为独立进程使用，会阻塞主线程
当使用BackgroundScheduler时会在后台运行，不会发生阻塞

from apscheduler.schedulers.blocking import BlockingScheduler
from apscheduler.schedulers.background import BackgroundScheduler

scheduler = BackgroundScheduler()
scheduler.start()  # 此处程序不会发生阻塞

scheduler = BlockingScheduler()
scheduler.start()  # 此处程序会发生阻塞

3.2 executors 执行器

执行器控制执行方式

在定时任务该执行时，ThreadPoolExecutor以线程方式执行任务，ProcessPoolExecutor以进程方式执行任务

# 方式1： 线程
from apscheduler.executors.pool import ThreadPoolExecutor
executors = {
    'default': ThreadPoolExecutor(20)   # 最多20个线程同时执行
}
scheduler = BackgroundScheduler(executors=executors)


# 方式2： 进程
from apscheduler.executors.pool import ProcessPoolExecutor
executors = {
    'default': ProcessPoolExecutor(3)  # 最多3个进程同时运行
}
scheduler = BackgroundScheduler(executors=executors)

3.3 Trigger 触发器

触发器控制的是什么时候会执行任务

1）date 在特定的时间日期执行

from datetime import date

# 在2020年11月11日00:00:00执行
sched.add_job(my_job, 'date', run_date=date(2020, 11, 11))

# 在2020年11月1日16:30:05
sched.add_job(my_job, 'date', run_date=datetime(2020, 11, 11, 16, 30, 5))
sched.add_job(my_job, 'date', run_date='2009-11-06 16:30:05')

# 立即执行
sched.add_job(my_job, 'date')  
sched.start()

2）interval 经过指定的时间间隔执行

from datetime import datetime

# 每两小时执行一次
sched.add_job(job_function, 'interval', hours=2)

# 在2010年10月10日09:30:00 到2014年6月15日的时间内，每两小时执行一次
sched.add_job(job_function, 'interval', hours=2, start_date='2010-10-10 09:30:00', end_date='2014-06-15 11:00:00')

时间间隔可选seconds 、minutes、hours、days、weeks、start_date、end_date 、timezone

3）cron 按指定的周期执行

# 每天早上10点30执行
sched.add_job(job_function, 'cron', hour=10, minute=30)

# 在6、7、8、11、12月的第三个周五的00:00, 01:00, 02:00和03:00 执行
sched.add_job(job_function, 'cron', month='6-8,11-12', day='3rd fri', hour='0-3')

# 在2014年5月30日前的周一到周五的5:30执行
sched.add_job(job_function, 'cron', day_of_week='mon-fri', hour=5, minute=30, end_date='2014-05-30')

可选周期second、minute、hour、day、day_of_week、month、year、start_date、end_date、timezone

参数	类型	可选值
year	int或str	年份，4个数字
month	int或str	月份，1-12
day	int或str	日，1-31
week	int或str	1-53
day_of_week	int或str	0-6或者英文简写mon、tue、wed、thu、fri、sat、sun
hour	int或str	0-23
minute	int或str	0-59
second	int或str	0-59
start_date	datetime或str	最早可能触发的日期/时间
end_date	datetime或str	最晚可能触发的日期/时间
timezone	datetime.tzinfo或str	时区
jitter	int或None	最多延迟执行秒数

说明：

说明

• 第一个工作日始终是星期一
• 年月日时分秒不设置默认为“*”

4.任务管理

4.1 添加、移除、暂停、恢复、停止

# 方式1： 通过对象
job = scheduler.add_job(myfunc, 'interval', minutes=2)  # 添加任务
job.remove()  # 移除任务
job.pause() # 暂停任务
job.resume()  # 恢复任务

# 方式2： 通过任务id
scheduler.add_job(myfunc, 'interval', minutes=2, id='my_job_id')  # 添加任务    
scheduler.remove_job('my_job_id')  # 移除任务
scheduler.pause_job('my_job_id')  # 暂停任务
scheduler.resume_job('my_job_id')  # 恢复任务

scheduler.shutdown()  # 停止任务

4.2 修改任务

# 方式1： 通过对象
job.modify(max_instances=6, name='Alternate name')

# 方式2： 通过任务id
scheduler.reschedule_job('my_job_id', trigger='cron', minute='*/5')

5.在Django使用APScheduler

pip install django-apscheduler 

INSTALLED_APPS = [
    ......
    'django_apscheduler',# 定时执行任务
]

一定要先迁移

python manage.py migrate

迁移完了之后可以在任意一个APP的views.py写定时代码

使用装饰器写法

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

# 1.实例化调度器
scheduler = BackgroundScheduler()
# 2.调度器使用DjangoJobStore()
scheduler.add_jobstore(DjangoJobStore(), "default")
try:
    # 3.设置定时任务
    # 另一种方式为每天固定时间执行任务，对应代码为
    @register_job(scheduler,"cron", hour=4)
    def my_job():
        # 这里写你要执行的任务
        pass
    # 4.注册定时任务(0.4.0版本之后不需要注册)
    # register_events(scheduler)
    # 5.开启定时任务
    scheduler.start()
except Exception as e:
    print(e)
    # 有错误就停止定时器
    scheduler.shutdown()

普通写法

def my_job():
    print("do django-apscheduler job")

scheduler = BackgroundScheduler()
scheduler.add_jobstore(DjangoJobStore(), "default")
 # 此处触发器是每分钟执行一次，如果不知道ID它会随机生成一个32位ID
scheduler.add_job(my_job, trigger=CronTrigger(minute="*/1"), id="my_job") 
try:
    scheduler.start()
except KeyboardInterrupt:
    scheduler.shutdown()

6.报错

apscheduler.jobstores.base.ConflictingIdError: 'Job identifier (my_job) conflicts with an existing job'
如果你在启动任务的时候指定了任务ID，可能会出现这种ID冲突的情况，只需要进入数据库找到django_apscheduler_djangojob表，修改ID或者删掉记录即可

关于在Django使用crontab实现定时任务可以参考
【Django】页面静态化和crontab定时任务