跳转到内容
Lingchu Bot

调度器

调度器

Section titled “调度器”

Lingchu Bot 在 nonebot-plugin-apscheduler 之上运行一个持久化调度器。任务规格存储在数据库中，使周期任务在重启后依然存活；处理器注册表将持久化的键映射到运行时可调用对象。调度器在启动期间初始化，并随其他运行时服务一起关闭。

架构

Section titled “架构”

调度器分为持有运行时调度的服务层与持有持久化的仓库层。

层 文件 职责
服务 services/scheduler.py 注册处理器、持久化任务规格、重新装填已启用任务、分发执行
仓库 repositories/scheduler_jobs.py 对 ScheduledJob 行做 CRUD、JSON 载荷编解码
模型 database/models/scheduler.py lingchu_scheduled_jobs 的 ORM 模型
适配器 nonebot_plugin_apscheduler 提供服务使用的进程内 scheduler

服务通过 NoneBot 的 require() 依赖 nonebot_plugin_apscheduler，并从其中导入共享的 scheduler 单例。APScheduler 本身由宿主工程配置（例如通过 NoneBot 配置），Lingchu 仅在其之上添加、移除与重新装填任务。

数据模型

Section titled “数据模型”

database/models/scheduler.py 在 lingchu_scheduled_jobs 表上定义 ScheduledJob ORM 模型。它使用 database/_dialect_compat.py 中的跨方言兼容类型（compat_string、CompatText、CompatBoolean、CompatDateTimeTZ），使同一套 schema 可在 SQLite、PostgreSQL、MySQL 与 MariaDB 上工作。

列 类型 说明
id Integer (Identity) 主键
job_id compat_string(128) 唯一、有索引；APScheduler 的 id
handler_key compat_string(128) 有索引；在内存处理器注册表中查找
trigger_type compat_string(32) APScheduler 触发器名（如 interval、cron）
trigger_kwargs CompatText JSON 编码的触发器构造参数
args CompatText JSON 编码的处理器位置参数（默认 []）
kwargs CompatText JSON 编码的处理器关键字参数（默认 {}）
enabled CompatBoolean 有索引；启动时仅调度已启用任务
coalesce CompatBoolean APScheduler coalesce 选项（默认 True）
max_instances Integer APScheduler max_instances 选项（默认 1）
misfire_grace_time Integer (可空) APScheduler misfire_grace_time 选项
created_at / updated_at CompatDateTimeTZ 有索引；默认 utc_now()

job_id 有唯一约束（uq_lingchu_scheduled_jobs_job_id）。仓库以 ensure_ascii=False 且键排序的 JSON 编码 trigger_kwargs、args、kwargs，并在解码时做严格类型校验（对象/数组/对象）。

处理器注册表

Section titled “处理器注册表”

服务维护一个内存中的 _handlers: dict[str, SchedulerHandler] 映射。处理器是任意 Callable[..., Awaitable[Any] | Any] —— 同步或异步均可 —— 通过 register_scheduler_handler(key, handler) 注册。

当一个持久化任务触发时，execute_persistent_job(job_id)：

  1. 通过 repository.get_job_spec(job_id) 加载 ScheduledJob 规格，
  2. 若任务缺失或 enabled = false 则提前返回，
  3. 查找 handler = _handlers.get(job.handler_key)，未注册时给出警告，
  4. 通过 repository.decode_job_payload(job) 解码 args / kwargs，并
  5. 通过 _maybe_await 等待处理器（或返回其同步结果）。

处理器在启动时、initialize_scheduler_service() 运行前注册。初始化之后也允许注册，但那些任务只有在其规格被持久化并调度后才会运行。

公共 API

Section titled “公共 API”

services/scheduler.py 暴露以下函数：

函数 用途
register_scheduler_handler(key, handler) 注册持久化任务使用的处理器键
clear_scheduler_handlers() 清空所有已注册处理器（测试辅助）
register_persistent_job(...) 持久化任务规格，enabled = true 时调度
remove_persistent_job(job_id) 移除运行时调度条目并删除持久化规格
initialize_scheduler_service() 将已启用的持久化任务重新装填进运行时调度器
execute_persistent_job(job_id) 加载并分发一个持久化任务（用作 APScheduler 可调用对象）
shutdown_scheduler_service() 预留的关闭钩子，供未来调度器清理使用

