Python定時(shí)任務(wù)APScheduler的實(shí)例實(shí)例詳解

更新時(shí)間：2019年07月22日 08:53:06 作者：mangM

APScheduler 支持三種調(diào)度任務(wù)：固定時(shí)間間隔，固定時(shí)間點(diǎn)（日期），Linux 下的 Crontab 命令。這篇文章主要介紹了Python定時(shí)任務(wù)APScheduler的使用,需要的朋友可以參考下

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

一、基本架構(gòu)

觸發(fā)器 triggers：設(shè)定觸發(fā)任務(wù)的條件
描述一個(gè)任務(wù)何時(shí)被觸發(fā)，按日期或按時(shí)間間隔或按 cronjob 表達(dá)式三種方式觸發(fā)
任務(wù)存儲(chǔ)器 job stores：存放任務(wù)，可以放內(nèi)存(默認(rèn))或數(shù)據(jù)庫
注：調(diào)度器之間不能共享任務(wù)存儲(chǔ)器
執(zhí)行器 executors：用于執(zhí)行任務(wù)，可設(shè)定執(zhí)行模式
將指定的作業(yè)提交到線程池或者進(jìn)程池中運(yùn)行，任務(wù)完成通知調(diào)度器觸發(fā)相應(yīng)的事件。
調(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ù)會(huì)阻塞當(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 后臺(tái)調(diào)度器

當(dāng)前線程不會(huì)阻塞，調(diào)度器后臺(tái)執(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á)式	參數(shù)類型	描述
*	所有	通配符。例： minutes=* 即每分鐘觸發(fā)
*/a	所有	可被a整除的通配符。
a-b	所有	范圍a-b觸發(fā)
a-b/c	所有	范圍a-b，且可被c整除時(shí)觸發(fā)
xth y	日	第幾個(gè)星期幾觸發(fā)。x為第幾個(gè)，y為星期幾
last x	日	一個(gè)月中，最后個(gè)星期幾觸發(fā)
last	日	一個(gè)月最后一天觸發(fā)
x,y,z	所有	組合表達(dá)式，可以組合確定值或上方的表達(dá)式

注：當(dāng)設(shè)置的時(shí)間間隔小于，任務(wù)的執(zhí)行時(shí)間，線程會(huì)阻塞住，等待執(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ù)存儲(chǔ)器為 memoryjobstore ,如果想自己配置的話可以執(zhí)行以下操作

需求：

兩個(gè)任務(wù)儲(chǔ)存器分別搭配兩個(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)

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

除了 BlockingScheduler 外，其他非阻塞的調(diào)度器都會(huì)立即返回，運(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ù)會(huì)執(zhí)行一次
    coalesce：false，恢復(fù)后的任務(wù)會(huì)執(zhí)行N次配合misfire_grace_time使用
    misfire_grace_time設(shè)置時(shí)間差值，由于某些原因沒有運(yùn)行，再次提交時(shí)，大于設(shè)置的時(shí)間，實(shí)例不會(huì)運(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級(jí)別，這樣就能捕獲異常信息
import logging
logging.basicConfig()
logging.getLogger('apscheduler').setLevel(logging.DEBUG)

總結(jié)

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

您可能感興趣的文章: