wehub-resource-sync
355 lines
11 KiB
Markdown
355 lines
11 KiB
Markdown
# 调度器重构可行性分析：使用 APScheduler
|
||
|
||
> **使用说明**：日常开发与配置 Cron/间隔任务请参阅 [scheduler.md](./scheduler.md)。
|
||
|
||
## 1. 当前实现分析
|
||
|
||
### 1.1 现有功能
|
||
- **任务类型支持**：
|
||
- Cron 任务（6位表达式：秒 分 时 日 月 周）
|
||
- 间隔任务（interval，支持秒/分/小时）
|
||
- 一次性任务（date，指定日期时间执行）
|
||
|
||
- **API 接口**：
|
||
- `add_cron_job(func, cron, job_id, **kwargs)`
|
||
- `add_interval_job(func, interval, job_id, **kwargs)`
|
||
- `add_date_job(func, run_date, job_id, **kwargs)`
|
||
- `remove_job(job_id)`
|
||
- `get_job(job_id)`
|
||
- `list_jobs()`
|
||
- `start()` / `stop()`
|
||
- `add_scheduled_job(job: ScheduledJob)`
|
||
|
||
- **集成点**：
|
||
- 与 `Application` 生命周期集成（启动/停止）
|
||
- 自动配置系统自动注册装饰器标记的任务
|
||
- 支持 `ScheduledJob` 类对象
|
||
- 配置支持（时区、最大工作线程数、启用/禁用）
|
||
|
||
### 1.2 当前实现的问题
|
||
- 自定义实现，维护成本高
|
||
- Cron 表达式解析逻辑复杂，可能存在边界情况
|
||
- 时区处理需要手动实现（依赖 pytz）
|
||
- 任务执行状态跟踪有限
|
||
- 缺少任务持久化能力
|
||
- 错误处理和重试机制简单
|
||
|
||
## 2. APScheduler 能力分析
|
||
|
||
### 2.1 APScheduler 核心特性
|
||
- **触发器（Triggers）**：
|
||
- `CronTrigger`：完整的 Cron 表达式支持（5位或6位）
|
||
- `IntervalTrigger`：固定间隔执行
|
||
- `DateTrigger`：指定日期时间执行
|
||
|
||
- **执行器（Executors）**：
|
||
- `ThreadPoolExecutor`：线程池执行
|
||
- `ProcessPoolExecutor`：进程池执行
|
||
- `AsyncIOExecutor`：异步执行
|
||
- `GeventExecutor`：Gevent 协程执行
|
||
|
||
- **任务存储（Job Stores）**：
|
||
- `MemoryJobStore`：内存存储（默认）
|
||
- `SQLAlchemyJobStore`：数据库持久化
|
||
- `RedisJobStore`：Redis 持久化
|
||
- `MongoDBJobStore`：MongoDB 持久化
|
||
|
||
- **调度器（Schedulers）**：
|
||
- `BlockingScheduler`：阻塞式调度器
|
||
- `BackgroundScheduler`：后台线程调度器（推荐）
|
||
- `AsyncIOScheduler`：异步调度器
|
||
- `GeventScheduler`：Gevent 调度器
|
||
- `TornadoScheduler`：Tornado 调度器
|
||
- `TwistedScheduler`：Twisted 调度器
|
||
|
||
### 2.2 APScheduler 优势
|
||
- ✅ 成熟稳定，社区活跃
|
||
- ✅ 完整的 Cron 表达式支持（包括复杂表达式）
|
||
- ✅ 内置时区支持（无需手动处理）
|
||
- ✅ 任务持久化能力
|
||
- ✅ 丰富的任务状态和事件监听
|
||
- ✅ 更好的错误处理和日志
|
||
- ✅ 支持任务暂停/恢复
|
||
- ✅ 支持任务修改（修改触发器）
|
||
|
||
## 3. 功能映射关系
|
||
|
||
### 3.1 任务类型映射
|
||
|
||
| 当前实现 | APScheduler | 映射方式 |
|
||
|---------|------------|---------|
|
||
| `add_cron_job(func, cron, ...)` | `scheduler.add_job(func, CronTrigger.from_crontab(cron), ...)` | ✅ 直接映射 |
|
||
| `add_interval_job(func, interval, ...)` | `scheduler.add_job(func, IntervalTrigger(seconds=interval), ...)` | ✅ 直接映射 |
|
||
| `add_date_job(func, run_date, ...)` | `scheduler.add_job(func, DateTrigger(run_date=...), ...)` | ✅ 直接映射 |
|
||
|
||
### 3.2 API 接口映射
|
||
|
||
| 当前方法 | APScheduler 对应方法 | 兼容性 |
|
||
|---------|-------------------|--------|
|
||
| `add_cron_job()` | `add_job(..., trigger=CronTrigger(...))` | ✅ 可封装保持兼容 |
|
||
| `add_interval_job()` | `add_job(..., trigger=IntervalTrigger(...))` | ✅ 可封装保持兼容 |
|
||
| `add_date_job()` | `add_job(..., trigger=DateTrigger(...))` | ✅ 可封装保持兼容 |
|
||
| `remove_job(job_id)` | `remove_job(job_id)` | ✅ 完全兼容 |
|
||
| `get_job(job_id)` | `get_job(job_id)` | ✅ 完全兼容 |
|
||
| `list_jobs()` | `get_jobs()` | ✅ 可封装保持兼容 |
|
||
| `start()` | `start()` | ✅ 完全兼容 |
|
||
| `stop()` | `shutdown()` | ⚠️ 需要封装（方法名不同） |
|
||
|
||
### 3.3 配置映射
|
||
|
||
| 当前配置 | APScheduler 配置 | 映射方式 |
|
||
|---------|-----------------|---------|
|
||
| `scheduler.enabled` | 通过 `start()` 控制 | ✅ 逻辑控制 |
|
||
| `scheduler.timezone` | `timezone` 参数 | ✅ 直接映射 |
|
||
| `scheduler.max_workers` | `executors['default']['max_workers']` | ✅ 直接映射 |
|
||
|
||
## 4. 重构方案设计
|
||
|
||
### 4.1 方案一：完全封装（推荐）
|
||
|
||
**设计思路**：保持现有 API 接口不变，内部使用 APScheduler 实现。
|
||
|
||
**优点**：
|
||
- ✅ 完全向后兼容，现有代码无需修改
|
||
- ✅ 可以逐步迁移，降低风险
|
||
- ✅ 保留自定义扩展能力
|
||
|
||
**实现要点**：
|
||
```python
|
||
from apscheduler.schedulers.background import BackgroundScheduler
|
||
from apscheduler.triggers.cron import CronTrigger
|
||
from apscheduler.triggers.interval import IntervalTrigger
|
||
from apscheduler.triggers.date import DateTrigger
|
||
|
||
class Scheduler:
|
||
def __init__(self, config):
|
||
# 创建 APScheduler 实例
|
||
self._scheduler = BackgroundScheduler(
|
||
timezone=config.get('scheduler.timezone', 'UTC'),
|
||
executors={
|
||
'default': {
|
||
'type': 'threadpool',
|
||
'max_workers': config.get('scheduler.max_workers', 10)
|
||
}
|
||
}
|
||
)
|
||
# 保持兼容性的映射
|
||
self._jobs = {} # job_id -> APScheduler Job
|
||
self._scheduled_jobs = {} # ScheduledJob 对象
|
||
|
||
def add_cron_job(self, func, cron, job_id=None, **kwargs):
|
||
# 转换 cron 表达式格式（6位 -> APScheduler 格式）
|
||
trigger = self._parse_cron(cron)
|
||
job = self._scheduler.add_job(
|
||
func,
|
||
trigger=trigger,
|
||
id=job_id,
|
||
**kwargs
|
||
)
|
||
self._jobs[job_id] = job
|
||
return job_id
|
||
```
|
||
|
||
### 4.2 方案二：直接使用 APScheduler
|
||
|
||
**设计思路**：直接暴露 APScheduler 的 API，简化封装层。
|
||
|
||
**优点**：
|
||
- ✅ 代码更简洁
|
||
- ✅ 可以直接使用 APScheduler 的高级特性
|
||
|
||
**缺点**：
|
||
- ❌ 需要修改现有代码
|
||
- ❌ 破坏向后兼容性
|
||
|
||
### 4.3 推荐方案：方案一（完全封装）
|
||
|
||
**理由**：
|
||
1. 保持 API 兼容性，现有代码无需修改
|
||
2. 可以逐步增强功能（如添加持久化）
|
||
3. 保留扩展空间
|
||
|
||
## 5. 实施步骤
|
||
|
||
### 5.1 阶段一：基础重构（保持兼容）
|
||
1. **创建新的 Scheduler 类**（基于 APScheduler）
|
||
- 实现所有现有 API 方法
|
||
- 保持方法签名和返回值一致
|
||
- 内部使用 APScheduler
|
||
|
||
2. **Cron 表达式转换**
|
||
- 当前格式：`"秒 分 时 日 月 周"`（6位）
|
||
- APScheduler 格式：`"分 时 日 月 周"`（5位）或 `CronTrigger` 对象
|
||
- 需要实现转换函数
|
||
|
||
3. **测试验证**
|
||
- 单元测试覆盖所有 API
|
||
- 集成测试验证现有功能
|
||
- 确保装饰器和自动配置正常工作
|
||
|
||
### 5.2 阶段二：功能增强（可选）
|
||
1. **添加任务持久化**
|
||
- 支持 SQLAlchemyJobStore
|
||
- 支持 RedisJobStore
|
||
|
||
2. **增强任务管理**
|
||
- 任务暂停/恢复
|
||
- 任务修改触发器
|
||
- 任务执行历史
|
||
|
||
3. **事件监听**
|
||
- 任务执行成功/失败事件
|
||
- 任务错过执行事件
|
||
|
||
### 5.3 阶段三：优化和文档
|
||
1. 性能优化
|
||
2. 文档更新
|
||
3. 示例代码更新
|
||
|
||
## 6. 关键技术点
|
||
|
||
### 6.1 Cron 表达式转换
|
||
|
||
**当前格式**（6位）：
|
||
```
|
||
"秒 分 时 日 月 周"
|
||
例如："0 0 * * * *" # 每小时
|
||
```
|
||
|
||
**APScheduler 格式**（5位或 CronTrigger）：
|
||
```python
|
||
# 方式1：5位字符串（标准 Cron）
|
||
"分 时 日 月 周"
|
||
"0 * * * *" # 每小时
|
||
|
||
# 方式2：CronTrigger 对象（推荐）
|
||
CronTrigger(second=0, minute=0, hour='*', day='*', month='*', day_of_week='*')
|
||
```
|
||
|
||
**转换函数**：
|
||
```python
|
||
def _parse_cron(self, cron_expr: str) -> CronTrigger:
|
||
"""将6位 Cron 表达式转换为 APScheduler CronTrigger"""
|
||
parts = cron_expr.split()
|
||
if len(parts) == 6:
|
||
second, minute, hour, day, month, weekday = parts
|
||
return CronTrigger(
|
||
second=second,
|
||
minute=minute,
|
||
hour=hour,
|
||
day=day,
|
||
month=month,
|
||
day_of_week=weekday
|
||
)
|
||
elif len(parts) == 5:
|
||
# 标准5位格式，秒默认为0
|
||
return CronTrigger.from_crontab(cron_expr)
|
||
else:
|
||
raise ValueError(f"无效的 Cron 表达式: {cron_expr}")
|
||
```
|
||
|
||
### 6.2 ScheduledJob 集成
|
||
|
||
需要将 `ScheduledJob.execute()` 方法适配到 APScheduler：
|
||
|
||
```python
|
||
def add_scheduled_job(self, job: ScheduledJob, job_id: Optional[str] = None):
|
||
# 包装 execute 方法，保持状态跟踪
|
||
def wrapped_execute():
|
||
return job.execute()
|
||
|
||
# 根据 trigger 类型添加任务
|
||
if isinstance(job.trigger, str):
|
||
return self.add_cron_job(wrapped_execute, job.trigger, job_id)
|
||
# ... 其他类型
|
||
```
|
||
|
||
### 6.3 时区处理
|
||
|
||
APScheduler 内置时区支持，无需手动处理：
|
||
|
||
```python
|
||
from pytz import timezone
|
||
|
||
scheduler = BackgroundScheduler(
|
||
timezone=timezone('Asia/Shanghai') # 直接支持时区对象
|
||
)
|
||
```
|
||
|
||
### 6.4 任务状态管理
|
||
|
||
APScheduler 提供任务状态：
|
||
- `pending`：等待执行
|
||
- `running`：正在执行
|
||
- `completed`：已完成
|
||
- `failed`：执行失败
|
||
|
||
可以通过 `get_job(job_id).next_run_time` 获取下次执行时间。
|
||
|
||
## 7. 风险评估
|
||
|
||
### 7.1 兼容性风险
|
||
- **风险**：Cron 表达式格式差异
|
||
- **缓解**：实现转换函数，确保兼容
|
||
|
||
### 7.2 行为差异风险
|
||
- **风险**：APScheduler 的执行时机可能与当前实现略有差异
|
||
- **缓解**：充分测试，特别是边界情况
|
||
|
||
### 7.3 性能风险
|
||
- **风险**：APScheduler 可能有额外的开销
|
||
- **缓解**：性能测试，通常 APScheduler 性能更好
|
||
|
||
## 8. 测试策略
|
||
|
||
### 8.1 单元测试
|
||
- 所有 API 方法的测试
|
||
- Cron 表达式转换测试
|
||
- 任务添加/删除测试
|
||
|
||
### 8.2 集成测试
|
||
- 装饰器自动注册测试
|
||
- ScheduledJob 集成测试
|
||
- 应用生命周期集成测试
|
||
|
||
### 8.3 兼容性测试
|
||
- 现有示例代码运行测试
|
||
- 现有项目迁移测试
|
||
|
||
## 9. 结论
|
||
|
||
### 9.1 可行性评估
|
||
✅ **高度可行**
|
||
|
||
**理由**：
|
||
1. APScheduler 功能完全覆盖现有需求
|
||
2. API 可以完全兼容
|
||
3. 项目已依赖 APScheduler
|
||
4. 重构收益明显（稳定性、功能、维护性）
|
||
|
||
### 9.2 推荐方案
|
||
**采用方案一（完全封装）**，分阶段实施：
|
||
1. 第一阶段：基础重构，保持兼容
|
||
2. 第二阶段：功能增强（可选）
|
||
3. 第三阶段：优化和文档
|
||
|
||
### 9.3 预期收益
|
||
- ✅ 减少代码量（约 200+ 行 → 100+ 行）
|
||
- ✅ 提高稳定性（使用成熟库）
|
||
- ✅ 增强功能（持久化、事件监听等）
|
||
- ✅ 降低维护成本
|
||
- ✅ 更好的错误处理
|
||
|
||
### 9.4 实施建议
|
||
1. **先创建新实现**，保留旧实现作为备份
|
||
2. **充分测试**后再替换
|
||
3. **逐步迁移**，可以先支持两种实现并存
|
||
4. **文档更新**，说明新特性
|
||
|
||
## 10. 参考资源
|
||
|
||
- APScheduler 官方文档：https://apscheduler.readthedocs.io/
|
||
- APScheduler GitHub：https://github.com/agronholm/apscheduler
|
||
- Cron 表达式参考：https://en.wikipedia.org/wiki/Cron
|
||
|