在我們的日常工作自動化測試當(dāng)中，幾乎超過一半的功能都需要利用定時的任務(wù)來推動觸發(fā)，例如在我們項(xiàng)目中有一個定時監(jiān)控模塊，根據(jù)自己設(shè)置的頻率定時跑測試用例，定時檢測是否存在線上緊急任務(wù)等等，這些都涉及到了有關(guān)定時任務(wù)的問題，很多情況下，大多數(shù)人會選擇window的任務(wù)計劃程序，但如果程序不在window平臺下運(yùn)行，就不能定時啟動了；當(dāng)然也可利用time模塊的time.sleep()方法使程序休眠來達(dá)到定時任務(wù)的目的，但定時任務(wù)多了，代碼可能看起來不太那么友好且有很大的局限性，因此，此時的 Apscheduler 框架是你的不二選擇。

Apscheduler

Apscheduler基于Quartz的一個python定時任務(wù)框架，實(shí)現(xiàn)Quart的所有功能，相關(guān)的接口調(diào)用起來比較方便，目前其提供了基于日期、固定時間間隔以及corntab類型的任務(wù)，并且同時可進(jìn)行持久化任務(wù)；同時它提供了多種不同的調(diào)用器，方便開發(fā)者根據(jù)自己的需求進(jìn)行使用，也方便與數(shù)據(jù)庫等第三方的外部持久化儲存機(jī)制進(jìn)行協(xié)同工作，非常強(qiáng)大。

基本原理

總的來說，主要是利用python threading Event和Lock鎖來寫的。scheduler在主循環(huán)(main_loop)中， 反復(fù)檢查是否有需要執(zhí)行的任務(wù)，完成任務(wù)的檢查函數(shù)為 _process_jobs，主要有那個幾個步驟：

1、 詢問儲存的每個 jobStore ，是否有到期要執(zhí)行的任務(wù)。

2、 due_jobs 不為空，則計算這些jobs中每個job需要運(yùn)行的時間點(diǎn)，時間一到就提交給submit作任務(wù)調(diào)度。

3、在主循環(huán)中，如果不間斷地調(diào)用，而實(shí)際上沒有要執(zhí)行的job，這會造成資源浪費(fèi)。因此在程序中，如果每次掉用 _process_jobs 后，進(jìn)行了預(yù)先判斷，判斷下一次要執(zhí)行的job（離現(xiàn)在最近的）還要多長時間，作為返回值告訴main_loop, 這時主循環(huán)就可以去睡一覺，等大約這么長時間后再喚醒，執(zhí)行下一次 _process_jobs 。

安裝

1､可以直接使用pip進(jìn)行安裝

2､源碼安裝

### 基礎(chǔ)概念

在Apscheduler中主要有以下幾個非常重要的概念，主要如下：

觸發(fā)器（trigger）：

某一個工作到來時引發(fā)的事件，包含調(diào)度的邏輯，每一個作業(yè)都有它自己的觸發(fā)器，用于決定哪個作業(yè)任務(wù)會執(zhí)行，除了它們初始化配置之外，其完全是無狀態(tài)的?？偟膩碚f就是 一個任務(wù)應(yīng)該在什么時候執(zhí)行

執(zhí)行器（executor）：

主要是處理作業(yè)的運(yùn)行，它將要執(zhí)行的作業(yè)放在新的線程或者線程池中運(yùn)行。執(zhí)行完畢之后，再通知調(diào)度器?；诰€程池的操作，可以針對不同類型的作業(yè)任務(wù)，更為高效的使用CPU的計算資源。

作業(yè)存儲（job stores）

保存要調(diào)度的任務(wù)，其中除了默認(rèn)的作業(yè)存儲是把作業(yè)保存在內(nèi)存中，其他的作業(yè)存儲是將作業(yè)保存在數(shù)據(jù)庫中。一個作業(yè)的數(shù)據(jù)將在保存在持久化的作業(yè)存儲之前，會對作業(yè)執(zhí)行序列化操作，當(dāng)重新讀取作業(yè)時，再執(zhí)行反序列化操作。同時，調(diào)度器不能分享同一個作業(yè)存儲。作業(yè)存儲支持主流的存儲機(jī)制：如redis，mongodb，關(guān)系型數(shù)據(jù)庫，內(nèi)存等等。

