使用背景

實(shí)際項(xiàng)目中，需要驗(yàn)證打點(diǎn)數(shù)據(jù)在各個(gè)系統(tǒng)中收集是否一致，而部分節(jié)點(diǎn)打點(diǎn)數(shù)據(jù)收集是通過異步任務(wù)實(shí)現(xiàn)的，等待時(shí)間比較久。為應(yīng)對業(yè)務(wù)異步操作處理，實(shí)現(xiàn)異步數(shù)據(jù)的收集，經(jīng)過調(diào)研后，選擇了 APScheduler 框架。

什么是 APScheduler 框架？

APScheduler 是基于 Quartz(一個(gè)功能豐富的開源任務(wù)調(diào)度系統(tǒng)) 的一個(gè) Python 定時(shí)任務(wù)框架，使用起來簡單且方便，提供了基于日期、固定時(shí)間間隔以及 crontab 類型的任務(wù)，并且可以持久化任務(wù)，基于這些功能可以快速實(shí)現(xiàn) python 的定時(shí)輪詢?nèi)蝿?wù)系統(tǒng)。

使用 APScheduler 框架，可以通過 pip 安裝

pip install apscheduler

APScheduler 框架包含四個(gè)組成部分

觸發(fā)器 (trigger)觸發(fā)器包含調(diào)度邏輯，每一個(gè)作業(yè)有它自己的觸發(fā)器，用于決定接下來哪一個(gè)作業(yè)會運(yùn)行；除了他們自己初始的配置，觸發(fā)器本身是無狀態(tài)的。

作業(yè)存儲 (job store)作業(yè)存儲存儲被調(diào)度的作業(yè)，默認(rèn)的作業(yè)存儲是簡單的將作業(yè)保存到內(nèi)存中，如果選擇其他方式也可以將作業(yè)保存到數(shù)據(jù)庫中；一個(gè)作業(yè)數(shù)據(jù)的保存將會在持久化作業(yè)存儲的時(shí)候被序列化，然后在加載時(shí)被反序列化；調(diào)度器無法分享同一個(gè)作業(yè)存儲。

執(zhí)行器 (executor)執(zhí)行器處理作業(yè)的運(yùn)行，一般通過在作業(yè)中提交制定好的可調(diào)用對象到一個(gè)線程中或者線程池中來執(zhí)行；在作業(yè)完成時(shí)，執(zhí)行器會去通知調(diào)度器。

調(diào)度器 (scheduler)調(diào)度器是 APScheduler 的核心，所有相關(guān)的組件都要通過它來定義，已配置好的任務(wù)也是要通過它來調(diào)度。

APScheduler 在 flask 中使用

因?yàn)?scheduler 任務(wù)需要耗費(fèi)較多時(shí)間，因此當(dāng)在項(xiàng)目中收到 flask 的接口請求后，可以通過線程異步處理耗時(shí)任務(wù)，先將 “正在處理” 作為接口結(jié)果返回，
示例代碼結(jié)構(gòu)如下：

編寫任務(wù)函數(shù)，開始 APScheduler 的調(diào)度

在通過 flask 接口拿到需要的任務(wù)參數(shù)后，便可以創(chuàng)建調(diào)度任務(wù)。在創(chuàng)建調(diào)度任務(wù)之前，我們需要確定要選擇哪一種調(diào)度器、job 存儲、執(zhí)行器和觸發(fā)器，
調(diào)度器的選擇主要基于編程環(huán)境以及 APScheduler 的用途，

這里我們根據(jù)需要選擇 BackgroundScheduler。

在 job 存儲的選擇上，需要根據(jù)自己的 job 是否需要持久化，因?yàn)闆]有特殊的需求，所以使用默認(rèn)的內(nèi)存方式

執(zhí)行器的選擇需要依據(jù) job 的類型，默認(rèn)的線程池執(zhí)行器apscheduler.executors.pool.ThreadPoolExecutor 已經(jīng)可以滿足大多數(shù)情況。

管理 job 的調(diào)度方式需要選擇一個(gè)合適的觸發(fā)器，APScheduler 內(nèi)置三種觸發(fā)器；

因?yàn)槲覀兊淖詣踊枰獙Ω鱾€(gè)子環(huán)節(jié)進(jìn)行驗(yàn)證，當(dāng)上一個(gè)環(huán)節(jié)成功后才能進(jìn)行下一個(gè)環(huán)節(jié)的驗(yàn)證，因此選擇 apscheduler.triggers.interval，以固定的時(shí)間間隔運(yùn)行 job。

部分項(xiàng)目代碼

periodic_task 是項(xiàng)目中的任務(wù)調(diào)度函數(shù)；首先實(shí)例化了一個(gè) BackgroundScheduler 調(diào)度器，接著向調(diào)度器添加 job，添加的 job 為 data_task 函數(shù)，同時(shí)定義了 job 的觸發(fā)器，指定固定的時(shí)間間隔為 58 秒。

其中 data_task 描述了具體的 job 細(xì)節(jié)，即分別判斷當(dāng)前不同的任務(wù)節(jié)點(diǎn)執(zhí)行相應(yīng)的驗(yàn)證過程，并將每一步的驗(yàn)證狀態(tài)記錄到數(shù)據(jù)庫中，這樣在下一次執(zhí)行 data_task 時(shí)，就可以去校驗(yàn)新的環(huán)節(jié)；

啟動調(diào)度器使用 start 函數(shù)，結(jié)束調(diào)度器使用 shutdown 函數(shù)；

shutdown 函數(shù)可以指定停止條件，在本項(xiàng)目中，因?yàn)椴襟E比較多，一旦有環(huán)節(jié)出錯(cuò)，就需要結(jié)束任務(wù)，保存已驗(yàn)證的環(huán)節(jié)，因此在拿到任務(wù)結(jié)果時(shí)，不論是整個(gè)驗(yàn)證成功的結(jié)果，還是某個(gè)環(huán)節(jié)出錯(cuò)的結(jié)果，都會停止本次調(diào)度，結(jié)束掉本次驗(yàn)證。

總結(jié)

• APScheduler 在 flask 中使用需要用到線程池異步去處理耗時(shí)任務(wù)；
• 使用 APScheduler 需要配置好合適的調(diào)度器、job 存儲、執(zhí)行器和觸發(fā)器；
• 在業(yè)務(wù)中驗(yàn)證復(fù)雜連續(xù)的步驟可以使用輪詢的方式，并設(shè)置好任務(wù)結(jié)束的條件，不僅可以校驗(yàn)每一步的驗(yàn)證結(jié)果而且有環(huán)節(jié)出錯(cuò)也不影響整個(gè)流程。

以上就是python自動化測試中APScheduler Flask的應(yīng)用示例的詳細(xì)內(nèi)容，更多關(guān)于python自動化測試APScheduler Flask的資料請關(guān)注腳本之家其它相關(guān)文章！