介紹

APScheduler的全稱是Advanced Python Scheduler。它是一個輕量級的 Python 定時任務(wù)調(diào)度框架。APScheduler 支持三種調(diào)度任務(wù)：固定時間間隔，固定時間點(diǎn)（日期），Linux 下的 Crontab 命令。同時，它還支持異步執(zhí)行、后臺執(zhí)行調(diào)度任務(wù)。

APScheduler基于Quartz的一個Python定時任務(wù)框架，實(shí)現(xiàn)了Quartz的所有功能，使用起來十分方便。

安裝

pip install apscheduler

官方地址

https://apscheduler.readthedocs.io/en/latest/userguide.html#starting-the-scheduler

基本概念

1.APScheduler四大組件

• 觸發(fā)器 triggers ：用于設(shè)定觸發(fā)任務(wù)的條件

• 任務(wù)儲存器 job stores：用于存放任務(wù)，把任務(wù)存放在內(nèi)存或數(shù)據(jù)庫中

• 執(zhí)行器 executors： 用于執(zhí)行任務(wù)，可以設(shè)定執(zhí)行模式為單線程或線程池

• 調(diào)度器 schedulers： 把上方三個組件作為參數(shù)，通過創(chuàng)建調(diào)度器實(shí)例來運(yùn)行

1.1 觸發(fā)器triggers

觸發(fā)器包含調(diào)度邏輯。每個任務(wù)都有自己的觸發(fā)器，用于確定何時應(yīng)該運(yùn)行作業(yè)。除了初始配置之外，觸發(fā)器完全是無狀態(tài)的。

1.2 任務(wù)儲存器 job stores

默認(rèn)情況下，任務(wù)存放在內(nèi)存中。也可以配置存放在不同類型的數(shù)據(jù)庫中。如果任務(wù)存放在數(shù)據(jù)庫中，那么任務(wù)的存取有一個序列化和反序列化的過程，同時修改和搜索任務(wù)的功能也是由任務(wù)儲存器實(shí)現(xiàn)。

注意一個任務(wù)儲存器不要共享給多個調(diào)度器，否則會導(dǎo)致狀態(tài)混亂

1.3 執(zhí)行器 executors

任務(wù)會被執(zhí)行器放入線程池或進(jìn)程池去執(zhí)行，執(zhí)行完畢后，執(zhí)行器會通知調(diào)度器。

1.4 調(diào)度器 schedulers

一個調(diào)度器由上方三個組件構(gòu)成，一般來說，一個程序只要有一個調(diào)度器就可以了。開發(fā)者也不必直接操作任務(wù)儲存器、執(zhí)行器以及觸發(fā)器，因?yàn)檎{(diào)度器提供了統(tǒng)一的接口，通過調(diào)度器就可以操作組件，比如任務(wù)的增刪改查。

調(diào)度器工作流程：

2. 調(diào)度器組件詳解

根據(jù)開發(fā)需求選擇相應(yīng)的組件，下面是不同的調(diào)度器組件：

• BlockingScheduler 阻塞式調(diào)度器：適用于只跑調(diào)度器的程序。
• BackgroundScheduler 后臺調(diào)度器：適用于非阻塞的情況，調(diào)度器會在后臺獨(dú)立運(yùn)行。
• AsyncIOScheduler AsyncIO調(diào)度器，適用于應(yīng)用使用AsnycIO的情況。
• GeventScheduler Gevent調(diào)度器，適用于應(yīng)用通過Gevent的情況。
• 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)用。

2.1 任務(wù)儲存器的選擇

要看任務(wù)是否需要持久化。如果你運(yùn)行的任務(wù)是無狀態(tài)的，選擇默認(rèn)任務(wù)儲存器MemoryJobStore就可以應(yīng)付。但是，如果你需要在程序關(guān)閉或重啟時，保存任務(wù)的狀態(tài)，那么就要選擇持久化的任務(wù)儲存器。如果，作者推薦使用SQLAlchemyJobStore并搭配PostgreSQL作為后臺數(shù)據(jù)庫。這個方案可以提供強(qiáng)大的數(shù)據(jù)整合與保護(hù)功能。

2.2 執(zhí)行器的選擇

同樣要看你的實(shí)際需求。默認(rèn)的ThreadPoolExecutor線程池執(zhí)行器方案可以滿足大部分需求。如果，你的程序是計(jì)算密集型的，那么最好用ProcessPoolExecutor進(jìn)程池執(zhí)行器方案來充分利用多核算力。也可以將ProcessPoolExecutor作為第二執(zhí)行器，混合使用兩種不同的執(zhí)行器。

配置一個任務(wù)，就要設(shè)置一個任務(wù)觸發(fā)器。觸發(fā)器可以設(shè)定任務(wù)運(yùn)行的周期、次數(shù)和時間。

