目錄
使用背景
實際項目中，需要驗證打點數(shù)據(jù)在各個系統(tǒng)中收集是否一致，而部分節(jié)點打點數(shù)據(jù)收集是通過異步任務實現(xiàn)得，等待時間比較久。為應對業(yè)務異步操作處理，實現(xiàn)異步數(shù)據(jù)得收集，經(jīng)過調研后，選擇了 APScheduler 框架。
什么是 APScheduler 框架？
APScheduler 是基于 Quartz(一個功能豐富得開源任務調度系統(tǒng)) 得一個 Python 定時任務框架，使用起來簡單且方便，提供了基于日期、固定時間間隔以及 crontab 類型得任務，并且可以持久化任務，基于這些功能可以快速實現(xiàn) python 得定時輪詢任務系統(tǒng)。
使用 APScheduler 框架，可以通過 pip 安裝
pip install apscheduler
APScheduler 框架包含四個組成部分
觸發(fā)器 (trigger)觸發(fā)器包含調度邏輯，每一個作業(yè)有它自己得觸發(fā)器，用于決定接下來哪一個作業(yè)會運行；除了他們自己初始得配置，觸發(fā)器本身是無狀態(tài)得。
作業(yè)存儲 (job store)作業(yè)存儲存儲被調度得作業(yè)，默認得作業(yè)存儲是簡單得將作業(yè)保存到內存中，如果選擇其他方式也可以將作業(yè)保存到數(shù)據(jù)庫中；一個作業(yè)數(shù)據(jù)得保存將會在持久化作業(yè)存儲得時候被序列化，然后在加載時被反序列化；調度器無法分享同一個作業(yè)存儲。
執(zhí)行器 (executor)執(zhí)行器處理作業(yè)得運行，一般通過在作業(yè)中提交制定好得可調用對象到一個線程中或者線程池中來執(zhí)行；在作業(yè)完成時，執(zhí)行器會去通知調度器。
調度器 (scheduler)調度器是 APScheduler 得核心，所有相關得組件都要通過它來定義，已配置好得任務也是要通過它來調度。
APScheduler 在 flask 中使用
因為 scheduler 任務需要耗費較多時間，因此當在項目中收到 flask 得接口請求后，可以通過線程異步處理耗時任務，先將 “正在處理” 作為接口結果返回，
示例代碼結構如下：
編寫任務函數(shù)，開始 APScheduler 得調度
在通過 flask 接口拿到需要得任務參數(shù)后，便可以創(chuàng)建調度任務。在創(chuàng)建調度任務之前，我們需要確定要選擇哪一種調度器、job 存儲、執(zhí)行器和觸發(fā)器，
調度器得選擇主要基于編程環(huán)境以及 APScheduler 得用途，
這里我們根據(jù)需要選擇 BackgroundScheduler。
在 job 存儲得選擇上，需要根據(jù)自己得 job 是否需要持久化，因為沒有特殊得需求，所以使用默認得內存方式
執(zhí)行器得選擇需要依據(jù) job 得類型，默認得線程池執(zhí)行器apscheduler.executors.pool.ThreadPoolExecutor 已經(jīng)可以滿足大多數(shù)情況。
管理 job 得調度方式需要選擇一個合適得觸發(fā)器，APScheduler 內置三種觸發(fā)器；
因為我們得自動化需要對各個子環(huán)節(jié)進行驗證，當上一個環(huán)節(jié)成功后才能進行下一個環(huán)節(jié)得驗證，因此選擇 apscheduler.triggers.interval，以固定得時間間隔運行 job。
部分項目代碼
periodic_task 是項目中得任務調度函數(shù)；首先實例化了一個 BackgroundScheduler 調度器，接著向調度器添加 job，添加得 job 為 data_task 函數(shù)，同時定義了 job 得觸發(fā)器，指定固定得時間間隔為 58 秒。
其中 data_task 描述了具體得 job 細節(jié)，即分別判斷當前不同得任務節(jié)點執(zhí)行相應得驗證過程，并將每一步得驗證狀態(tài)記錄到數(shù)據(jù)庫中，這樣在下一次執(zhí)行 data_task 時，就可以去校驗新得環(huán)節(jié)；
啟動調度器使用 start 函數(shù)，結束調度器使用 shutdown 函數(shù)；
shutdown 函數(shù)可以指定停止條件，在本項目中，因為步驟比較多，一旦有環(huán)節(jié)出錯，就需要結束任務，保存已驗證得環(huán)節(jié)，因此在拿到任務結果時，不論是整個驗證成功得結果，還是某個環(huán)節(jié)出錯得結果，都會停止本次調度，結束掉本次驗證。
總結
- APScheduler 在 flask 中使用需要用到線程池異步去處理耗時任務；
- 使用 APScheduler 需要配置好合適得調度器、job 存儲、執(zhí)行器和觸發(fā)器；
- 在業(yè)務中驗證復雜連續(xù)得步驟可以使用輪詢得方式，并設置好任務結束得條件，不僅可以校驗每一步得驗證結果而且有環(huán)節(jié)出錯也不影響整個流程。
以上就是python自動化測試中APScheduler Flask得應用示例得詳細內容，更多關于python自動化測試APScheduler Flask得資料請關注之家其它相關內容！