官網(wǎng)文檔：http://apscheduler.readthedoc...
API：http://apscheduler.readthedoc...

APScheduler是一個(gè)python的第三方庫(kù)，用來(lái)提供python的后臺(tái)程序。包含四個(gè)組件，分別是：

triggers： 任務(wù)觸發(fā)器組件，提供任務(wù)觸發(fā)方式

job stores： 任務(wù)商店組件，提供任務(wù)保存方式

executors： 任務(wù)調(diào)度組件，提供任務(wù)調(diào)度方式

schedulers： 任務(wù)調(diào)度組件，提供任務(wù)工作方式

pip 安裝

$ pip install apscheduler

源碼安裝

$ python setup.py install

from apscheduler.schedulers.blocking import BlockingScheduler
import time

# 實(shí)例化一個(gè)調(diào)度器
scheduler = BlockingScheduler()
 
def job1():
    print "%s: 執(zhí)行任務(wù)"  % time.asctime()

# 添加任務(wù)并設(shè)置觸發(fā)方式為3s一次
scheduler.add_job(job1, "interval", seconds=3)

# 開(kāi)始運(yùn)行調(diào)度器
scheduler.start()

輸出：

$ python first.py
Fri Sep  8 20:41:55 2017: 執(zhí)行任務(wù)
Fri Sep  8 20:41:58 2017: 執(zhí)行任務(wù)
...

trigger提供任務(wù)的觸發(fā)方式，共三種方式：

date：只在某個(gè)時(shí)間點(diǎn)執(zhí)行一次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: 每隔一段時(shí)間執(zhí)行一次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!")

scheduler組件提供執(zhí)行的方式，在不同的運(yùn)用環(huán)境中選擇合適的方式

BlockingScheduler: 進(jìn)程中只運(yùn)行調(diào)度器時(shí)的方式

from apscheduler.schedulers.blocking import BlockingScheduler
import time

scheduler = BlockingScheduler()
 
def job1():
    print "%s: 執(zhí)行任務(wù)"  % time.asctime()

scheduler.add_job(job1, "interval", seconds=3)
scheduler.start()

BackgroundScheduler: 不想使用任何框架時(shí)的方式

from apscheduler.schedulers.background import BackgroundScheduler
import time

scheduler = BackgroundScheduler()
 
def job1():
    print "%s: 執(zhí)行任務(wù)"  % 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組件提供任務(wù)的調(diào)度方式

base

debug

gevent

pool(max_workers=10)

twisted

jobstore提供任務(wù)的各種持久化方式

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

如果使用了任務(wù)的存儲(chǔ)，開(kāi)啟時(shí)最好添加replace_existing=True，否則每次開(kāi)啟都會(huì)創(chuàng)建任務(wù)的副本
開(kāi)啟后任務(wù)不會(huì)馬上啟動(dòng)，可修改trigger參數(shù)

# 根據(jù)任務(wù)實(shí)例刪除
job = scheduler.add_job(myfunc, "interval", minutes=2)
job.remove()

# 根據(jù)任務(wù)id刪除
scheduler.add_job(myfunc, "interval", minutes=2, id="my_job_id")
scheduler.remove_job("my_job_id")

job = scheduler.add_job(myfunc, "interval", minutes=2)
# 根據(jù)任務(wù)實(shí)例
job.pause()
job.resume()

# 根據(jù)任務(wù)id暫停
scheduler.add_job(myfunc, "interval", minutes=2, id="my_job_id")
scheduler.pause_job("my_job_id")
scheduler.resume_job("my_job_id")

修飾：job.modify(max_instances=6, name="Alternate name")
重設(shè)：scheduler.reschedule_job("my_job_id", trigger="cron", minute="*/5")

開(kāi)啟 scheduler.start()

關(guān)閉 scheduler.shotdown(wait=True | False)

暫停 scheduler.pause()

繼續(xù) scheduler.resume()

監(jiān)聽(tīng) 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)

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_defaults, timezone=utc)