apscheduler 定时任务框架

1、APScheduler简介：

    Python的一个定时任务框架，知足用户定时执行或者周期性执行任务的需求，提供了基于日期date、固定的时间间隔interval、以及相似于Linux上的定时任务crontab类型的定时任务。而且该框架不只能够添加、删除定时任务，还能够将任务存储到数据库中，实现任务的持久化。

Python的第三方库，用来提供Python的后台程序。包含四个组件，分别是：

triggers：任务触发器组件，提供任务触发方式

job stores： 任务商店组件，提供任务保存方式

executors：任务调度组件，提供任务调度方式

schedulers: 任务调度组件，提供任务工做方式

 2、APScheduler安装

1）利用pip安装(推荐)

# pip install apscheduler
2 基于源码：https://pypi.python.org/pypi/APScheduler/

# python setup.py install

3、基本概念

一、 APScheduler有四种组件及相关说明

1） triggers(触发器)：触发器包含调度逻辑，每个做业有它本身的触发器，用于决定接下来哪个做业会运行，除了他们本身初始化配置外，触发器彻底是无状态的。
2 ）job stores(做业存储)：用来存储被调度的做业，默认的做业存储器是简单地把做业任务保存在内存中，其余做业存储器能够将任务保存到各类数据库中，支持MongoDB、Redis、SQLAlchemy存储方式。当对做业任务进行持久化存储的时候，做业的数据将被序列化，从新读取做业时在反序列化。

3）executors(执行器)：执行器用来执行定时任务，只是将须要执行的任务放在新的线程或者线程池中运行。看成业任务完成时，执行器就会通知调度器。对于执行器，默认状况下选择ThreadPoolExecutor就能够了，可是若是涉及到一下特殊任务如比较消耗CPU的任务则能够选择ProcessPoolExecutor，固然根据实际需求能够同时使用两种执行器

4）schedulers(调度器)：调度器是将其余部分联系在一块儿，通常在应用程序中只有一个调度器，应用开发者不会直接操做触发器、任务存储以及执行器。相反调度器提供了处理的接口。经过调度器完成任务的存储以及执行器的配置操做，如能够添加、移除、修改任务做业。

APScheduler提供了多种调度器，能够根据具体需求来选择合适的调度器，经常使用的调度器有：

BlockingScheduler：适合于只在进程中运行单个任务的状况，一般在调度器是你惟一要运行的东西时使用

BackgroundScheduler：适合于要求任务在程序后台运行的状况，当但愿调度器在应用后台执行时使用。

AsyncIOScheduler:适合于使用asyncio框架的状况

GeventScheduler:适合于使用gevent框架的状况

TornadoScheduler:适合于使用Tornado框架的应用

TwistedScheduler:适合使用Twisted框架的应用

QtScheduler:适合使用QT的状况

二、配置调度器

APScheduler提供了许多不一样的方式来配置调度器，你可使用一个配置字典或者做为参数关键字的方式传入。你也能够先建立调度器。在配置和添加做业，这样能够在不一样的环境中获得更大的灵活性。

三、简单的实例

 

from apscheduler.schedulers.blocking import BlockingScheduler import time #实例化一个调度器 scheduler = BlockingScheduler() def job1(): print "%s: 执行任务" % time.asctime() # 添加任务并设置触发方式为3s一次 scheduler.add_job(job1, 'interval', seconds=3) #开始运行调度器 scheduler.start()

 

4、各组件功能

 一、trigger组件

trigger提供任务的触发方式，共三种方式:

      date：只在某个时间点执行一次run_date(datetime|str)

scheduler.add_job(my_job, 'date', run_date=date(2017, 9, 8), args=[]) scheduler.add_job(my_job, 'date', run_date=datetime(2017, 9, 8, 21, 30, 5), args=[]) scheduler.add_job(my_job, 'date', run_date='2019-6-12 21:30:05', args=[]) # The 'date' trigger and datetime.now() as run_date are implicit sched.add_job(my_job, args=[[])

 interval:每隔一段时间执行一次weeks=0 | days=0 | hours=0 | minutes=0 | seconds=0,

start_date=None, end_date=None, timezone=None

