APScheduler 定时任务与分布式锁

APScheduler 定时任务与分布式锁

学习目标

理解 Khoj 如何使用 APScheduler 实现定时任务调度，以及在分布式部署中如何通过 ProcessLock 实现 Leader 选举和任务互斥。

项目实践

调度器架构

Khoj 使用 BackgroundScheduler + DjangoJobStore 实现持久化的定时任务调度：

启动流程

在 main.py 的 run() 中：

from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.jobstores.django import DjangoJobStore

scheduler = BackgroundScheduler()
scheduler.add_jobstore(DjangoJobStore(), "default")
scheduler.start()

所有定时任务通过 DjangoJobStore 持久化到 PostgreSQL。即使服务重启，任务状态（下次执行时间、执行历史）也不会丢失。

Schedule Leader 选举

在多实例部署中，如果每个实例都运行相同的定时任务，会导致重复执行（如重复发送通知、重复索引）。

Khoj 通过 ProcessLock 表的 SCHEDULE_LEADER 操作锁确保只有一个实例执行定时任务：

class ProcessLock(models.Model):
    class Operation(models.TextChoices):
        INDEX_CONTENT = "index_content"       # 内容索引锁
        SCHEDULED_JOB = "scheduled_job"        # 调度任务锁
        SCHEDULE_LEADER = "schedule_leader"    # Leader 选举锁
        APPLY_MIGRATIONS = "apply_migrations"  # Migration 锁

    name = models.CharField(max_length=200, choices=Operation.choices, unique=True)
    started_at = models.DateTimeField(auto_now_add=True)
    max_duration_in_seconds = models.IntegerField(default=43200)  # 12 小时

工作原理：多个实例启动后，第一个成功获取 SCHEDULE_LEADER 锁的实例成为 Leader，其他实例检测到锁已被占用，不执行定时任务。Leader 实例死亡后（锁过期），下一个实例会获取锁成为新 Leader。

自动化任务 API

用户通过 API 创建自定义的定时查询：

POST /api/automation
  - subject: 任务名称
  - query: 查询内容（自动加上 /automated_task 前缀）
  - crontime: 标准 5 字段 cron 表达式

GET /api/automation      # 列出所有自动化任务
PUT /api/automation      # 编辑任务
DELETE /api/automation   # 删除任务
POST /api/automation/trigger  # 手动触发

Crontime 处理：

• 使用 cron_descriptor 库验证 cron 表达式的合法性
• 截断为 5 字段标准格式
• ? 转换为 *（兼容 Quartz cron 语法）
• 禁止分钟级 cron：最小粒度为每小时（防止 API 滥用）

每个自动化任务创建独立的 Conversation 会话，标题为 Automation: {subject}。

进程锁的应用场景

锁类型	用途	超时时间
INDEX_CONTENT	确保只有一个进程在索引内容	12 小时
SCHEDULED_JOB	确保定时任务不重叠执行	12 小时
SCHEDULE_LEADER	Leader 选举	12 小时
APPLY_MIGRATIONS	确保 migration 只运行一次	12 小时

问题与规避

问题	影响	规避策略
Leader 实例死亡但锁未释放	新实例无法成为 Leader	max_duration_in_seconds（12 小时）过期后自动释放
分钟级 cron 滥用 API	过量 LLM 调用产生费用	明确禁止分钟级 cron，最小粒度为每小时
Django APScheduler 与内存调度不一致	任务状态可能在两者间不同步	统一使用 DjangoJobStore 持久化
多实例 Leader 切换时的任务中断	正在执行的任务被中断	当前设计不优雅处理，新 Leader 会跳过正在执行的任务

设计取舍

为什么使用 APScheduler 而非 Celery？

Celery 是更强大的分布式任务队列，但需要 Redis/RabbitMQ 作为 broker，增加了部署复杂度。Khoj 的定时任务相对简单（主要是周期性查询触发），APScheduler + DjangoJobStore 足够覆盖这些场景，且只需要 PostgreSQL 一个外部依赖。

为什么 ProcessLock 没有自动续期？

ProcessLock 的设计是一次性获取锁，执行完成后释放。如果任务执行时间超过 max_duration_in_seconds，锁会自动释放——这意味着另一个实例可能获取锁并开始执行相同任务。对于 12 小时的超时，这在 Khoj 的使用场景中不太可能出现。但如果任务需要长时间运行（如全量索引），需要考虑更细粒度的锁或进度追踪。

参考来源

• APScheduler 文档
• Django APScheduler