3. APScheduler有三種內(nèi)置的觸發(fā)器

• date 日期：觸發(fā)任務(wù)運(yùn)行的具體日期
• interval 間隔：觸發(fā)任務(wù)運(yùn)行的時間間隔
• cron 周期：觸發(fā)任務(wù)運(yùn)行的周期
• calendarinterval：當(dāng)您想要在一天中的特定時間以日歷為基礎(chǔ)的間隔運(yùn)行任務(wù)時使用

一個任務(wù)也可以設(shè)定多種觸發(fā)器，比如，可以設(shè)定同時滿足所有觸發(fā)器條件而觸發(fā)，或者滿足一項(xiàng)即觸發(fā)。

3.0 觸發(fā)器代碼示例

date 是最基本的一種調(diào)度，作業(yè)任務(wù)只會執(zhí)行一次。它表示特定的時間點(diǎn)觸發(fā)。它的參數(shù)如下：

• run_date(datetime or str):任務(wù)運(yùn)行的日期或者時間
• timezone(datetime.tzinfo or str):指定時區(qū)

from datetime import date from apscheduler.schedulers.blocking import BlockingSchedulerscheduler = BlockingScheduler() def my_job(text):print(text)# 注意：run_date參數(shù)可以是date類型、datetime類型或文本類型。 # 在2019年4月15日執(zhí)行 scheduler.add_job(my_job, 'date', run_date=date(2019, 4, 15), args=['測試任務(wù)']) # datetime類型（用于精確時間） # scheduler.add_job(my_job, 'date', run_date=datetime(2019, 4, 15, 17, 30, 5), args=['測試任務(wù)']) # 字符串 #scheduler.add_job(my_job, 'date', run_date='2009-11-06 16:30:05', args=['測試任務(wù)'])scheduler.start()

3.2 interval 周期觸發(fā)任務(wù)

固定時間間隔觸發(fā)。interval 間隔調(diào)度，參數(shù)如下：

• weeks(int)：間隔幾周
• days(int)：間隔幾天
• hours(int)：間隔幾小時
• minutes(int)：間隔幾分鐘
• seconds(int)：間隔多少秒
• start_date(datetime or str)：開始日期
• end_date(datetime or str)：結(jié)束日期
• timezone(datetime.tzinfo or str)：時區(qū)

