APScheduler 支持三種調(diào)度任務(wù)：固定時(shí)間間隔，固定時(shí)間點(diǎn)（日期），Linux 下的 Crontab 命令。同時(shí)，它還支持異步執(zhí)行、后臺執(zhí)行調(diào)度任務(wù)。

一、基本架構(gòu)

1. 觸發(fā)器 triggers：設(shè)定觸發(fā)任務(wù)的條件
2. 描述一個(gè)任務(wù)何時(shí)被觸發(fā)，按日期或按時(shí)間間隔或按 cronjob 表達(dá)式三種方式觸發(fā)
3. 任務(wù)存儲器 job stores：存放任務(wù)，可以放內(nèi)存(默認(rèn))或數(shù)據(jù)庫
4. 注：調(diào)度器之間不能共享任務(wù)存儲器
5. 執(zhí)行器 executors：用于執(zhí)行任務(wù)，可設(shè)定執(zhí)行模式
6. 將指定的作業(yè)提交到線程池或者進(jìn)程池中運(yùn)行，任務(wù)完成通知調(diào)度器觸發(fā)相應(yīng)的事件。
7. 調(diào)度器 schedulers：將上方三個(gè)組件作為參數(shù)，創(chuàng)建調(diào)度器實(shí)例執(zhí)行。

協(xié)調(diào)三個(gè)組件的運(yùn)行。

二、調(diào)度器組件(schedulers)

• BlockingScheduler 阻塞式調(diào)度器

調(diào)度程序是進(jìn)程中唯一運(yùn)行的進(jìn)程，調(diào)用start函數(shù)會阻塞當(dāng)前線程，不能立即返回。

            
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 后臺調(diào)度器

當(dāng)前線程不會阻塞，調(diào)度器后臺執(zhí)行

            
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()
time.sleep(10)
          

注：10秒執(zhí)行完后，程序結(jié)束。

• AsyncIOScheduler AsyncIO調(diào)度器

適用于使用了asyncio的情況

            
from apscheduler.schedulers.asyncio import AsyncIOScheduler
import asyncio
...
...
try:
 asyncio.get_event_loop().run_forever()
except (KeyboardInterrupt, SystemExit):
 pass
          

• GeventScheduler Gevent調(diào)度器

使用了Gevent的情況

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

• TornadoScheduler Tornado調(diào)度器

適用于構(gòu)建Tornado應(yīng)用

• TwistedScheduler Twisted調(diào)度器

適用于構(gòu)建Twisted應(yīng)用

• QtScheduler Qt調(diào)度器

適用于構(gòu)建Qt應(yīng)用

三、觸發(fā)器組件(trigger)

date ：只在某個(gè)時(shí)間點(diǎn)執(zhí)行一次，具體日期

run_date(datetime|str)

            
scheduler.add_job(my_job, 'date', run_date=datetime(2019, 7, 12, 15, 30, 5), args=[])
scheduler.add_job(my_job, 'date', run_date="2019-07-12", args=[])
          

timezone 指定時(shí)區(qū)

interval ：每隔一段時(shí)間允許一次，時(shí)間間隔

            
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)
          

cron ：任務(wù)的運(yùn)行周期

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

除了week和 day_of_week，它們的默認(rèn)值是?*

例如?day=1, minute=20 ，這就等于 ?year='*', month='*', day=1, week='*', day_of_week='*', hour='*', minute=20, second=0 ，工作將在每個(gè)月的第一天以每小時(shí)20分鐘的時(shí)間執(zhí)行

表達(dá)式類型

注：當(dāng)設(shè)置的時(shí)間間隔小于，任務(wù)的執(zhí)行時(shí)間，線程會阻塞住，等待執(zhí)行完了才能執(zhí)行下一個(gè)任務(wù)，可以設(shè)置?max_instance 指定一個(gè)任務(wù)同一時(shí)刻有多少個(gè)實(shí)例在運(yùn)行，默認(rèn)為1

四、配置調(diào)度器

線程池執(zhí)行器默認(rèn)為10，內(nèi)存任務(wù)存儲器為?memoryjobstore ,如果想自己配置的話可以執(zhí)行以下操作

需求：