調(diào)度器（scheduler）：

負(fù)責(zé)將上面幾個組件聯(lián)系在一起，一般在應(yīng)用中只有一個調(diào)度器，程序開發(fā)者不會直接操作觸發(fā)器、作業(yè)存儲或執(zhí)行器，而是利用調(diào)度器提供了處理這些合適的接口，作業(yè)存儲和執(zhí)行器的配置都是通過在調(diào)度器中完成的。

在我們的使用過程中，選擇合適的 調(diào)度器 是根據(jù)我們的開發(fā)環(huán)境以及實(shí)際應(yīng)用來決定的，根據(jù)IO模型的不同，主要有下面一些常見的調(diào)度器：

• BlockingScheduler：適合于只在進(jìn)程中運(yùn)行單個任務(wù)的情況
• BackgroundScheduler: 適合于不運(yùn)行使用其他框架時，并希望在程序后臺執(zhí)行的情況
• AsyncIOScheduler：適合于使用asyncio框架的情況
• GeventScheduler: 適合于使用gevent框架的情況
• TornadoScheduler: 適合于使用Tornado框架的應(yīng)用
• TwistedScheduler: 適合使用Twisted框架的應(yīng)用
• QtScheduler: 適合使用QT的情況

而對于 作業(yè)存儲 ，如果是非持久性作業(yè)，使用默認(rèn)的 MemoryStore 就行了，若是持久性任務(wù)，那么就需要根據(jù)應(yīng)用環(huán)境來進(jìn)行選擇。

大多數(shù)情況下， 執(zhí)行器 選擇 ThreadPoolExecutor 就夠用了，但如果涉及到比較消耗CPU的作業(yè)，就可以選擇ProcessPoolExecutor* ，以充分利用多核CPU。當(dāng)然也可以同時配置使用兩個執(zhí)行器，將進(jìn)程池 ProcessPoolExecutor 調(diào)度器作為你的第二個執(zhí)行器。

配置調(diào)度器

Apscheduler框架提供了許多調(diào)度器的配置方法，既可以使用配置字典，也可以直接傳遞配置參數(shù)給調(diào)度器使用； 同時支持先初始化調(diào)度器，添加完作業(yè)任務(wù)后，再來配置調(diào)度器等。

說了這么多，我們可以來先舉個簡單的例子：

上面的代碼生成一個默認(rèn)的調(diào)度器，默認(rèn)使用名為 default 的 MemoryJobStore，以及使用默認(rèn)名為 default 的 ThreadPoolExecutor ，最大線程數(shù)為10 。

下面進(jìn)行一個復(fù)雜的配置，同時使用兩個作業(yè)存儲和兩個執(zhí)行器，在這個配置中，修改默認(rèn)的配置參數(shù)，jobstored指的是job持久化，默認(rèn)job運(yùn)行在內(nèi)存中，可持久化在數(shù)據(jù)庫，指定為mongo的MongoDBJobStore或者是使用sqlite的SQLAlchemyJobStore，同時可指定多種jobstore。

coalesce ：當(dāng)由于某種原因?qū)е履硞€job積攢了好幾次沒有實(shí)際運(yùn)行（比如說系統(tǒng)掛了5分鐘后恢復(fù)，有一個任務(wù)是每分鐘跑一次的，按道理說這5分鐘內(nèi)本來是“計劃”運(yùn)行5次的，但實(shí)際沒有執(zhí)行），如果coalesce為True，下次這個job被submit給executor時，只會執(zhí)行1次，也就是最后這次，如果為False，那么會執(zhí)行5次（不一定，因?yàn)檫€有其他條件，看后面misfiregracetime的解釋）。

max_instance ：每個job在同一時刻能夠運(yùn)行的最大實(shí)例數(shù),默認(rèn)情況下為1個,可以指定為更大值,這樣即使上個job還沒運(yùn)行完同一個job又被調(diào)度的話也能夠再開一個線程執(zhí)行。

