FastApi-RESTful 项目中的周期性任务处理指南

引言

在现代Web应用开发中，周期性任务处理是一个常见需求。无论是定时清理数据库过期记录、定期更新缓存，还是执行系统维护任务，都需要一种可靠的方式来处理周期性任务。FastApi-RESTful项目提供了一个优雅的解决方案——@repeat_every装饰器。

为什么需要专门的周期性任务处理

传统的做法是在启动事件中触发一个循环，但这种方法存在几个关键问题：

1. 启动阻塞：如果循环不结束，服务器无法正常启动
2. 事件循环阻塞：同步IO操作会阻塞整个事件循环
3. 异常处理：循环中的异常容易被忽略，导致任务静默失败

FastApi-RESTful的@repeat_every装饰器完美解决了这些问题，为开发者提供了开箱即用的周期性任务处理能力。

基本用法

简单示例

让我们看一个清理过期令牌的示例：

from fastapi import FastAPI
from fastapi_restful.tasks import repeat_every
from fastapi_restful.session import FastAPISessionMaker

app = FastAPI()
database_url = "sqlite:///./test.db"
sessionmaker = FastAPISessionMaker(database_url)

@app.on_event("startup")
@repeat_every(seconds=60 * 60)  # 每小时执行一次
async def remove_expired_tokens() -> None:
    async with sessionmaker.context_session() as session:
        # 这里执行删除过期令牌的逻辑
        pass

关键特性

1. 自动间隔执行：通过seconds参数指定执行间隔
2. 与FastAPI无缝集成：配合@app.on_event("startup")使用
3. 异步/同步支持：同时支持async def和普通def函数
4. 线程安全：同步函数会自动在线程池中执行

高级配置

延迟首次执行

默认情况下，任务会立即执行第一次调用。如果需要延迟首次执行，可以设置wait_first=True：

@repeat_every(seconds=3600, wait_first=True)
async def delayed_task():
    # 这个任务会在服务器启动1小时后首次执行
    pass

异常处理

@repeat_every提供了灵活的异常处理机制：

import logging

logger = logging.getLogger(__name__)

@repeat_every(seconds=60, logger=logger)
async def task_with_logging():
    # 如果这里抛出异常，会被记录到日志中
    pass

执行次数限制

如果需要限制任务执行次数，可以使用max_repetitions参数：

@repeat_every(seconds=60, max_repetitions=10)
async def limited_task():
    # 这个任务只会执行10次
    pass

最佳实践

1. 日志记录：始终为周期性任务配置日志记录，便于问题排查
2. 异常处理：根据任务重要性决定是否设置raise_exceptions=True
3. 资源清理：长时间运行的任务要注意资源释放
4. 测试验证：在开发环境充分测试周期性任务的可靠性

内部机制解析

@repeat_every装饰器的核心实现原理是：

1. 创建一个异步任务循环
2. 在每次循环中执行目标函数
3. 根据配置处理异常和日志
4. 维护执行计数和间隔控制

这种设计确保了任务执行的可靠性，同时不会影响主事件循环的性能。

总结

FastApi-RESTful的周期性任务处理功能为开发者提供了强大而简单的工具，使得定时任务的实现变得轻而易举。无论是简单的定时清理，还是复杂的周期性业务逻辑，都可以通过@repeat_every装饰器优雅地实现。通过合理配置参数，开发者可以构建出既可靠又易于维护的周期性任务系统。