周期性任务

Asynq

周期性任务

1 概述

Asynq 提供两种周期性任务管理方式：

方式	适用场景	复杂度
Scheduler	简单的静态定时任务，cron 表达式硬编码	低
PeriodicTaskManager	需要动态增删定时任务，配置可外部化	中

两者均通过 Redis 分布式锁保证同一时刻只有一个调度实例在工作，避免重复触发。

2 Scheduler（基础调度器）

2.1 创建与配置

scheduler := asynq.NewScheduler(
    asynq.RedisClientOpt{Addr: "localhost:6379"},
    &asynq.SchedulerOpts{
        Location: time.Local,          // 时区
        EnqueueError: func(task *asynq.Task, opts []asynq.Option, err error) {
            log.Printf("scheduler enqueue error: %v", err)
        },
    },
)

2.2 SchedulerOpts 字段

字段	类型	默认值	说明
`Location`	`*time.Location`	`time.UTC`	Cron 表达式解析的时区
`Logger`	`Logger`	默认	调度器日志
`LogLevel`	`LogLevel`	`InfoLevel`	日志级别
`EnqueueError`	`func(task *Task, opts []Option, err error)`	无	入队失败回调
`HeartbeatInterval`	`time.Duration`	0	调度器心跳间隔（v0.25.1+）
`PreEnqueueFunc`	`func(task *Task, opts []Option)`	无	入队前回调，可修改任务属性
`PostEnqueueFunc`	`func(info *TaskInfo, err error)`	无	入队后回调

2.3 注册定时任务

task, _ := tasks.NewEmailTask(0, "daily-report")
entryID, err := scheduler.Register("0 9 * * *", task, asynq.Queue("default"))

Cron 表达式格式：

┌─────────── 分钟 (0-59)
│ ┌─────────── 小时 (0-23)
│ │ ┌─────────── 日 (1-31)
│ │ │ ┌─────────── 月 (1-12)
│ │ │ │ ┌─────────── 星期 (0-6, 0=周日)
│ │ │ │ │
* * * * *

2.4 Cron 表达式示例

表达式	含义
`"0 9 * * *"`	每天上午 9:00
`"0 0 1 * *"`	每月 1 号 0:00
`"/5 * * *"`	每 5 分钟
`"0 9 * * 1-5"`	工作日（周一至周五）上午 9:00
`"0 0,12 * * *"`	每天 0:00 和 12:00
`"@every 30s"`	每 30 秒
`"@every 1h30m"`	每 1 小时 30 分钟
`"@every 24h"`	每 24 小时

2.5 启动与关闭

// 方式 1：阻塞运行
if err := scheduler.Run(); err != nil {
    log.Fatal(err)
}

// 方式 2：非阻塞
if err := scheduler.Start(); err != nil {
    log.Fatal(err)
}
defer scheduler.Shutdown()

2.6 注销定时任务

// 注册时获取 entryID
entryID, err := scheduler.Register("* * * * *", task)

// 根据需要注销
if err := scheduler.Unregister(entryID); err != nil {
    log.Printf("unregister failed: %v", err)
}

2.7 完整示例

func main() {
    scheduler := asynq.NewScheduler(
        asynq.RedisClientOpt{Addr: "localhost:6379"},
        &asynq.SchedulerOpts{
            Location: time.Local,
            EnqueueError: func(task *asynq.Task, opts []asynq.Option, err error) {
                log.Printf("enqueue error: task=%s err=%v", task.Type(), err)
            },
            PostEnqueueFunc: func(info *asynq.TaskInfo, err error) {
                if err != nil {
                    log.Printf("post enqueue error: %v", err)
                } else {
                    log.Printf("scheduled task enqueued: id=%s", info.ID)
                }
            },
        },
    )

    // 每日报表
    reportTask, _ := tasks.NewReportExportTask("daily", "pdf")
    scheduler.Register("0 9 * * *", reportTask, asynq.Queue("default"))

    // 每小时数据同步
    syncTask, _ := tasks.NewDataSyncTask()
    scheduler.Register("0 * * * *", syncTask, asynq.Queue("low"))

    // 每周清理（周日凌晨 3 点）
    cleanupTask, _ := tasks.NewCleanupTask()
    scheduler.Register("0 3 * * 0", cleanupTask, asynq.Queue("low"))

    if err := scheduler.Run(); err != nil {
        log.Fatal(err)
    }
}

3 PeriodicTaskManager（动态调度管理器）

PeriodicTaskManager 支持从外部配置源动态加载定时任务，适合需要运行时增减任务而不重启服务的场景。

3.1 组件架构

┌──────────────────────────────┐
│ PeriodicTaskConfigProvider   │  ← 实现此接口，从 DB/文件/配置中心读取配置
│ .GetConfigs()                │
└──────────┬───────────────────┘
           │ 定期同步 (SyncInterval)
           ▼