misfire_grace_time ：單位為秒,假設(shè)有這么一種情況,當(dāng)某一job被調(diào)度時剛好線程池都被占滿,調(diào)度器會選擇將該job排隊(duì)不運(yùn)行,misfiregracetime參數(shù)則是在線程池有可用線程時會比對該job的應(yīng)調(diào)度時間跟當(dāng)前時間的差值,如果差值<misfiregracetime時,調(diào)度器會再次調(diào)度該job.反之該job的執(zhí)行狀態(tài)為EVENTJOBMISSED了,即錯過運(yùn)行.</misfire。

啟動/關(guān)閉調(diào)度器

使用 start() 方法來啟動調(diào)度器，其中須注意的是 BlockingScheduler 需要在初始化之后才能執(zhí)行 start() ，對于其他的調(diào)度器，調(diào)用 start() 方法都會直接返回，然后可以繼續(xù)執(zhí)行后面的初始化操作。同時，調(diào)度器啟動之后，就不能再更改它的配置了。

在默認(rèn)情況下，調(diào)度器會等所有的作業(yè)任務(wù)完成后，自動關(guān)閉所有的調(diào)度器及作業(yè)存儲。若在使用過程中不想等待，可以將 wait 參數(shù)選項(xiàng)設(shè)為 False ，則表示直接關(guān)閉：

調(diào)度器監(jiān)聽事件

可以給調(diào)度器添加事件監(jiān)聽器，調(diào)度器事件只有在某些情況下才會被觸發(fā)，并且可以攜帶某些有用的信息。通過給 add_listener() 傳遞合適的 mask 參數(shù)，可以只監(jiān)聽幾種特定的事件類型，具體類型可看源碼中的 event.exception 或者 event.code 值來做識別判斷。

作業(yè)及作業(yè)存儲

jobstore提供給scheduler一個序列化jobs的統(tǒng)一抽象，提供對scheduler中job的增刪改查接口，根據(jù)存儲backend的不同，分以下幾種：

MemoryJobStore ：沒有序列化，jobs就存在內(nèi)存里，增刪改查也都是在內(nèi)存中操作

SQLAlchemyJobStore ：所有sqlalchemy支持的數(shù)據(jù)庫都可以做為backend，增刪改查操作轉(zhuǎn)化為對應(yīng)backend的sql語句

MongoDBJobStore ：用mongodb作backend

RedisJobStore : 用redis作backend

Job是框架承接目前需要執(zhí)行的工作和任務(wù)，我們可以在系統(tǒng)運(yùn)行過程中進(jìn)行動態(tài)的增加、修改、刪除、查詢等操作。

1、添加作業(yè)

上面是通過 add_job() 來添加作業(yè)，另外還有一種方式是通過修飾器 scheduled_job 來動態(tài)裝飾 Job 的實(shí)際函數(shù)

2、移除作業(yè)

3、暫停作業(yè)

4、恢復(fù)作業(yè)

5、修改作業(yè)

6、獲取Job列表

獲得調(diào)度作業(yè)的列表，可以使用 get_jobs() 來完成，它會返回所有的job實(shí)例，同時也可使用 print_jobs() 來輸出所有格式化的作業(yè)列表。也可以利用 get_job(任務(wù)ID) 獲取指定任務(wù)的作業(yè)列表

作業(yè)運(yùn)行控制

add_job() 方法的第二個參數(shù)是trigger，它管理著作業(yè)任務(wù)的調(diào)度方式，它可以被設(shè)置為 data 、 interval 、 corn 三種類別。對于不同的設(shè)置類別，對應(yīng)的參數(shù)也有所不同，具體如下：

1、corn 定時調(diào)度，即規(guī)定在某一時刻執(zhí)行

使用例子：

2、interval間隔調(diào)度，即每隔多久執(zhí)行一次

3、data定時調(diào)度，即設(shè)置后作業(yè)只會執(zhí)行一次，是最基本的調(diào)度模式

總結(jié)

Apscheduler是一個非常強(qiáng)大且易用的類庫，可以方便我們快速的搭建一些強(qiáng)大的定時任務(wù)或者定時監(jiān)控類的調(diào)度系統(tǒng)，在實(shí)際工作中非常有用，同時其也提供了不少的擴(kuò)展點(diǎn)。

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

最新評論