Funboost Timing Jobs

Funboost 定时任务

概述

Funboost 集成 APScheduler 实现定时任务发布。ApsJobAdder 封装了 APScheduler，按 cron/interval/date 触发器定时推送消息到队列。

核心原则： 使用 ApsJobAdder — 绝对禁止直接用 apscheduler.add_job 调用 funboost 任务函数。

适用场景

• 按固定间隔运行任务（每 N 秒/分钟/小时）
• Cron 风格调度（如每天凌晨 2 点）
• 指定时间一次性执行
• 分布式调度（Redis 作业存储）

铁律

绝对禁止 使用 apscheduler.add_job(my_task, ...) — 这会在调度进程中直接执行函数，完全绕过队列。必须使用 ApsJobAdder，它在调度时间点调用 push 发送消息。

重要提醒： ApsJobAdder 的第一个参数必须是 @boost 装饰后的函数（即 booster），严禁传入普通未装饰的函数。只有经过 @boost 装饰的函数才拥有 .push() 方法，ApsJobAdder 内部依赖此方法定时将消息推入队列。

速查表

核心代码模式

注意： 以下示例使用 REDIS_ACK_ABLE broker，需配置 Redis 连接。改用 BrokerEnum.MEMORY_QUEUE 可零依赖本地测试。

from funboost import boost, BoosterParams, BrokerEnum, ApsJobAdder

@boost(BoosterParams(
    queue_name="scheduled_task",
    broker_kind=BrokerEnum.REDIS_ACK_ABLE,
    qps=5,
))
def cleanup_expired_data(table_name: str):
    print(f"清理 {table_name}")

if __name__ == "__main__":
    cleanup_expired_data.consume()

    # ⚠️ ApsJobAdder 第一个参数必须是 @boost 装饰后的函数（即 booster），
    #    禁止传入普通未装饰的函数！否则定时 push 无法进入队列。
    # 间隔：每 30 秒执行（默认 job_store_kind='memory'，重启后丢失）
    ApsJobAdder(cleanup_expired_data).add_push_job(
        trigger="interval",
        seconds=30,
        kwargs={"table_name": "sessions"},
        id="cleanup_sessions",
        replace_existing=True,
    )

    # Cron：每天凌晨 2:00 执行
    ApsJobAdder(cleanup_expired_data).add_push_job(
        trigger="cron",
        hour=2,
        minute=0,
        kwargs={"table_name": "logs"},
        id="cleanup_logs_daily",
        replace_existing=True,
    )

使用 Redis 作业存储（分布式）

ApsJobAdder(cleanup_expired_data, job_store_kind="redis").add_push_job(
    trigger="interval",
    seconds=60,
    kwargs={"table_name": "cache"},
    id="cleanup_cache",
)

使用 job_store_kind="redis" 后，调度信息持久化到 Redis，进程重启不丢失，多实例部署时只有一个会触发。

一次性定时任务

from datetime import datetime

ApsJobAdder(send_report).add_push_job(
    trigger="date",
    run_date=datetime(2026, 7, 1, 9, 0, 0),
    kwargs={"report_type": "monthly"},
    id="july_report",
)

传递位置参数

ApsJobAdder(my_task).add_push_job(
    trigger="interval",
    seconds=10,
    args=(1, 2),      # push 的位置参数
    id="my_interval_job",
)

完整示例

from funboost import boost, BoosterParams, BrokerEnum, ApsJobAdder


@boost(BoosterParams(
    queue_name="heartbeat_queue",
    broker_kind=BrokerEnum.REDIS_ACK_ABLE,
))
def heartbeat(service_name: str):
    import time
    print(f"[{time.strftime('%H:%M:%S')}] 心跳来自 {service_name}")

if __name__ == "__main__":
    heartbeat.consume()

    ApsJobAdder(heartbeat, job_store_kind="redis").add_push_job(
        trigger="interval",
        seconds=5,
        kwargs={"service_name": "api-server"},
        id="api_heartbeat",
    )

常见错误

相关 Skill

• using-funboost-basics — 基础使用入门
• funboost-funweb-ops — Web 管理界面运维