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