定时任务

定时任务模块基于 APScheduler 实现，支持 Cron 表达式调度、任务执行日志、动态启停等功能。包含 job（任务定义）、job_log（执行日志）、scheduler（调度器）三个子模块。

模块结构

src/modules/job/
├── models.py       # 任务模型
├── schemas.py      # 表单验证
├── repository.py   # 数据访问层
├── service.py      # 业务逻辑层
├── scheduler.py    # 调度器管理
└── job_executor.py # 任务执行器

src/modules/job_log/
├── models.py       # 执行日志模型
├── repository.py   # 日志数据访问
└── service.py      # 日志业务逻辑

任务模型

# src/modules/job/models.py
# ============================================================
# 定时任务模型
# ============================================================
class Job(base_model, base_db):
    """定时任务模型类"""
    # ============================================================
    # 表名配置
    # ============================================================
    __tablename__ = DB_PREFIX + "job"
    __table_comment__ = "定时任务表"

    # ============================================================
    # 字段定义
    # ============================================================
    # 任务名称
    job_name = Column(String(255), nullable=False, index=True, comment="任务名称")
    # 任务别名
    job_alias = Column(String(255), nullable=True, comment="任务别名")
    # 任务分组
    job_group = Column(String(255), nullable=True, comment="任务分组")
    # 任务触发器
    job_trigger = Column(String(255), nullable=True, comment="任务触发器")
    # 任务状态：0-未发布 1-运行中 2-暂停 3-删除
    status = Column(Integer, nullable=False, default=0, server_default=text('0'), index=True,
                    comment="任务状态：0-未发布 1-运行中 2-暂停 3-删除")
    # Cron表达式
    cron_expression = Column(String(255), nullable=True, comment="Cron表达式")
    # 执行策略：1-立即执行 2-执行一次 3-放弃执行
    execute_policy = Column(Integer, nullable=False, default=1, server_default=text('1'),
                            comment="执行策略：1-立即执行 2-执行一次 3-放弃执行")
    # 是否同步任务：0-否 1-是
    is_sync = Column(Integer, nullable=False, default=0, server_default=text('0'), comment="是否同步任务：0-否 1-是")
    # 任务URL
    url = Column(String(255), nullable=True, comment="任务URL")
    # 任务参数
    params = Column(Text, nullable=True, comment="任务参数")
    # 任务备注
    note = Column(String(255), nullable=True, comment="任务备注")

    # ============================================================
    # 内置方法
    # ============================================================
    def __str__(self):
        """返回任务名称作为字符串表示"""
        return "定时任务：{}".format(self.job_name)

执行日志模型

# src/modules/job_log/models.py
# ============================================================
# 定时任务日志模型
# ============================================================
class JobLog(base_model, base_db):
    """定时任务日志模型类"""
    # ============================================================
    # 表名配置
    # ============================================================
    __tablename__ = DB_PREFIX + "job_log"
    __table_comment__ = "定时任务日志表"

    # ============================================================
    # 字段定义
    # ============================================================
    # 任务ID
    job_id = Column(Integer, nullable=False, default=0, server_default=text('0'), index=True, comment="任务ID")
    # 任务名称
    job_name = Column(String(255), nullable=False, comment="任务名称")
    # 任务分组
    job_group = Column(String(255), nullable=True, comment="任务分组")
    # 任务触发器
    job_trigger = Column(String(255), nullable=True, comment="任务触发器")
    # 任务日志信息
    job_message = Column(String(500), nullable=True, comment="任务日志信息")
    # Cron表达式
    cron_expression = Column(String(255), nullable=True, comment="Cron表达式")
    # 执行状态：0-正常 1-失败
    status = Column(Integer, nullable=False, default=0, server_default=text('0'), index=True, comment="执行状态：0-正常 1-失败")
    # 执行开始时间
    start_time = Column(DateTime, nullable=True, comment="执行开始时间")
    # 执行结束时间
    end_time = Column(DateTime, nullable=True, comment="执行结束时间")
    # 执行耗时，单位：毫秒
    consume_time = Column(BigInteger, nullable=False, default=0, server_default=text('0'), comment="执行耗时，单位：毫秒")
    # 异常信息
    exception_info = Column(Text, nullable=True, comment="异常信息")

    # ============================================================
    # 内置方法
    # ============================================================
    def __str__(self):
        """返回任务名称作为字符串表示"""
        return "定时任务日志：{}".format(self.job_name)

调度器管理

# src/modules/job/scheduler.py
from apscheduler.schedulers.asyncio import AsyncIOScheduler

scheduler = AsyncIOScheduler()

def init_scheduler():
    """启动时加载所有启用的任务"""
    jobs = job_repo.get_all(status=0)
    for job in jobs:
        add_job(job)
    return len(jobs)

def add_job(job):
    """添加任务到调度器"""
    scheduler.add_job(
        func=execute_job,
        trigger='cron',
        id=str(job.id),
        args=[job.id],
        **parse_cron(job.cron_expression),
    )

def shutdown_scheduler():
    """关闭调度器"""
    scheduler.shutdown()

生命周期

调度器在应用启动时初始化（_combined_lifespan），加载所有状态为"运行"的任务。应用关闭时自动停止调度器。

Cron 表达式

字段	允许值	允许的特殊字符
秒	0-59	, - * /
分	0-59	, - * /
时	0-23	, - * /
日	1-31	, - * /
月	1-12	, - * /
周	0-7	, - * /

常用示例：

0 0/5 * * * ? 每5分钟执行
0 0 2 * * ? 每天凌晨2点执行
0 0 0 1 * ? 每月1号零点执行
0 0 * * * ? 每小时执行

API 接口

接口	方法	权限节点	说明
/api/v1/job/list	GET	sys:job:list	任务列表
/api/v1/job/add	POST	sys:job:add	新增任务
/api/v1/job/update	PUT	sys:job:update	编辑任务
/api/v1/job/delete/{id}	DELETE	sys:job:delete	删除任务
/api/v1/job/status	PUT	sys:job:status	启停任务
/api/v1/job/run	POST	sys:job:run	立即执行
/api/v1/jobLog/list	GET	sys:jobLog:list	执行日志

总结

定时任务模块具备以下特点：

1. APScheduler：基于成熟的调度框架，支持 Cron 表达式
2. 动态管理：通过接口动态添加、修改、启停任务
3. 执行日志：自动记录每次执行的开始时间、耗时、状态、异常
4. 并发控制：可配置是否允许并发执行
5. 错过策略：配置错过执行时的处理策略
6. 生命周期：应用启动自动加载，关闭自动停止