APScheduler定时框架

Python定时库APScheduler原理及用法

APScheduler简介

APscheduler全称Advanced Python Scheduler

作用为在指定的时间规则执行指定的作业。

• 指定时间规则的方式可以是间隔多久执行，可以是指定日期时间的执行，也可以类似Linux系统中Crontab中的方式执行任务。
• 指定的任务就是一个Python函数。

APScheduler四个组件

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

1. 调度器(scheduler)

APScheduler提供了七种调度器：

BlockingScheduler：适合于只在进程中运行单个任务的情况，通常在调度器是你唯一要运行的东西时使用。
BackgroundScheduler: 适合于要求任何在程序后台运行的情况，当希望调度器在应用后台执行时使用。
AsyncIOScheduler：适合于使用asyncio异步框架的情况
GeventScheduler: 适合于使用gevent框架的情况
TornadoScheduler: 适合于使用Tornado框架的应用
TwistedScheduler: 适合使用Twisted框架的应用
QtScheduler: 适合使用QT的情况

配置调度器

APScheduler提供了许多不同的方式来配置调度器，你可以使用一个配置字典或者作为参数关键字的方式传入。你也可以先创建调度器，再配置和添加作业。

下面来看一个简单的 BlockingScheduler 例子

from apscheduler.schedulers.blocking import BlockingScheduler
from datetime import datetime

def job():
    print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
# 定义BlockingScheduler
sched = BlockingScheduler()
sched.add_job(job, 'interval', seconds=5)
sched.start()

上述代码创建了一个 BlockingScheduler，并使用默认内存存储和默认执行器。(默认选项分别是 MemoryJobStore 和 ThreadPoolExecutor，其中线程池的最大线程数为10)。配置完成后使用 start() 方法来启动。

2.作业存储(job store)

APScheduler提供了四种存储方式：

MemoryJobStore
sqlalchemy
mongodb
redis

存储被调度的作业，默认的作业存储是简单地把作业保存在内存中，其他的作业存储是将作业保存在数据库中。一个作业的数据将在保存在持久化作业存储时被序列化，并在加载时被反序列化。调度器不能分享同一个作业存储。

APScheduler 默认使用 MemoryJobStore，可以修改使用 DB 存储方案

如果想要显式设置 job store(使用mongo存储)和 executor 可以这样写：

from datetime import datetime
from pymongo import MongoClient
from apscheduler.schedulers.blocking import BlockingScheduler
from apscheduler.jobstores.memory import MemoryJobStore
from apscheduler.jobstores.mongodb import MongoDBJobStore
from apscheduler.executors.pool import ThreadPoolExecutor, ProcessPoolExecutor
# MongoDB 参数
host = '127.0.0.1'
port = 27017
client = MongoClient(host, port)
# 输出时间
def job():
    print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
# 存储方式
jobstores = {
    'mongo': MongoDBJobStore(collection='job', database='test', client=client),
    'default': MemoryJobStore()
}
executors = {
    'default': ThreadPoolExecutor(10),
    'processpool': ProcessPoolExecutor(3)
}
job_defaults = {
    'coalesce': False,
    'max_instances': 3
}
scheduler = BlockingScheduler(jobstores=jobstores, executors=executors, job_defaults=job_defaults)
scheduler.add_job(job, 'interval', seconds=5, jobstore='mongo')
scheduler.start()

  在运行程序5秒后，第一次输出时间。
  在 MongoDB 中可以看到 job 的状态

2.1对 job 的操作

添加 job

添加job有两种方式：

1. add_job()
2. scheduled_job()

第二种方法只适用于应用运行期间不会改变的 job，而第一种方法返回一个apscheduler.job.Job 的实例，可以用来改变或者移除 job。

from apscheduler.schedulers.blocking import BlockingScheduler
sched = BlockingScheduler()
# 装饰器
@sched.scheduled_job('interval', id='my_job_id', seconds=5)
def job_function():
    print("Hello World")
