前言
先来说说我们公司Java代码增量覆盖率的实现，大致分为两部分，第一部分定时获取全量测试覆盖率报告，第二部分通过diff操作统计增量代码覆盖率，刚开始，全量覆盖率的获取是脚本形式实现，然后使用crontab设置定时任务来获取全量报告，第二部分是以服务的形势来实现（后端使用Django，前端vue+element ui）,前段时间觉得使用Django太重了，后端使用fastapi框架进行了重构，优化代码的同时将第一部分内容也融入到服务当中。本文就介绍一下，如何在fastapi服务中增加定时任务？
ApScheduler
ApScheduler 是基于 Python 的任务调度库，它允许以各种方式安排和执行任务，从而实现定时执行、周期性执行和异步执行等功能。ApScheduler 提供了多种调度器类型，包括简单调度器、定时调度器和 Cron 调度器，根据具体需求选择适合的调度策略。
简单使用
安装 ApScheduler：
pip install apscheduler
创建调度器和任务函数：
首先，导入必要的模块并创建一个调度器对象和任务函数：
from apscheduler.schedulers.background import BackgroundScheduler
def job():
print("定时任务执行中...")
添加任务和设置触发器：
接下来，将任务添加到调度器中，并设置触发器类型和参数。
scheduler = BackgroundScheduler()
scheduler.add_job(job, 'interval', seconds=10)
代码中我们将任务设置为每隔 10 秒执行一次
启动调度器：
使用 start() 方法启动调度器，开始执行任务：
scheduler.start()
特性
多种触发器类型
ApScheduler 支持多种触发器类型，包括简单触发器、日期触发器和 Cron 触发器，根据不同的需求灵活地触发任务。
- SimpleTrigger（简单触发器） ：指定间隔时间执行任务。
- DateTrigger（日期触发器） ：指定具体的日期和时间执行任务。
- CronTrigger（Cron 触发器） ：使用 Cron 表达式定义任务执行规则。
下面，我们分别看看各个触发器，先写一个公共代码，免得下面重复太多，
def job():
print("定时任务执行中...")
scheduler = BackgroundScheduler()
# 这里将任务与指定的触发器绑定到调度器上。
......
scheduler.start()
# 保持主线程持续运行
try:
while True:
pass
except (KeyboardInterrupt, SystemExit):
scheduler.shutdown()
代码中......就是我们下面要挨个将的触发器使用方法
SimpleTrigger（简单触发器）：
SimpleTrigger 用于指定间隔时间执行任务，适用于周期性执行任务的场景。
案例：每隔30秒执行一次
使用方法一：
from apscheduler.triggers.simple import SimpleTrigger
trigger = SimpleTrigger(seconds=30)
scheduler.add_job(job, trigger=trigger)
使用方法二：
scheduler.add_job(job, 'interval', seconds=30)
DateTrigger（日期触发器）：
DateTrigger 用于指定具体的日期和时间执行任务，适用于单次执行任务的场景。
案例：指定具体时间执行
使用方法一：
from apscheduler.triggers.date import DateTrigger
trigger = DateTrigger(run_date=datetime(2024, 3, 18, 14, 33, 0))
scheduler.add_job(job, trigger=trigger)
使用方法二：
scheduler.add_job(job, 'date', run_date=datetime(2024, 3, 18, 14, 33, 0))
CronTrigger（Cron 触发器）：
CronTrigger 使用 Cron 表达式定义任务执行规则，适用于复杂的定时调度需求。
案例：每分钟执行任务
使用方法一：标准 crontab 表达式
from apscheduler.triggers.cron import CronTrigger
trigger = CronTrigger.from_crontab('* * * * *')
scheduler.add_job(job, trigger=trigger)
使用方法二：
sched.add_job(job, 'cron', minute='*')
CombinedTrigger（组合触发器）：
CombinedTrigger 可以将多个触发器组合在一起，实现更灵活的任务调度策略。
案例：结合 SimpleTrigger 和 DateTrigger，实现每隔 10 秒执行一次任务，同时在指定日期执行任务：
from apscheduler.triggers.combining import AndTrigger
simple_trigger = SimpleTrigger(seconds=10)
date_trigger = DateTrigger(run_date=datetime(2024, 3, 20, 10, 0, 0))
combined_trigger = AndTrigger([simple_trigger, date_trigger])
通过合理选择和配置不同的触发器类型，可以满足各种不同的任务调度需求，实现灵活多样的任务调度方案。
持久化存储
ApScheduler 支持将任务调度信息存储到数据库中，以便在应用程序重启后能够恢复任务状态。
APScheduler提供了多种持久化存储的支持，包括以下几种常见的方式：
- SQLAlchemy 存储：APScheduler允许将任务和调度器状态存储在SQLAlchemy支持的关系型数据库中。这种方式适用于需要使用关系型数据库进行数据存储和管理的场景，如MySQL、PostgreSQL等。
- MongoDB 存储：APScheduler也支持将任务和调度器状态存储在MongoDB中，这种方式适用于对NoSQL数据库有需求的项目。
- Redis 存储：通过将任务和调度器状态存储在Redis中，可以实现快速的内存级别存储和访问，适用于需要高性能的场景。
- 文件系统存储：除了数据库和NoSQL存储外，APScheduler还支持将任务和调度器状态保存在本地文件系统中，适用于简单的应用程序或者测试环境。
持久化存储可以帮助应用程序在重启后恢复先前的任务状态，确保任务不会丢失或重复执行。
多种调度器类型
ApScheduler 提供了多种调度器类型，包括简单调度器、线程池调度器和进程池调度器等，以满足不同的任务执行需求。
- BackgroundScheduler（后台调度器）：这是最常见的调度器类型，用于在后台执行任务。它可以在应用程序启动时创建，并持续运行，负责调度和执行定时任务。
- BlockingScheduler（阻塞调度器）：与BackgroundScheduler类似，但它会阻塞当前线程直到调度器被关闭。适用于简单的脚本或独立应用程序，需要等待所有任务执行完成后再退出。
- AsyncIOScheduler（异步IO调度器）：用于基于asyncio的异步应用程序，支持异步任务执行。适用于需要与异步IO操作集成的应用程序。
- GeventScheduler（Gevent调度器）：用于基于Gevent的协程应用程序，允许在协程中执行任务。适用于需要利用协程实现高并发的应用程序。
- TornadoScheduler（Tornado调度器）：专门为Tornado框架设计的调度器类型，兼容Tornado的事件循环。适用于与Tornado框架集成的应用程序。
不同的调度器类型适用于不同的应用场景，根据具体项目的需求和使用框架的特点来选择合适的调度器类型是很重要的。
可以看到支持异步，很适合在fastapi中使用啊，我们看看异步中如何使用
import asyncio
from apscheduler.schedulers.asyncio import AsyncIOScheduler
async def async_job():
# 异步任务的逻辑
await asyncio.sleep(1)
print("异步任务执行完成")
# 创建AsyncIOScheduler对象
scheduler = AsyncIOScheduler()
# 添加异步任务
scheduler.add_job(async_job, 'interval', seconds=5)
# 启动调度器
scheduler.start()
# 进入事件循环
asyncio.get_event_loop().run_forever()
可以看到，使用AsyncIOScheduler调度器，它允许在异步环境中执行任务。
代码案例
ApScheduler基础使用了解的差不多了，我们在fastapi中接入ApScheduler，实现全量覆盖率统计的定时任务。
核心代码如下：
class Scheduler(object):
scheduler: BackgroundScheduler = None
@staticmethod
def init(scheduler):
Scheduler.scheduler = scheduler
@staticmethod
def configure(**kwargs):
Scheduler.scheduler.configure(**kwargs)
@staticmethod
def start():
Scheduler.scheduler.start()
@staticmethod
def add_coverage_task(task_id, cron):
return Scheduler.scheduler.add_job(
func=Executor.run_coverage_task,
args=(task_id,),
id=str(task_id),
trigger=CronTrigger.from_crontab(cron),
)
@app.on_event("startup")
def init_scheduler():
"""
初始化定时任务
:return:
"""
# SQLAlchemyJobStore指定存储链接
job_store = {
"default": SQLAlchemyJobStore(
url=settings.SQLALCHEMY_DATABASE_URL,
engine_options={"pool_recycle": 1500},
pickle_protocol=3,
)
}
scheduler = BackgroundScheduler()
Scheduler.init(scheduler)
Scheduler.configure(jobstores=job_store)
Scheduler.start()
该函数的主要作用是初始化定时任务调度器，并配置使用SQLAlchemyJobStore进行持久化存储。
简单解释一下这段代码：
-
@app.on_event("startup")：这是FastAPI框架提供的一个装饰器，用于指定在应用启动时执行的初始化操作。 -
def init_scheduler():：定义了一个名为init_scheduler的函数，用于初始化定时任务调度器。 -
job_store = {...}：在这里创建了一个名为job_store的字典，用于配置定时任务调度器的存储方式。其中使用了SQLAlchemyJobStore作为默认的存储，并指定了数据库的连接URL、数据库引擎选项等参数。 -
scheduler = BackgroundScheduler()：创建了一个后台调度器scheduler实例，用于执行定时任务调度。 -
Scheduler.init(scheduler)：这里调用了Scheduler.init方法，将刚刚创建的调度器实例初始化。 -
Scheduler.configure(jobstores=job_store)：使用Scheduler.configure方法配置了调度器的各种参数，包括指定了使用的存储方式为上面定义的job_store。 -
Scheduler.start()：最后调用Scheduler.start方法启动了定时任务调度器，使其开始执行任务调度操作。
然后前端页面添加任务时，调用add_coverage_task来添加定时任务，这样就将全量覆盖率任务集成到fastapi服务中了。
最后
我们最后还是在总结一下ApScheduler的核心组件，看官方文档是这样写的：
- triggers
- job stores
- executors
- schedulers
翻译一下，大概就是触发器、工作商店、执行者、调度程序，触发器就是定时任务啥时候触发，工作商店就是定时任务存在哪里，执行者就是执行任务的程序，调度程序就是不同任务之间的调度管理程序，包括任务的添加、删除和修改。
当然上文介绍了只是ApScheduler的基本使用方法，想要了解更多API，可以翻阅官方文档：https://apscheduler.readthedocs.io/