跳转到内容

APScheduler 定时任务与分布式锁

APScheduler 定时任务与分布式锁

学习目标

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

项目实践

调度器架构

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

启动流程

main.pyrun() 中:

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_LEADERLeader 选举12 小时
APPLY_MIGRATIONS确保 migration 只运行一次12 小时

问题与规避

问题影响规避策略
Leader 实例死亡但锁未释放新实例无法成为 Leadermax_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 的使用场景中不太可能出现。但如果任务需要长时间运行(如全量索引),需要考虑更细粒度的锁或进度追踪。

参考来源