使用背景

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

什么是 APScheduler 框架？

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

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

pip install apscheduler

APScheduler 框架包含四個組成部分

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

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

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

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

APScheduler 在 flask 中使用

因為 scheduler 任務(wù)需要耗費較多時間，因此當在項目中收到 flask 的接口請求后，可以通過線程異步處理耗時任務(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 是否需要持久化，因為沒有特殊的需求，所以使用默認的內(nèi)存方式

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

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

因為我們的自動化需要對各個子環(huán)節(jié)進行驗證，當上一個環(huán)節(jié)成功后才能進行下一個環(huán)節(jié)的驗證，因此選擇 apscheduler.triggers.interval，以固定的時間間隔運行 job。

部分項目代碼

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

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

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

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

總結(jié)

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

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

最新評論