FastAPI-template 项目中 Taskiq 定时任务调度器的配置与调试

概述

在使用 FastAPI-template 项目时，开发者经常需要配置后台定时任务。Taskiq 是一个强大的 Python 异步任务队列系统，它提供了灵活的定时任务调度功能。本文将详细介绍如何正确配置 Taskiq 的定时任务调度器，并解决常见的调度器不工作的问题。

Taskiq 调度器基础配置

Taskiq 的调度器配置需要几个关键组件：

1. 结果后端：通常使用 Redis 作为存储任务结果的中间件
2. 消息代理：负责在调度器和工作者之间传递消息
3. 调度源：定义任务何时以及如何被触发

基本配置示例如下：

from taskiq.redis import RedisAsyncResultBackend
from taskiq import TaskiqScheduler
from taskiq.schedule_sources import LabelScheduleSource

result_backend = RedisAsyncResultBackend(
    redis_url="redis://localhost:6379/1"
)

broker = ListQueueBroker(
    "redis://localhost:6379/1"
).with_result_backend(result_backend)

scheduler = TaskiqScheduler(
    broker=broker, 
    sources=[LabelScheduleSource(broker)]
)

定时任务定义

定义定时任务时，可以通过装饰器参数指定调度规则：

@broker.task(schedule=[{
    "cron": "*/1 * * * *",  # 每分钟执行一次
    "args": [10],           # 传递给任务的参数
    "kwargs": {},           # 关键字参数
    "labels": {}            # 任务标签
}])
async def heavy_task(a: int) -> int:
    logger.info(f"执行任务，参数: {a}")
    return 100 + a

Docker 环境下的正确部署

在 Docker 环境中部署 Taskiq 时，必须将调度器(scheduler)和工作进程(worker)分开运行：

services:
  taskiq-scheduler:
    command:
      - taskiq
      - scheduler
      - -fsd
      - market_insights.tkq:scheduler
   
  taskiq-worker:
    command:
      - taskiq
      - worker
      - -fsd
      - market_insights.tkq:broker

关键点说明：

• -fsd 参数表示启用文件系统发现，自动发现项目中的任务
• 调度器和工作进程必须引用不同的模块入口

常见问题排查

1. 调度器日志显示运行但任务未执行：

   • 检查工作进程是否正常运行
   • 确认 Redis 连接配置正确
   • 验证任务函数是否被正确导入

2. 任务执行无输出：

   • 确保日志级别设置为 INFO 或 DEBUG
   • 检查任务函数是否有语法错误
   • 验证任务参数类型是否匹配

3. 调度时间不准确：

   • 检查系统时区设置
   • 确认 cron 表达式格式正确

最佳实践建议

1. 监控与日志：为调度器和工作进程配置详细的日志记录，便于问题追踪

2. 任务幂等性：确保定时任务可以安全地多次执行而不会产生副作用

3. 资源隔离：在 Docker 中为调度器和工作进程分配独立的容器，避免相互影响

4. 测试策略：

   • 开发环境使用较短的间隔(如每分钟)测试
   • 生产环境调整为实际需要的频率
   • 使用 mock 测试任务函数逻辑

通过以上配置和调试方法，开发者可以确保 Taskiq 定时任务在 FastAPI-template 项目中稳定可靠地运行。