register_persistent_job() 在 handler_key 不在 _handlers 中时抛出 ValueError，因此拼写错误不会静默持久化一个永远不会运行的任务。它始终通过 repository.save_job_spec()（按 job_id upsert）保存规格，并仅在 enabled = true 时调用 _schedule_runtime_job()。

_schedule_runtime_job() 调用 scheduler.add_job() 时设置 replace_existing = True，因此重新注册任务会更新其触发器而非复制。

内置周期任务：消息存储清理

Section titled “内置周期任务：消息存储清理”

唯一的内置周期任务是消息存储清理任务。start/startup.py 在启动期间将其接入：

register_scheduler_handler(
    SCHEDULER_CLEANUP_HANDLER_KEY,
    cleanup_expired_messages,
)
await initialize_scheduler_service()

SCHEDULER_CLEANUP_HANDLER_KEY 在 services/message_store.py 中定义为 "message_store.cleanup_expired_messages"。处理器 cleanup_expired_messages() 删除 created_at 早于 message_store_retention_days 天的 MessageRecord、AuditRecord，以及 QQ + OneBot V11 + NoneBot 分区表（QQOneBotV11NoneBotEventRecord、QQOneBotV11NoneBotAuditRecord）。

清理逻辑遵循 core/runtime_config.py 中的两个运行时配置开关：

  • message_store_enabled（环境变量 LINGCHU_MESSAGE_STORE_ENABLED，默认 true）—— 若为 false，清理立即返回 (0, True)。
  • message_store_cleanup_enabled（环境变量 LINGCHU_MESSAGE_STORE_CLEANUP_ENABLED，默认 true）—— 若为 false，跳过清理。

当 message_store_retention_days 为 0 时，禁用按天过期，记录被无限期保留。同一清理逻辑也会在关闭期间通过 shutdown_message_store() 运行一次，因此即使没有调度周期任务，一次干净停机也会修剪过期记录。

添加自定义周期任务

Section titled “添加自定义周期任务”

注册一个自定义周期任务的步骤：

  1. 实现处理器为同步或异步可调用对象。将业务逻辑放在 services/ 或 repositories/ 中，避免在处理器内产生 NoneBot 匹配器副作用。

    from services.scheduler import register_scheduler_handler
    

    async def refresh_external_cache() -> None:
        ...
    

    register_scheduler_handler("my_feature.refresh_cache", refresh_external_cache)

  2. 在 initialize_scheduler_service() 运行前注册处理器键。 start/startup.py 的启动流程会先注册内置处理器，再调用 initialize_scheduler_service()；之后注册的处理器仅对在那之后持久化并调度的任务生效。

  3. 持久化并调度任务：调用 register_persistent_job() 并传入 APScheduler 触发器：

    await register_persistent_job(
        job_id="my_feature.refresh_cache",
        handler_key="my_feature.refresh_cache",
        trigger_type="interval",
        trigger_kwargs={"minutes": 30},
    )

  4. 通过 coalesce、max_instances、misfire_grace_time 调优 APScheduler 选项以控制误火行为。coalesce = True（默认）会将多次遗漏的运行合并为一次。

  5. 不再需要时用 remove_persistent_job(job_id) 移除任务。 它会移除运行时调度条目（若任务从未调度过则容忍 JobLookupError）并删除持久化规格。

持久化的 args 与 kwargs 以 JSON 编码，因此只能向处理器传入 JSON 可序列化的值。复杂对象必须在处理器内部根据其标识符重建。

生命周期

Section titled “生命周期”

调度器通过 start/startup.py 与 hooks/handlers/lifecycle.py 中的生命周期钩子参与 NoneBot 驱动生命周期：

  • driver.on_startup 调用 start.startup.startup()，其中注册处理器并调用 initialize_scheduler_service()。
  • driver.on_shutdown 先调用 shutdown_scheduler_service()（目前为预留空操作），再调用 shutdown_message_store()。

initialize_scheduler_service() 在加载持久化规格时容忍 DatabaseError：记录失败日志并直接返回，不调度任何任务，因此启动期间数据库问题永远不会阻止机器人启动。单个任务调度失败（例如触发器参数无效）会被记录并跳过，其余任务仍会被调度。