shedule 和 apscheduler

一、`schedule` 模块

schedule 是 Python 中一个轻量级的定时任务调度库，它可以完成每分钟、每小时、每天、每周几等特定日期的定时任务。因此十分方便我们执行一些轻量级的定时任务。

Schedule 是 Python3 的一个第三方模块，安装方式如下：

pip3 install schedule

示例：

import schedule
import timedef job():print("I'm working...")# 每十分钟
schedule.every(10).minutes.do(job)
# 每小时
schedule.every().hour.do(job)
# 每天的10:30
schedule.every().day.at("10:30").do(job)
# 每隔 5 天到 10 天
schedule.every(5).to(10).days.do(job)
# 每周一
schedule.every().monday.do(job)
# 每周三的 13:15
schedule.every().wednesday.at("13:15").do(job)while True:# 运行所有任务schedule.run_pending()time.sleep(1)

如果，函数中带有参数，则可以如下：

import schedule
import timedef job(name):print("her name is : ", name)name = xiaona
schedule.every(10).minutes.do(job, name)
schedule.every().hour.do(job, name)
schedule.every().day.at("10:30").do(job, name)
schedule.every(5).to(10).days.do(job, name)
schedule.every().monday.do(job, name)
schedule.every().wednesday.at("13:15").do(job, name)while True:schedule.run_pending()time.sleep(1)

存在的问题：

需要定时运行的函数job不应当是死循环类型的，也就是说，这个线程应该有一个执行完毕的出口。一是因为线程万一僵死，会是非常棘手的问题；二是下一次定时任务还会开启一个新的线程，执行次数多了就会演变成灾难。
如果schedule的时间间隔设置得比job执行的时间短，一样会线程堆积形成灾难，也就是说，我job的执行时间是1个小时，但是我定时任务设置的是5分钟一次，那就会一直堆积线程。

二、`apscheduler` 模块

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

triggers：任务触发器组件，提供任务触发方式
job stores：任务商店组件，提供任务保存方式
executors：任务调度组件，提供任务调度方式
schedulers：任务调度组件，提供任务工作方式

1、安装`apscheduler`

pip3 install apscheduler

2、简单的示例

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()

输出：

$ python first.py
Fri Sep  8 20:41:55 2017: 执行任务
Fri Sep  8 20:41:58 2017: 执行任务
...

3、各组件的功能

（1）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='2017-9-08 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='2018-06-15 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!")

（2）scheduler 组件

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

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

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

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

from apscheduler.schedulers.background import BackgroundScheduler
import timescheduler = 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

（3）executors 组件

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

base
debug
gevent
pool(max_workers=10)
twisted

（4）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')

4、任务操作

（1）添加任务add_job

如果使用了任务的存储，开启时最好添加replace_existing=True，否则每次开启都会创建任务的副本
开启后任务不会马上启动，可修改trigger参数

（2）删除任务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')

（3）任务的暂停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')
scheduler.resume_job('my_job_id')

（4）任务的修饰modify和重设reschedule_job

修饰：job.modify(max_instances=6, name='Alternate name')
重设：scheduler.reschedule_job('my_job_id', trigger='cron', minute='*/5')

5、调度器操作

开启 scheduler.start()
关闭 scheduler.shotdown(wait=True | False)
暂停 scheduler.pause()
继续 scheduler.resume()
监听 http://apscheduler.readthedoc…

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)

6、官方示例

from pytz import utcfrom apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.jobstores.mongodb import MongoDBJobStore
from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
from apscheduler.executors.pool import ThreadPoolExecutor, ProcessPoolExecutorjobstores = {'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_defaults, timezone=utc)