大家好~我是米洛

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

欢迎关注我的公众号米洛的测开日记,获取最新文章教程!

回顾

上一节我们调研了一下市面上的定时任务方案,最终确定为APScheduler,但据说在uvicorn下还有一些坑。

没关系,笔者也是在摸索阶段。如果有遇到问题,解决了也可以给大家参考。

这篇主要给大家介绍一下APScheduler的基本信息和使用方法。

四个名词

测试平台系列(72) 了解ApScheduler基本用法 - 图1

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

存储介质

上一节我们提到了定时任务持久化。

那么在APScheduler中支持哪些存储介质呢?看一张图:

测试平台系列(72) 了解ApScheduler基本用法 - 图2

可以看到,它支持多种介质,包括我们最不愿意用的memory(内存)。因为如果服务重启,数据就都没了,所以我们不考虑使用这个。

  • SQLAlchemy
    这是我们常用的方法,通过它可以写入到mysql,在不引入其他组件的情况下完全可以考虑。
  • MongoDB
    NoSQL的一种,比较方便,我们秉着不额外增加组件的信念,暂不考虑。
  • Redis
    redis后续我们自身服务也需要用到,但把它作为一个”数据库”来使用,还得考虑redis备份等情况,当然优点就是会比mysql更快。
    至于其他的zk,rethinkdb,由于我们暂时也不会引用,所以不分析了。
    综合而言,还是用SQLAlchemy代价最小,成本最低。

运行模式

在运行demo的过程中,发现APScheduler很强大,可以支持各种并发模式。

测试平台系列(72) 了解ApScheduler基本用法 - 图3

由于我们是异步Asycio,所以我们选择第三种执行器。其他的包括Gevent等也是Flask/Django等框架的好选择。

了解基本API

  • 配置scheduler
  1. from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
  2. from apscheduler.schedulers.asyncio import AsyncIOScheduler
  3. # 选择job存储介质为sqlalchemy
  4. job_store = {
  5. 'default': SQLAlchemyJobStore(url=Config.SQLALCHEMY_DATABASE_URI)
  6. }
  7. # 选择调度程序为AsyncIOScheduler
  8. scheduler = AsyncIOScheduler()
  9. # 配置好对应的程序
  10. scheduler.configure(jobstores=job_store)
  11. # 启动
  12. scheduler.start()
  • 添加job
  1. # 把方法: myfunc添加到定时任务,每2分钟执行一次
  2. scheduler.add_job(myfunc, 'interval', minutes=2, id='my_job_id')

与pity结合

在FastApi中,有个startup的钩子方法,意味着当服务启动的时候会自动执行该方法。我们来看看怎么用的:

测试平台系列(72) 了解ApScheduler基本用法 - 图4

@pity.on_event意思是监听事件,后面的参数: “startup”代表pity服务启动。

也就是说,当pity启动的时候,则初始化Scheduler。

这里我封装了一个Scheduler类,后面会给大家介绍里面的具体方法。

下一节我们编写测试计划相关接口,并与APScheduler结合起来,完成一整套定时任务功能。