Server 配置详解

Asynq

Server 配置详解

1 Server 创建

srv := asynq.NewServer(redisConnOpt, cfg)

两个参数：

redisConnOpt：Redis 连接配置（见 09-Redis配置与高可用）
cfg：asynq.Config 结构体，控制 Server 的所有行为

2 Config 全部字段

2.1 并发与队列

字段	类型	默认值	说明
`Concurrency`	`int`	CPU 核心数	同时运行的 Worker goroutine 上限
`Queues`	`map[string]int`	`{"default": 1}`	队列名 → 权重，权重越高被消费的概率越大
`StrictPriority`	`bool`	`false`	严格优先级：高权重队列清空后才处理低权重

asynq.Config{
    Concurrency: 20,
    Queues: map[string]int{
        "critical": 6,
        "default":  3,
        "low":      1,
    },
    StrictPriority: false, // 默认：按权重概率
}

2.2 队列权重详解

默认模式（加权概率）：

权重 6:3:1 → critical 被选中概率 60%，default 30%，low 10%

低权重队列不会被"饿死"，只是被消费的频率更低
适合大部分场景

严格优先级模式（StrictPriority）：

critical 队列为空时 → 才消费 default
default 队列为空时 → 才消费 low

低权重队列在高峰期可能被"饿死"
适合核心业务需绝对优先处理的关键场景

asynq.Config{
    Queues: map[string]int{
        "critical": 6,
        "default":  3,
        "low":      1,
    },
    StrictPriority: true,
}

2.3 重试与错误处理

字段	类型	默认值	说明
`RetryDelayFunc`	`func(n int, e error, t *Task) time.Duration`	指数退避	自定义重试延迟策略
`IsFailure`	`func(err error) bool`	`func(err error) bool { return true }`	判断错误是否计入重试次数
`ErrorHandler`	`asynq.ErrorHandler`	无（仅日志）	任务耗尽所有重试后的回调

asynq.Config{
    // 自定义重试延迟：前 3 次每次固定 30 秒，之后每次翻倍
    RetryDelayFunc: func(n int, e error, t *Task) time.Duration {
        if n <= 3 {
            return 30 * time.Second
        }
        return time.Duration(math.Pow(2, float64(n-3))) * 30 * time.Second
    },
    // 某些错误不消耗重试次数
    IsFailure: func(err error) bool {
        if errors.Is(err, ErrTemporary) {
            return false // 临时错误，不扣重试配额
        }
        return true
    },
    // 最终失败回调
    ErrorHandler: asynq.ErrorHandlerFunc(func(ctx context.Context, t *Task, err error) {
        log.Printf("TASK DEAD: type=%s id=%s err=%v", t.Type(), t.ResultWriter().TaskID(), err)
        // 发送告警、写入死信表等
    }),
}

详细讨论见 05-重试与错误处理

2.4 超时与优雅关闭

字段	类型	默认值	说明
`ShutdownTimeout`	`time.Duration`	0（无限等待）	优雅关闭时等待正在执行的任务完成的最大时长

asynq.Config{
    ShutdownTimeout: 30 * time.Second, // 关闭时等待最多 30 秒
}

2.5 健康检查

字段	类型	默认值	说明
`HealthCheckInterval`	`time.Duration`	0（不启用）	健康检查执行间隔
`HealthCheckFunc`	`func(error)`	无	健康检查回调函数

asynq.Config{
    HealthCheckInterval: 15 * time.Second,
    HealthCheckFunc: func(err error) {
        if err != nil {
            log.Printf("health check failed: %v", err)
        }
    },
}

2.6 任务调度

字段	类型	默认值	说明
`DelayedTaskCheckInterval`	`time.Duration`	5s	Forwarder 检查延迟任务的时间间隔

asynq.Config{
    DelayedTaskCheckInterval: 3 * time.Second, // 提高延迟任务的调度精度
}

2.7 聚合 (Group/Aggregation)

字段	类型	默认值	说明
`GroupGracePeriod`	`time.Duration`	0	聚合宽限期：首个任务入组后等待新任务的最短时间
`GroupMaxDelay`	`time.Duration`	0	最大聚合延迟：组内首个任务等待的最长时间
`GroupMaxSize`	`int`	0	最大聚合大小：达到此数量后立即触发聚合回调
`GroupAggregator`	`GroupAggregator`	无	自定义聚合函数

asynq.Config{
    GroupGracePeriod: 10 * time.Second,  // 至少等 10 秒
    GroupMaxDelay:    5 * time.Minute,   // 最多等 5 分钟
    GroupMaxSize:     100,               // 达到 100 条立即触发
}

详细讨论见 07-高级特性

2.8 清理与回收

字段	类型	默认值	说明
`JanitorInterval`	`time.Duration`	8s	Janitor 清理过期数据的间隔
`JanitorBatchSize`	`int`	100	每次清理的批次大小

asynq.Config{
    JanitorInterval:  10 * time.Second,
    JanitorBatchSize: 200,
}

2.9 日志

字段	类型	默认值	说明
`Logger`	`asynq.Logger`	默认 log.Logger	自定义日志实现
`LogLevel`	`asynq.LogLevel`	`InfoLevel`	日志级别

// 自定义 Logger（实现 asynq.Logger 接口）
type MyLogger struct {
    logger *slog.Logger
}

func (l *MyLogger) Debug(args ...interface{})           { l.logger.Debug(fmt.Sprint(args...)) }
func (l *MyLogger) Info(args ...interface{})            { l.logger.Info(fmt.Sprint(args...)) }
func (l *MyLogger) Warn(args ...interface{})            { l.logger.Warn(fmt.Sprint(args...)) }
func (l *MyLogger) Error(args ...interface{})           { l.logger.Error(fmt.Sprint(args...)) }
func (l *MyLogger) Printf(format string, args ...interface{}) { l.logger.Info(fmt.Sprintf(format, args...)) }

