测试平台系列(72)|测试平台系列(72) 了解ApScheduler基本用法测试平台系列(72)了解ApSchedul

大家好~我是米洛！

我正在从0到1打造一个开源的接口测试平台, 也在编写一套与之对应的完整教程，希望大家多多支持。

欢迎关注我的公众号测试开发坑货，获取最新文章教程!

回顾
上一节我们调研了一下市面上的定时任务方案，最终确定为APScheduler，但据说在uvicorn下还有一些坑。
没关系，笔者也是在摸索阶段。如果有遇到问题，解决了也可以给大家参考。
这篇主要给大家介绍一下APScheduler的基本信息和使用方法。
四个名词

文章图片

触发器: 定时任务什么时候触发
工作商店: 定时任务存在哪里，内存？MySQL?Redis?
执行器: 执行任务的程序
调度程序: 不同任务之间的调度管理程序，还包括任务的添加和修改等

存储介质
在上一节我们提到了定时任务持久化。
那么在APScheduler中支持哪些存储介质呢？看一张图:

文章图片

【测试平台系列(72)|测试平台系列(72) 了解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结合起来，完成一整套定时任务功能。