:::tip 定时任务 定时任务模块基于 APScheduler 实现，支持 Cron 表达式调度、任务执行日志、动态启停等功能。包含 job（任务定义）、job_log（执行日志）、scheduler（调度器）三个子模块。 ::: ## 模块结构 ``` src/modules/job/ ├── models.py # 任务模型 ├── schema.py # 表单验证（WTForms） ├── repository.py # 数据访问层 ├── service.py # 业务逻辑层 ├── router.py # 路由控制器 ├── scheduler_service.py # 调度器管理 ├── executor.py # 任务执行器 ├── mail_service.py # 邮件任务服务 └── sms_service.py # 短信任务服务 src/modules/job_log/ ├── models.py # 执行日志模型 ├── repository.py # 日志数据访问 ├── service.py # 日志业务逻辑 └── router.py # 路由控制器 ``` ## 任务模型 ```python # src/modules/job/models.py # ============================================================ # 定时任务模型 # ============================================================ class Job(BaseModel, BaseDB): """定时任务模型类""" # ============================================================ # 表名配置 # ============================================================ __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) ``` ## 执行日志模型 ```python # src/modules/job_log/models.py # ============================================================ # 定时任务日志模型 # ============================================================ class JobLog(BaseModel, BaseDB): """定时任务日志模型类""" # ============================================================ # 表名配置 # ============================================================ __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) ``` ## 调度器管理 ```python # src/modules/job/scheduler_service.py from apscheduler.schedulers.background import BackgroundScheduler scheduler = BackgroundScheduler(timezone="Asia/Shanghai") 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() ``` :::tip 生命周期 调度器在 `create_app()` 中通过 `scheduler.init_app(app)` 初始化，加载所有状态为"运行"的任务。应用关闭时自动停止调度器。 ::: ## 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. 生命周期：应用启动自动加载，关闭自动停止 ```