大家好~我是
米洛！
我正在從0到1打造一個開源的接口測試平台, 也在編寫一套與之對應的完整教程，希望大家多多支持。
歡迎關注我的公眾號米洛的測開日記，獲取最新文章教程!
回顧
上一節我們調研了一下市面上的定時任務方案，最終確定為APScheduler，但據說在uvicorn下還有一些坑。
沒關系，筆者也是在摸索階段。如果有遇到問題，解決了也可以給大家參考。
這篇主要給大家介紹一下APScheduler的基本信息和使用方法。
四個名詞
- 觸發器: 定時任務什么時候觸發
- 工作商店: 定時任務存在哪里，內存？MySQL?Redis?
- 執行器: 執行任務的程序
- 調度程序: 不同任務之間的調度管理程序，還包括任務的添加和修改等
存儲介質
在上一節我們提到了定時任務持久化。
那么在APScheduler中支持哪些存儲介質呢？看一張圖:
可以看到，它支持多種介質，包括我們最不願意用的memory(內存)。因為如果服務重啟，數據就都沒了，所以我們不考慮使用這個。
-
SQLAlchemy
這是我們常用的方法，通過它可以寫入到mysql，在不引入其他組件的情況下完全可以考慮。
-
MongoDB
NoSQL的一種，比較方便，我們
秉着不額外增加組件的信念，暫不考慮。 -
Redis
redis后續我們自身服務也需要用到，但把它作為一個"數據庫"來使用，還得考慮redis備份等情況，當然優點就是會比mysql更快。
至於其他的zk，rethinkdb，由於我們暫時也不會引用，所以不分析了。
綜合而言，還是用SQLAlchemy
代價最小，成本最低。
運行模式
在運行demo的過程中，發現APScheduler很強大，可以支持各種並發模式。
由於我們是異步Asycio，所以我們選擇第三種執行器。其他的包括Gevent等也是Flask/Django等框架的好選擇。
了解基本API
- 配置scheduler
from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
from apscheduler.schedulers.asyncio import AsyncIOScheduler
# 選擇job存儲介質為sqlalchemy
job_store = {
'default': SQLAlchemyJobStore(url=Config.SQLALCHEMY_DATABASE_URI)
}
# 選擇調度程序為AsyncIOScheduler
scheduler = AsyncIOScheduler()
# 配置好對應的程序
scheduler.configure(jobstores=job_store)
# 啟動
scheduler.start()
- 添加job
# 把方法: myfunc添加到定時任務，每2分鍾執行一次
scheduler.add_job(myfunc, 'interval', minutes=2, id='my_job_id')
與pity結合
在FastApi中，有個startup的鈎子方法，意味着當服務啟動的時候會自動執行該方法。我們來看看怎么用的:
@pity.on_event意思是監聽事件，后面的參數: "startup"代表pity服務啟動。
也就是說，當pity啟動的時候，則初始化Scheduler。
這里我封裝了一個Scheduler類，后面會給大家介紹里面的具體方法。
下一節我們編寫測試計划相關接口，並與APScheduler結合起來，完成一整套定時任務功能。