┌──────────────────────────────┐
│ PeriodicTaskManager          │
│ - 对比新旧配置                │
│ - 注册新任务                  │
│ - 注销已删除任务              │
│ - 通过 Redis 分布式锁防重复    │
└──────────────────────────────┘

3.2 基础用法

// 1. 定义配置提供者
type DBConfigProvider struct {
    db *sql.DB
}

func (p *DBConfigProvider) GetConfigs() ([]*asynq.PeriodicTaskConfig, error) {
    rows, err := p.db.Query("SELECT cron_expr, task_type, queue, payload FROM scheduled_tasks WHERE enabled = true")
    if err != nil {
        return nil, err
    }
    defer rows.Close()

    var configs []*asynq.PeriodicTaskConfig
    for rows.Next() {
        var cronExpr, taskType, queue string
        var payload []byte
        rows.Scan(&cronExpr, &taskType, &queue, &payload)

        task := asynq.NewTask(taskType, payload)
        configs = append(configs, &asynq.PeriodicTaskConfig{
            Cronspec: cronExpr,
            Task:     task,
            Opts:     []asynq.Option{asynq.Queue(queue)},
        })
    }
    return configs, nil
}

// 2. 创建 PeriodicTaskManager
provider := &DBConfigProvider{db: db}
mgr, err := asynq.NewPeriodicTaskManager(
    asynq.PeriodicTaskManagerOpts{
        RedisConnOpt:               asynq.RedisClientOpt{Addr: "localhost:6379"},
        PeriodicTaskConfigProvider: provider,
        SyncInterval:               10 * time.Second, // 每 10 秒同步一次配置
    },
)
if err != nil {
    log.Fatal(err)
}

// 3. 启动
if err := mgr.Run(); err != nil {
    log.Fatal(err)
}

3.3 PeriodicTaskManagerOpts 字段

字段	类型	说明
`RedisConnOpt`	`RedisConnOpt`	Redis 连接配置
`PeriodicTaskConfigProvider`	`interface`	提供定时任务配置的接口
`SyncInterval`	`time.Duration`	同步配置的间隔（建议 5~60 秒）
`SchedulerOpts`	`*SchedulerOpts`	底层 Scheduler 的配置
`Logger`	`Logger`	自定义日志

3.4 PeriodicTaskConfig 结构

type PeriodicTaskConfig struct {
    Cronspec string       // Cron 表达式
    Task     *Task        // 要执行的 Task
    Opts     []Option     // 入队选项（Queue、MaxRetry 等）
}

3.5 分布式锁机制

PeriodicTaskManager 使用 Redis 分布式锁确保同一时刻只有一个实例在执行调度：

实例 A 获取锁 → 执行调度循环
实例 B 等待锁    → 锁被释放（A 崩溃/下线）
                → B 获取锁 → 继续调度

🔬 深入原理：锁通过 Redis 的 SET NX EX 实现，具有 TTL 过期保护。即使调度实例崩溃，锁也会在 TTL 后自动释放，由其他实例接管。

4 Scheduler vs PeriodicTaskManager 选择

场景	推荐方案
任务数量固定、cron 表达式不变	Scheduler
需要运行时动态增删任务	PeriodicTaskManager
配置通过配置文件管理	Scheduler（CI/CD 部署更新）
配置通过管理后台管理	PeriodicTaskManager
简单的几个定时任务	Scheduler
大规模多租户定时任务	PeriodicTaskManager + DB 存储

5 部署建议

┌──────────────────────┐
│   单独部署 Scheduler  │  ← 独立进程，仅运行 Scheduler/PeriodicTaskManager
│   (1 个实例即可)      │
└──────────────────────┘
            │
            │ Enqueue 定时任务
            ▼
┌──────────────────────┐
│   Worker 集群         │  ← 多个 Worker 进程处理任务
│   (可水平扩展)        │
└──────────────────────┘

💡 最佳实践：

Scheduler 和 Worker 分开部署，避免调度瓶颈影响任务处理
Scheduler 部署 单实例即可（依赖 Redis 锁防止重复调度）
Worker 可按需水平扩展
高可用场景：部署 2 个 Scheduler 实例，Redis 锁保证只有一个工作

6 常见问题

6.1 夏令时问题

// 使用固定时区避免夏令时导致的任务重复或遗漏
SchedulerOpts{Location: time.UTC}

// 或使用北京时间
loc, _ := time.LoadLocation("Asia/Shanghai")
SchedulerOpts{Location: loc}

6.2 任务耗时超过调度间隔

如果一个定时任务的执行时间超过了 cron 间隔，下一次调度不会重叠执行——Asynq 的调度是触发式的，仅在时间点将任务入队，不关心任务是否执行完成。

6.3 调度精度

Cron 的最小精度是分钟。如果需要更高精度（秒级），使用 @every 语法：

scheduler.Register("@every 10s", task)   // 每 10 秒
scheduler.Register("@every 500ms", task) // 不支持——最小 1 秒

重试与错误处理高级特性