• 兩個(gè)任務(wù)儲存器分別搭配兩個(gè)執(zhí)行器；同時(shí)，還要修改任務(wù)的默認(rèn)參數(shù)；最后還要改時(shí)區(qū)
• 名稱為“mongo”的?MongoDBJobStore
• 名稱為“default”的?SQLAlchemyJobStore
• 名稱為“ThreadPoolExecutor ”的?ThreadPoolExecutor ，最大線程20個(gè)
• 名稱“processpool”的?ProcessPoolExecutor ，最大進(jìn)程5個(gè)
• UTC時(shí)間作為調(diào)度器的時(shí)區(qū)
• 默認(rèn)為新任務(wù)關(guān)閉?合并模式 （）
• 設(shè)置新任務(wù)的默認(rèn)最大實(shí)例數(shù)為3

方法一：

            
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)
          

方法二：

            
from apscheduler.schedulers.background import BackgroundScheduler
# The "apscheduler." prefix is hard coded
scheduler = BackgroundScheduler({
 'apscheduler.jobstores.mongo': {
   'type': 'mongodb'
 },
 'apscheduler.jobstores.default': {
  'type': 'sqlalchemy',
  'url': 'sqlite:///jobs.sqlite'
 },
 'apscheduler.executors.default': {
  'class': 'apscheduler.executors.pool:ThreadPoolExecutor',
  'max_workers': '20'
 },
 'apscheduler.executors.processpool': {
  'type': 'processpool',
  'max_workers': '5'
 },
 'apscheduler.job_defaults.coalesce': 'false',
 'apscheduler.job_defaults.max_instances': '3',
 'apscheduler.timezone': 'UTC',
})
          

方法三：

            
from pytz import utc
from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
from apscheduler.executors.pool import ProcessPoolExecutor
jobstores = {
 'mongo': {'type': 'mongodb'},
 'default': SQLAlchemyJobStore(url='sqlite:///jobs.sqlite')
}
executors = {
 'default': {'type': 'threadpool', 'max_workers': 20},
 'processpool': ProcessPoolExecutor(max_workers=5)
}
job_defaults = {
 'coalesce': False,
 'max_instances': 3
}
scheduler = BackgroundScheduler()
# ..這里可以添加任務(wù)
scheduler.configure(jobstores=jobstores, executors=executors, job_defaults=job_defaults, timezone=utc)
          

五、啟動調(diào)度器

除了?BlockingScheduler 外，其他非阻塞的調(diào)度器都會立即返回，運(yùn)行之后的代碼。

BlockingScheduler 需要將運(yùn)行的代碼放在start()之前

1.添加任務(wù)

1.調(diào)用add_job()? #可以傳參max_instance,同一任務(wù)的運(yùn)行實(shí)例個(gè)數(shù)

? 當(dāng)有任務(wù)中途中斷，后面恢復(fù)后，有N個(gè)任務(wù)沒有執(zhí)行?
??? coalesce：true ，恢復(fù)的任務(wù)會執(zhí)行一次
??? coalesce：false，恢復(fù)后的任務(wù)會執(zhí)行N次配合misfire_grace_time使用
??? misfire_grace_time設(shè)置時(shí)間差值，由于某些原因沒有運(yùn)行，再次提交時(shí)，大于設(shè)置的時(shí)間，實(shí)例不會運(yùn)行。

2.裝飾器scheduled_job()

立即運(yùn)行可以不設(shè)置trigger參數(shù)

2.移除任務(wù)

            
# 根據(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')


          

3.暫停任務(wù)

            
job = scheduler.add_job(myfunc, 'interval', minutes=2)
# 根據(jù)任務(wù)實(shí)例
job.pause() #暫停
job.resume() #繼續(xù)
# 根據(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')
          

4.調(diào)度器操作

            
scheduler.start() #開啟
scheduler.shotdown(wait=True|False) #關(guān)閉 False 無論任務(wù)是否執(zhí)行，強(qiáng)制關(guān)閉
          

異常捕獲

            
# 可以添加apscheduler日志至DEBUG級別，這樣就能捕獲異常信息
import logging
logging.basicConfig()
logging.getLogger('apscheduler').setLevel(logging.DEBUG)
          

總結(jié)

以上所述是小編給大家介紹的Python定時(shí)任務(wù)APScheduler的實(shí)例實(shí)例詳解,希望對大家有所幫助，如果大家有任何疑問請給我留言，小編會及時(shí)回復(fù)大家的。在此也非常感謝大家對腳本之家網(wǎng)站的支持！
如果你覺得本文對你有幫助，歡迎轉(zhuǎn)載，煩請注明出處，謝謝！