并发控制与限流

asynq.Config{
    Concurrency: 20, // 最多同时处理 20 个任务
}

// Asynq 内部近似实现
sema := make(chan struct{}, Concurrency)

// Worker 获取执行权
sema <- struct{}{}        // 阻塞直到有空位
defer func() { <-sema }() // 释放

// 执行任务
handler.ProcessTask(ctx, task)

import "runtime"

srv := asynq.NewServer(redisOpt, asynq.Config{
    Concurrency: runtime.NumCPU() * 2,
})

Queues: map[string]int{
    "critical": 6,  // ~60% Workers → ~12 并发（当 Concurrency=20）
    "default":  3,  // ~30% Workers → ~6 并发
    "low":      1,  // ~10% Workers → ~2 并发
}

import "github.com/hibiken/asynq/x/rate"

// 创建一个针对 "image:process" 任务类型的信号量
// 全局最多同时处理 10 个该类型任务
sema := rate.NewSemaphore(redisOpt, "image:process", 10)

var globalSemaphore = rate.NewSemaphore(redisOpt, "image:process", 5)

func HandleImageProcess(ctx context.Context, task *asynq.Task) error {
    // 确保 context 有 deadline（信号量自动清理过期 token）
    ctx, cancel := context.WithTimeout(ctx, 10*time.Minute)
    defer cancel()

    // 尝试获取 token
    acquired, err := globalSemaphore.Acquire(ctx)
    if err != nil || !acquired {
        // 获取失败 → 返回可重试错误，稍后再试
        return &RateLimitError{RetryIn: 30 * time.Second}
    }
    defer globalSemaphore.Release(ctx)

    // 持有 token，执行任务
    return processImage(ctx, task)
}

package tasks

import (
    "context"
    "errors"
    "fmt"
    "time"

    "github.com/hibiken/asynq"
    "github.com/hibiken/asynq/x/rate"
)

// 按任务类型的分布式信号量
var taskSemaphores = map[string]*rate.Semaphore{
    "image:process":  rate.NewSemaphore(redisOpt, "image:process", 10),
    "video:transcode": rate.NewSemaphore(redisOpt, "video:transcode", 3),
    "llm:inference":  rate.NewSemaphore(redisOpt, "llm:inference", 2),
}

var ErrRateLimited = errors.New("rate limited")

func RateLimitHandler(next asynq.Handler) asynq.Handler {
    return asynq.HandlerFunc(func(ctx context.Context, t *asynq.Task) error {
        sema, ok := taskSemaphores[t.Type()]
        if !ok {
            // 无限制，直接执行
            return next.ProcessTask(ctx, t)
        }

        // 设置 deadline 确保 token 自动清理
        ctx, cancel := context.WithTimeout(ctx, 5*time.Minute)
        defer cancel()

        acquired, err := sema.Acquire(ctx)
        if err != nil || !acquired {
            return fmt.Errorf("%w: %v", ErrRateLimited, err)
        }
        defer sema.Release(ctx)

        return next.ProcessTask(ctx, t)
    })
}

srv := asynq.NewServer(redisOpt, asynq.Config{
    Concurrency: 50,

    // 限流错误不消耗重试次数
    IsFailure: func(err error) bool {
        if errors.Is(err, ErrRateLimited) {
            return false // 不消耗重试配额
        }
        return true
    },

    // 被限流时延迟更久
    RetryDelayFunc: func(n int, e error, t *asynq.Task) time.Duration {
        if errors.Is(e, ErrRateLimited) {
            return time.Duration(n+1) * 15 * time.Second
        }
        return asynq.DefaultRetryDelayFunc(n, e, t)
    },
})

// 场景：下游 API 每秒仅支持 10 次调用
// 多个 Worker 实例共享此限制

var apiSemaphore = rate.NewSemaphore(redisOpt, "external_api", 10)

func HandleAPITask(ctx context.Context, t *asynq.Task) error {
    ctx, cancel := context.WithTimeout(ctx, 30*time.Second)
    defer cancel()

    // 获取调用许可
    acquired, err := apiSemaphore.Acquire(ctx)
    if err != nil || !acquired {
        return fmt.Errorf("api rate limited: %w", err)
    }
    defer apiSemaphore.Release(ctx)

    // 调用外部 API
    resp, err := externalAPI.Call(ctx, t.Payload())
    if err != nil {
        return fmt.Errorf("api call failed: %w", err)
    }

    // 处理响应...
    return nil
}