scheduler.add_job(my_job, 'interval', hours=2) scheduler.add_job(my_job, 'interval', hours=2, start_date='2017-9-8 21:30:00', end_date='2019-06-12 21:30:00) @scheduler.scheduled_job('interval', id='my_job_id', hours=2) def my_job(): print("Hello World")

 

cron:使用Linux下crontab的方式(year=None, month=None, day=None, week=None, day_of_week=None, hour=None, minute=None, second=None, start_date=None, end_date=None, timezone=None)

 　     sched.add_job(my_job, 'cron', hour=3, minute=30)

    sched.add_job(my_job, 'cron', day_of_week='mon-fri', hour=5, minute=30, end_date='2017-10-
30')

    @sched.scheduled_job('cron', id='my_job_id', day='last sun')
    def some_decorated_task():

        print("I am printed at 00:00:00 on the last Sunday of every month!")

二、scheduler组件

scheduler组件提供执行的方式，在不一样的运行环境中选择合适的方式

 BlockingScheduler:进程中只运行调度器时的方式

from apscheduler.schedulers.blocking import BlockingScheduler import time scheduler = BlockingScheduler() def job1(): print "%s: 执行任务" % time.asctime() scheduler.add_job(job1, 'interval', seconds=3) scheduler.start()

BackgroundScheduler:不想使用任何框架时的方式

from apscheduler.schedulers.background import BackgroundScheduler import time scheduler = BackgroundScheduler() def job1(): print "%s:执行任务 " % time.asctime()  scheduler.add_job(job1, 'interval', seconds=3) scheduler.start() while True: pass

AsyncIOScheduler: asyncio module的方式( Python3)

from apscheduler.schedulers.asyncio import AsyncIOScheduler

try:

    import asyncio

except ImportError:

    import trollius as asyncio
...

...

# while True pass

try:

    asyncio.get_event_loop().run_forever()

except (KeyboardInterrupt, SystemExit):

    pass

GeventScheduler: gevent方式 

from apscheduler.schedulers.gevent import GeventScheduler ... ... g = scheduler.start() # while True:pass try: g.join() except (KeyboardInterrupt, SystemExit): pass

TornadoScheduler: Tornado方式

from tornado.ioloop import IOLoop from apscheduler.schedulers.tornado import TornadoScheduler ...  ... # while True:pass try: IOLoop.instance().start() except (KeyboardInterrupt, SystemExit): pass

TwistedScheduler: Twisted方式

from twisted.internet import reactor from apscheduler.schedulers.twisted import TwistedScheduler ...  ... # while True:pass try: reactor.run() except (KeyboardInterrupt, SystemExit): pass

 

QtScheduler: Qt方式

三、executors组件

executors组件提供任务的调度方式

base

debug
gevent 

pool(max_workers=10) 

twisted 

四、jobstore组件

jobstore提供任务的各类持久化方式

base

memory
mongodb
    scheduler.add_jobstore('mongodb', collection='example_jobs') 

redis

    scheduler.add_jobstore('redis', jobs_key='example.jobs', run_times_key='example.run_times')
rethinkdb
    scheduler.add_jobstore('rethinkdb', database='apscheduler_example') 

sqlalchemy

  scheduler.add_jobstore('sqlalchemy', url=url)

zookeeper

  scheduler.add_jobstore('zookeeper', path='/example_jobs')

 

5、任务操做

一、添加任务add_job(如上)

若是使用了任务的存储，开启时最好添加replace_existing=True，不然每次开启时都会建立任务的副本，开启后任务不会立刻启动，可修改triger参数

二、删除任务remove_job

#根据任务实例删除

job = scheduler.add_job(myfunc, 'interval', minutes=2) job.remove() # 根据任务id删除 scheduler.add_job(myfunc, 'interval', minutes=2, id='my_job_id') scheduler.remove_job('my_job_id')

三、任务的暂停pause_job和继续resume_job

job = scheduler.add_job(myfunc, 'interval', minutes=2)

#根据任务实例

job.pause()

job.resume()

# 根据任务id暂停

scheduler.add_job(myfunc, 'interval', minutes=2, id='my_job_id')

scheduler.pause_job('my_job_id')

四、任务的修饰modify和重设reschedule_job

修饰：job.modify(max_instances=6, name='Alternate name') 

重设：scheduler.reschedule_job('my_job_id', trigger='cron', minute='*/5')

五、调度器操做

开启：scheduler.start()

关闭：scheduler.shotdown(wait=True | False)

暂停：scheduler.pause()

继续：scheduler.resume() 

监听：http://apscheduler.readthedocs.io/en/v3.3.0/modules/events.html#module-apscheduler.events

def my_listener(event): if event.exception: print('The job crashed :(') else: print('The job worked :)') scheduler.add_listener(my_listener, EVENT_JOB_EXECUTED | EVENT_JOB_ERROR)

官方实例

from pytz import utc

from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.jobstores.mongodb import MongoDBJobStore
from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
from apscheduler.executors.pool import ThreadPoolExecutor, ProcessPoolExecutor

jobstores = {
    'mongo': MongoDBJobStore(),
    'default': SQLAlchemyJobStore(url='sqlite:///jobs.sqlite')

}
executors = {

    'default': ThreadPoolExecutor(20),

    'processpool': ProcessPoolExecutor(5)
}

job_defaults = {
    'coalesce': False,
    'max_instances': 3

}
scheduler = BackgroundScheduler(jobstores=jobstores, executors=executors,
job_defaults=job_default