# 开始
sched.start()

@sched.scheduled_job() 是 Python 的装饰器。

移除 job

移除 job 也有两种方法：

       * remove放到start前才有效

1. remove_job()
2. job.remove()

remove_job 使用 jobID 移除
job.remove() 使用 add_job() 返回的实例  

# 根据返回的实例来删除
job_name = scheduler.add_job(myfunc, 'interval', minutes=2)
job_name.remove()
# 根据定义的id来删除
scheduler.add_job(myfunc, 'interval', minutes=2, id='my_job_id')
scheduler.remove_job('my_job_id')

暂停和恢复 job

暂停一个 job：

scheduler.job.Job.pause() 

scheduler.schedulers.base.BaseScheduler.pause_job()

　　

恢复一个 job：

scheduler.job.Job.resume()

scheduler.schedulers.base.BaseScheduler.resume_job()

希望你还记得 apscheduler.job.Job 是 add_job() 返回的实例

获取 job 列表

获得可调度 job 列表，可以使用get_jobs() 来完成，它会返回所有的 job 实例。

也可以使用print_jobs() 来输出所有格式化的 job 列表。

scheduler.get_jobs()
scheduler.print_jobs()

修改 job

除了 jobID 之外 job 的所有属性都可以修改，使用 apscheduler.job.Job.modify() 或者 modify_job() 修改一个 job 的属性

job.modify(max_instances=6, name='Alternate name')
modify_job('my_job_id', trigger='cron', minute='*/5')

关闭 job

默认情况下调度器会等待所有的 job 完成后，关闭所有的调度器和作业存储。将 wait 选项设置为 False 可以立即关闭。

scheduler.shutdown()
scheduler.shutdown(wait=False)

　　

************************************************************************************************************************************************

3.执行器(executor)

   处理作业的运行，他们通常通过在作业中提交制定的可调用对象到一个线程或者进城池来进行。当作业完成时，执行器将会通知调度器。
最常用的 executor 有两种：

ProcessPoolExecutor
ThreadPoolExecutor

4. 触发器(trigger)

Trigger绑定到Job，在scheduler调度筛选Job时，根据触发器的规则计算出Job的触发时间，然后与当前时间比较
确定此Job是否会被执行，总之就是根据trigger规则计算出下一个执行时间。

APScheduler 有三种内建的 trigger:

date: 特定的时间点触发
interval: 固定时间间隔触发
cron: 在特定时间周期性地触发

date

最基本的一种调度，作业只会执行一次。它的参数如下：

• run_date (datetime|str) – the date/time to run the job at
• timezone (datetime.tzinfo|str) – time zone for run_date if it doesn’t have one already

from datetime import date
from apscheduler.schedulers.blocking import BlockingScheduler
sched = BlockingScheduler()
def my_job(text):
    print(text)
sched.add_job(my_job, 'date', run_date=date(2009, 11, 6))
sched.add_job(my_job, 'date', run_date=datetime(2009, 11, 6, 16, 30, 5))
sched.add_job(my_job, 'date', run_date='2009-11-06 16:30:05')
sched.start()　

cron  在特定时间定期运行 job

主要参数：

• year (int|str) – 4-digit year
• month (int|str) – month (1-12)
• day (int|str) – day of the (1-31)
• week (int|str) – ISO week (1-53)
• day_of_week (int|str) – number or name of weekday (0-6 or mon,tue,wed,thu,fri,sat,sun)
• hour (int|str) – hour (0-23)
• minute (int|str) – minute (0-59)
• second (int|str) – second (0-59)
• start_date (datetime|str) – earliest possible date/time to trigger on (inclusive)
• end_date (datetime|str) – latest possible date/time to trigger on (inclusive)
• timezone (datetime.tzinfo|str) – time zone to use for the date/time calculations (defaults to scheduler timezone)

cron其他参数格式用法示例如下：