from datetime import datetime from apscheduler.schedulers.blocking import BlockingSchedulerdef job_func():print("當(dāng)前時間：", datetime.datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S.%f")scheduler = BlockingScheduler()# 每2小時觸發(fā) scheduler.add_job(job_func, 'interval', hours=2)# 在 2019-04-15 17:00:00 ~ 2019-12-31 24:00:00 之間, 每隔兩分鐘執(zhí)行一次 job_func 方法 scheduler .add_job(job_func, 'interval', minutes=2, start_date='2019-04-15 17:00:00' , end_date='2019-12-31 24:00:00')# jitter振動參數(shù)，給每次觸發(fā)添加一個隨機(jī)浮動秒數(shù)，一般適用于多服務(wù)器，避免同時運(yùn)行造成服務(wù)擁堵。 scheduler.add_job(job_func, 'interval', hours=1, jitter=120)scheduler.start()

3.3 cron 觸發(fā)器

在特定時間周期性地觸發(fā)，和Linux crontab格式兼容。它是功能最強(qiáng)大的觸發(fā)器。

• year(int or str) 年，4位數(shù)字
• month(int or str) 月（范圍1-12）
• day(int or str) 日（范圍1-31）
• week(int or str) 周（范圍1-53）
• day_of_week(int or str) 周內(nèi)第幾天或者星期幾（范圍0-6或者mon,tue,wed,thu,fri,stat,sun）
• hour(int or str) 時（0-23）
• minute(int or str) 分（0-59）
• second(int or str) 秒（0-59）
• start_date(datetime or str) 最早開始日期（含）
• end_date(datetime or str) 最晚結(jié)束日期（含）
• timezone(datetime.tzinfo or str) 指定時區(qū)

表達(dá)式類型

表達(dá)式參數(shù)類型描述

*	所有	通配符。例：minutes=*即每分鐘觸發(fā)
*/a	所有	可被a整除的通配符。
a-b	所有	范圍a-b觸發(fā)
a-b/c	所有	范圍a-b，且可被c整除時觸發(fā)
xth y	日	第幾個星期幾觸發(fā)。x為第幾個，y為星期幾
last x	日	一個月中，最后個星期幾觸發(fā)
last	日	一個月最后一天觸發(fā)
x,y,z	所有	組合表達(dá)式，可以組合確定值或上方的表達(dá)式

注意：month和day_of_week參數(shù)分別接受的是英語縮寫jan– dec 和 mon – sun

import datetime from apscheduler.schedulers.background import BackgroundSchedulerdef job_func(text):print("當(dāng)前時間：", datetime.datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S.%f")[:-3])scheduler = BackgroundScheduler() # 在每年 1-3、7-9 月份中的每個星期一、二中的 00:00, 01:00, 02:00 和 03:00 執(zhí)行 job_func 任務(wù) scheduler.add_job(job_func, 'cron', month='1-3,7-9',day='0, tue', hour='0-3')scheduler.start()

使用 scheduled_job() 裝飾器添加任務(wù)：

@scheduler.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!")

注意：夏令時問題

有些timezone時區(qū)可能會有夏令時的問題。這個可能導(dǎo)致令時切換時，任務(wù)不執(zhí)行或任務(wù)執(zhí)行兩次。避免這個問題，可以使用UTC時間，或提前預(yù)知并規(guī)劃好執(zhí)行的問題。

pri# 在Europe/Helsinki時區(qū), 在三月最后一個周一就不會觸發(fā)；在十月最后一個周一會觸發(fā)兩次 scheduler.add_job(job_function, 'cron', hour=3, minute=30)

4. 配置調(diào)度程序

APScheduler提供了許多不同的方法來配置調(diào)度程序。您可以使用配置字典，也可以將選項(xiàng)作為關(guān)鍵字參數(shù)傳遞。您還可以先實(shí)例化調(diào)度程序，然后添加任務(wù)并配置調(diào)度程序。這樣您就可以在任何環(huán)境中獲得最大的靈活性

可以在BaseScheduler類的API引用中找到調(diào)度程序級別配置選項(xiàng)的完整列表 。調(diào)度程序子類還可以具有其各自API引用中記錄的其他選項(xiàng)。各個任務(wù)存儲和執(zhí)行程序的配置選項(xiàng)同樣可以在其API參考頁面上找到。

假設(shè)您希望在應(yīng)用程序中使用默認(rèn)作業(yè)存儲和默認(rèn)執(zhí)行程序運(yùn)行BackgroundScheduler：

from apscheduler.schedulers.background import BackgroundSchedulerscheduler = BackgroundScheduler()

這將為您提供一個BackgroundScheduler，其MemoryJobStore（內(nèi)存任務(wù)儲存器）名為“default”，ThreadPoolExecutor（線程池執(zhí)行器）名為“default”，默認(rèn)最大線程數(shù)為10。

假如你現(xiàn)在有這樣的需求，兩個任務(wù)儲存器分別搭配兩個執(zhí)行器；同時，還要修改任務(wù)的默認(rèn)參數(shù)；最后還要改時區(qū)?？梢詤⒖枷旅胬?#xff0c;它們是完全等價的。

• 名稱為“mongo”的MongoDBJobStore
• 名稱為“default”的SQLAlchemyJobStore
• 名稱為“ThreadPoolExecutor ”的ThreadPoolExecutor，最大線程20個
• 名稱“processpool”的ProcessPoolExecutor，最大進(jìn)程5個
• UTC時間作為調(diào)度器的時區(qū)
• 默認(rèn)為新任務(wù)關(guān)閉合并模式（）
• 設(shè)置新任務(wù)的默認(rèn)最大實(shí)例數(shù)為3

方法一：

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)

方法二：

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 utcfrom apscheduler.schedulers.background import BackgroundScheduler from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore from apscheduler.executors.pool import ProcessPoolExecutorjobstores = {'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)度器

啟動調(diào)度器是只需調(diào)用start()即可。除了BlockingScheduler，非阻塞調(diào)度器都會立即返回，可以繼續(xù)運(yùn)行之后的代碼，比如添加任務(wù)等。

對于BlockingScheduler，程序則會阻塞在start()位置，所以，要運(yùn)行的代碼必須寫在start()之前。

注意:調(diào)度器啟動后，就不可以修改配置。

5. 添加任務(wù)

添加任務(wù)方法有兩種：

通過調(diào)用add_job()

通過裝飾器scheduled_job()

5.1 利弊：

• 第一種方法是最常用的；第二種方法是最方便的，但缺點(diǎn)就是運(yùn)行時，不能修改任務(wù)。
• 第一種add_job()方法會返回一個apscheduler.job.Job實(shí)例，這樣就可以在運(yùn)行時，修改或刪除任務(wù)。

在任何時候你都能配置任務(wù)。但是如果調(diào)度器還沒有啟動，此時添加任務(wù)，那么任務(wù)就處于一個暫存的狀態(tài)。只有當(dāng)調(diào)度器啟動時，才會開始計(jì)算下次運(yùn)行時間。

還有一點(diǎn)要注意，如果你的執(zhí)行器或任務(wù)儲存器是會序列化任務(wù)的，那么這些任務(wù)就必須符合：

• 回調(diào)函數(shù)必須全局可用
• 回調(diào)函數(shù)參數(shù)必須也是可以被序列化的

重要提醒！

如果在程序初始化時，是從數(shù)據(jù)庫讀取任務(wù)的，那么必須為每個任務(wù)定義一個明確的ID，并且使用replace_existing=True，否則每次重啟程序，你都會得到一份新的任務(wù)拷貝，也就意味著任務(wù)的狀態(tài)不會保存。

內(nèi)置任務(wù)儲存器中，只有MemoryJobStore不會序列化任務(wù)；內(nèi)置執(zhí)行器中，只有ProcessPoolExecutor會序列化任務(wù)。

建議：如果想要立刻運(yùn)行任務(wù)，可以在添加任務(wù)時省略trigger參數(shù)

6. 移除任務(wù)

如果想從調(diào)度器移除一個任務(wù)，那么你就要從相應(yīng)的任務(wù)儲存器中移除它，這樣才算移除了。有兩種方式：

• 調(diào)用remove_job()，參數(shù)為：任務(wù)ID，任務(wù)儲存器名稱
• 在通過add_job()創(chuàng)建的任務(wù)實(shí)例上調(diào)用remove()方法

第二種方式更方便，但前提必須在創(chuàng)建任務(wù)實(shí)例時，實(shí)例被保存在變量中。對于通過scheduled_job()創(chuàng)建的任務(wù)，只能選擇第一種方式。

當(dāng)任務(wù)調(diào)度結(jié)束時（比如，某個任務(wù)的觸發(fā)器不再產(chǎn)生下次運(yùn)行的時間），任務(wù)就會自動移除。

job = scheduler.add_job(myfunc, 'interval', minutes=2) job.remove()# 同樣，通過任務(wù)的具體ID： scheduler.add_job(myfunc, 'interval', minutes=2, id='my_job_id') scheduler.remove_job('my_job_id')

7. 暫停和恢復(fù)任務(wù)

通過任務(wù)實(shí)例或調(diào)度器，就能暫停和恢復(fù)任務(wù)。如果一個任務(wù)被暫停了，那么該任務(wù)的下一次運(yùn)行時間就會被移除。在恢復(fù)任務(wù)前，運(yùn)行次數(shù)計(jì)數(shù)也不會被統(tǒng)計(jì)。

暫停任務(wù)，有以下兩個方法：

• apscheduler.job.Job.pause()
• apscheduler.schedulers.base.BaseScheduler.pause_job()

恢復(fù)任務(wù)

• apscheduler.job.Job.resume()
• apscheduler.schedulers.base.BaseScheduler.resume_job()

8. 獲取任務(wù)列表

通過get_jobs()就可以獲得一個可修改的任務(wù)列表。get_jobs()第二個參數(shù)可以指定任務(wù)儲存器名稱，那么就會獲得對應(yīng)任務(wù)儲存器的任務(wù)列表。

print_jobs()可以快速打印格式化的任務(wù)列表，包含觸發(fā)器，下次運(yùn)行時間等信息。

修改任務(wù)

通過apscheduler.job.Job.modify()或modify_job()，你可以修改任務(wù)當(dāng)中除了id的任何屬性。

比如：

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

如果想要重新調(diào)度任務(wù)（就是改變觸發(fā)器），你能通過apscheduler.job.Job.reschedule()或reschedule_job()來實(shí)現(xiàn)。這些方法會重新創(chuàng)建觸發(fā)器，并重新計(jì)算下次運(yùn)行時間。

比如：

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

9. 關(guān)閉調(diào)度器

scheduler.shutdown()

默認(rèn)情況下，調(diào)度器會先把正在執(zhí)行的任務(wù)處理完，再關(guān)閉任務(wù)儲存器和執(zhí)行器。但是，如果你就直接關(guān)閉，你可以添加參數(shù)：

scheduler.shutdown(wait=False)

上述方法不管有沒有任務(wù)在執(zhí)行，會強(qiáng)制關(guān)閉調(diào)度器。

10. 暫停、恢復(fù)任務(wù)進(jìn)程

# 暫停正在執(zhí)行的任務(wù) scheduler.pause()# 恢復(fù)任務(wù)： scheduler.resume()# 也可以在調(diào)度器啟動時，默認(rèn)所有任務(wù)設(shè)為暫停狀態(tài)。 scheduler.start(paused=True)

總結(jié)

以上是生活随笔為你收集整理的python 第三方模块之 APScheduler - 定时任务的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。