asynq.Config{
    Logger:   &MyLogger{logger: slog.Default()},
    LogLevel: asynq.WarnLevel,  // 仅输出 Warn 和 Error
}

LogLevel 取值：

级别	常量	说明
Debug	`asynq.DebugLevel`	全部日志，包含心跳、聚合、清理等内部信息
Info	`asynq.InfoLevel`	默认级别，包含任务处理、状态变更
Warn	`asynq.WarnLevel`	仅警告和错误
Error	`asynq.ErrorLevel`	仅错误（如 Redis 连接失败）
Fatal	`asynq.FatalLevel`	致命错误

2.10 BaseContext

v0.22.0+

asynq.Config{
    BaseContext: func() context.Context {
        return context.Background()
    },
}

BaseContext 定义了所有 Handler context 的父 context。默认返回 context.Background()。可用于在 context 中注入全局信息（如 logger、tracer）。

3 Server 生命周期方法

方法	签名	说明
`Run`	`func (srv *Server) Run(handler Handler) error`	阻塞运行，监听系统信号自动优雅关闭
`Start`	`func (srv *Server) Start(handler Handler) error`	非阻塞运行（内部启动 goroutine）
`Shutdown`	`func (srv *Server) Shutdown()`	触发优雅关闭
`Stop`	`func (srv *Server) Stop()`	停止 Server

3.1 Run vs Start

Run（阻塞模式）：

// 最简单的方式：阻塞直到 SIGINT/SIGTERM
if err := srv.Run(mux); err != nil {
    log.Fatal(err)
}

Start + 自定义信号处理（非阻塞模式）：

srv := asynq.NewServer(redisOpt, cfg)

if err := srv.Start(mux); err != nil {
    log.Fatal(err)
}

// 自行处理信号
sigCh := make(chan os.Signal, 1)
signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM)

<-sigCh
log.Println("Shutting down...")

// 先停 Asynq
srv.Shutdown()

// 再停其他资源
db.Close()
cache.Close()

log.Println("Shutdown complete")

💡 选择建议：

纯 Worker 进程 → 使用 Run
同一进程中还有 HTTP Server、gRPC Server 等其他组件 → 使用 Start

3.2 Shutdown vs Stop

方法	行为
`Shutdown()`	停止拉取新任务，等待正在执行的任务完成或 `ShutdownTimeout` 到期
`Stop()`	更快速地停止，不给正在执行的任务额外的宽限期

3.3 优雅关闭流程

收到信号 (SIGINT/SIGTERM)
    │
    ▼
停止拉取新任务
    │
    ▼
等待活跃任务完成（最多 ShutdownTimeout）
    │
    ├── 全部按时完成 → 正常退出
    │
    └── 超时 → 强制退出，未完成的任务成为孤儿任务
                   └── 由其他 Server 实例的 Recoverer 回收

🚨 陷阱：设置 ShutdownTimeout 时需要考虑任务的最长执行时间。如果超时时间过短，频繁重启会导致大量孤儿任务。

4 完整配置示例

srv := asynq.NewServer(
    asynq.RedisClientOpt{
        Addr:     "localhost:6379",
        Password: os.Getenv("REDIS_PASSWORD"),
        DB:       0,
    },
    asynq.Config{
        // ── 并发与队列 ──
        Concurrency: 20,
        Queues: map[string]int{
            "critical": 6,
            "default":  3,
            "low":      1,
        },
        StrictPriority: false,

        // ── 重试与错误 ──
        RetryDelayFunc: asynq.DefaultRetryDelayFunc,
        ErrorHandler: asynq.ErrorHandlerFunc(func(ctx context.Context, t *asynq.Task, err error) {
            log.Printf("[DEAD LETTER] type=%s err=%v", t.Type(), err)
        }),
        IsFailure: func(err error) bool {
            return !errors.Is(err, ErrTemporary)
        },

        // ── 超时与关闭 ──
        ShutdownTimeout: 30 * time.Second,

        // ── 健康检查 ──
        HealthCheckInterval: 15 * time.Second,
        HealthCheckFunc: func(err error) {
            if err != nil {
                metrics.HealthCheckFailed.Inc()
            }
        },

        // ── 聚合 ──
        GroupGracePeriod: 10 * time.Second,
        GroupMaxDelay:    5 * time.Minute,
        GroupMaxSize:     100,

        // ── 调度 ──
        DelayedTaskCheckInterval: 3 * time.Second,

        // ── 日志 ──
        Logger:   &slogLogger{},
        LogLevel: asynq.InfoLevel,
    },
)

5 并发数选取指南

任务类型	建议 Concurrency	原因
CPU 密集型（图像处理、加密）	CPU 核心数	避免上下文切换开销
IO 密集型（HTTP 调用、DB 查询）	CPU 核心数 × 2~4	充分利用 IO 等待时间
混合型	CPU 核心数 × 1.5~2	折中方案

import "runtime"

asynq.Config{
    Concurrency: runtime.NumCPU() * 2,  // IO 密集型
}

6 多 Server 水平扩展

// 多个 Server 实例共用同一个 Redis，自动实现任务分片
// 实例 A
srvA := asynq.NewServer(redisOpt, asynq.Config{Concurrency: 10})
// 实例 B
srvB := asynq.NewServer(redisOpt, asynq.Config{Concurrency: 10})
// 全局并发 = 20

所有实例共享同一个 Redis
通过 Redis 的原子操作实现无锁的任务分发
实例间无直接通信，天然支持动态扩缩容
⚠️ 注意：全局并发 = 各实例 Concurrency 之和

任务创建与入队任务处理与中间件