#表示2017年3月22日17时19分07秒执行该程序
sched.add_job(my_job, 'cron', year=2017,month = 03,day = 22,hour = 17,minute = 19,second = 07)
 
#表示任务在6,7,8,11,12月份的第三个星期五的00:00,01:00,02:00,03:00 执行该程序
sched.add_job(my_job, 'cron', month='6-8,11-12', day='3rd fri', hour='0-3')
 
#表示从星期一到星期五5:30（AM）直到2020-12-30 00:00:00
sched.add_job(my_job(), 'cron', day_of_week='mon-fri', hour=5, minute=30,end_date='2020-12-30')
 
#表示每5秒执行该程序一次，相当于interval 间隔调度中seconds = 5
sched.add_job(my_job, 'cron',second = '*/5')

　　

interval  以固定的时间间隔运行 job

 

参数：

• weeks (int) – number of weeks to wait
• days (int) – number of days to wait
• hours (int) – number of hours to wait
• minutes (int) – number of minutes to wait
• seconds (int) – number of seconds to wait
• start_date (datetime|str) – starting point for the interval calculation
• end_date (datetime|str) – latest possible date/time to trigger on
• timezone (datetime.tzinfo|str) – time zone to use for the date/time calculations

from apscheduler.schedulers.background import BackgroundScheduler


def job():
	print('job')


scheduler = BackgroundScheduler()
scheduler.start()

# 每 2 小时运行一次
scheduler.add_job(
	job,
	trigger='interval',
	hours=2
)

# 2019-10-01 00:00:00 到 2019-10-31 23:59:59 之间每 2 小时运行一次
scheduler.add_job(
	job,
	trigger='interval',
	hours=2,
	start_date='2019-10-01 00:00:00',
	end_date='2019-10-31 23:59:59',
)

调度器事件监听

 任何代码都可能抛出异常，如何第一时间知道，这才是我们最关心的，apscheduler 已经为我们想到了这些，提供了事件监听来解决这一问题。

  Event 事件

Event是APScheduler在进行某些操作时触发相应的事件，用户可以自定义一些函数来监听这些事件，
当触发某些Event时，做一些具体的操作
常见的比如。Job执行异常事件 EVENT_JOB_ERROR。Job执行时间错过事件 EVENT_JOB_MISSED。
目前APScheduler定义的Event：

EVENT_SCHEDULER_STARTED
EVENT_SCHEDULER_START
EVENT_SCHEDULER_SHUTDOWN
EVENT_SCHEDULER_PAUSED
EVENT_SCHEDULER_RESUMED
EVENT_EXECUTOR_ADDED
EVENT_EXECUTOR_REMOVED
EVENT_JOBSTORE_ADDED
EVENT_JOBSTORE_REMOVED
EVENT_ALL_JOBS_REMOVED
EVENT_JOB_ADDED
EVENT_JOB_REMOVED
EVENT_JOB_MODIFIED
EVENT_JOB_EXECUTED
EVENT_JOB_ERROR
EVENT_JOB_MISSED
EVENT_JOB_SUBMITTED
EVENT_JOB_MAX_INSTANCES

监听器使用：

　

import datetime

from apscheduler.schedulers.blocking import BlockingScheduler
import  jobContent
from apscheduler.events import EVENT_JOB_EXECUTED, EVENT_JOB_ERROR

# 定义BlockingScheduler调度器
scheduler = BlockingScheduler()

#job定义
def weather_job():
   jobContent.weather_tips()

#添加
def listener(event):
   if event.exception:
      print('任务出错了！！！！！！')
   else:
      print('任务照常运行...')

#添加job
#cron day_of_week ：0-6 or mon,tue,wed,thu,fri,sat,sun
#工作日天气提醒
scheduler.add_job(weather_job,'cron', day_of_week='0-4', hour=9, minute=30,id='weather_id')

scheduler.add_listener(listener,EVENT_JOB_EXECUTED|EVENT_JOB_ERROR)


#启动job
scheduler.start()