定时任务与异步处理

Go-Zero

定时任务与异步处理

go-zero 官方不直接提供定时任务和延迟队列组件，但通过 ServiceGroup + go-queue + asynq 的组合，可以构建完整的异步任务体系。

技术选型

方案	适用场景	底层依赖	推荐度
asynq	延迟队列 + 定时任务 + 任务管理	Redis	⭐⭐⭐⭐⭐
kq (go-queue)	高吞吐消息队列	Kafka	⭐⭐⭐⭐
dq (go-queue)	延迟队列	Beanstalkd	⭐⭐⭐
robfig/cron	简单定时任务	无外部依赖	⭐⭐
go-zero executors	批量聚合、防抖节流	内存	⭐⭐⭐

💡 推荐组合：kq 处理高吞吐消息 + asynq 处理延迟/定时任务，通过 ServiceGroup 统一管理。

asynq — 核心方案

为什么选 asynq？

只依赖 Redis — 大多数项目已有 Redis
三合一 — 消息队列 + 延迟队列 + 定时任务
自带 Web UI — 查看任务状态、暂停、归档
优先级队列 — 支持多级队列权重分配

安装

go get github.com/hibiken/asynq
go get github.com/hibiken/asynqmon  # Web UI

项目目录结构

使用 ServiceGroup 管理异步服务，放在每个业务的 cmd/mq/ 子目录下：

app/order/cmd/mq/
├── internal/
│   ├── config/
│   │   └── config.go            # 配置结构体
│   ├── listen/
│   │   ├── listen.go            # 统一管理所有 MQ 消费者
│   │   ├── kqMqs.go             # Kafka 消费者
│   │   └── asynqMqs.go          # asynq 消费者
│   ├── mqs/
│   │   ├── asynq/
│   │   │   ├── closeOrder.go    # 延迟关单逻辑
│   │   │   └── sendEmail.go     # 发送邮件逻辑
│   │   └── kq/
│   │       └── paymentUpdate.go  # Kafka 消费逻辑
│   └── svc/
│       └── servicecontext.go
└── main.go

asynq 消费端（Server）

任务处理器

// internal/mqs/asynq/closeOrder.go
package asynq

import (
    "context"
    "encoding/json"
    "github.com/hibiken/asynq"
)

const TypeCloseOrder = "order:close"

type CloseOrderPayload struct {
    OrderId int64 `json:"order_id"`
}

type CloseOrderHandler struct {
    ctx    context.Context
    svcCtx *svc.ServiceContext
}

func NewCloseOrderHandler(ctx context.Context, svcCtx *svc.ServiceContext) *CloseOrderHandler {
    return &CloseOrderHandler{ctx: ctx, svcCtx: svcCtx}
}

func (h *CloseOrderHandler) ProcessTask(ctx context.Context, t *asynq.Task) error {
    var p CloseOrderPayload
    if err := json.Unmarshal(t.Payload(), &p); err != nil {
        return err  // 不可恢复错误，不重试
    }

    // 幂等检查
    order, _ := h.svcCtx.OrderModel.FindOne(ctx, p.OrderId)
    if order.Status != model.StatusWaitPay {
        return nil  // 已处理，幂等返回
    }

    // 执行关单逻辑
    return h.svcCtx.OrderModel.UpdateStatus(ctx, p.OrderId, model.StatusClosed)
}

注册并启动 Server

// internal/listen/asynqMqs.go
func AsynqMqs(c config.Config, ctx context.Context, svcCtx *svc.ServiceContext) []service.Service {
    srv := asynq.NewServer(
        asynq.RedisClientOpt{
            Addr:     c.Redis.Host,
            Password: c.Redis.Pass,
            DB:       c.Redis.DB,
        },
        asynq.Config{
            Concurrency: 10,             // 并发处理数
            Queues: map[string]int{       // 队列权重
                "critical": 6,            // 关键任务 60%
                "default":  3,            // 普通任务 30%
                "low":      1,            // 低优先级 10%
            },
            Retry: 3,                     // 最大重试次数
        },
    )

    mux := asynq.NewServeMux()
    mux.HandleFunc(TypeCloseOrder, NewCloseOrderHandler(ctx, svcCtx).ProcessTask)
    mux.HandleFunc(TypeSendEmail, NewSendEmailHandler(ctx, svcCtx).ProcessTask)

    return []service.Service{
        &AsynqService{
            srv:    srv,
            mux:    mux,
            svcCtx: svcCtx,
        },
    }
}

// AsynqService 实现 service.Service 接口
type AsynqService struct {
    srv    *asynq.Server
    mux    *asynq.ServeMux
    svcCtx *svc.ServiceContext
}

func (a *AsynqService) Start() {
    go func() {
        if err := a.srv.Run(a.mux); err != nil {
            log.Fatalf("asynq server crashed: %v", err)
        }
    }()
    logx.Info("asynq server started")
}

func (a *AsynqService) Stop() {
    a.srv.Shutdown()
    logx.Info("asynq server stopped")
}

asynq 生产端（Client）

在 API/RPC 的 ServiceContext 中注入：

type ServiceContext struct {
    Config      config.Config
    AsynqClient *asynq.Client
}

func NewServiceContext(c config.Config) *ServiceContext {
    client := asynq.NewClient(asynq.RedisClientOpt{
        Addr:     c.Redis.Host,
        Password: c.Redis.Pass,
    })
    return &ServiceContext{
        Config:      c,
        AsynqClient: client,
    }
}

在 Logic 中发送任务：

// 延迟 30 分钟关闭未支付订单
func (l *CreateOrderLogic) CreateOrder(req *types.CreateOrderReq) error {
    // 1. 创建订单
    orderId, err := l.createOrderInDB(req)
    if err != nil {
        return err
    }

    // 2. 发送延迟任务：30 分钟后关闭未支付订单
    payload, _ := json.Marshal(asynqmq.CloseOrderPayload{OrderId: orderId})
    task := asynq.NewTask(asynqmq.TypeCloseOrder, payload)

    _, err = l.svcCtx.AsynqClient.Enqueue(task,
        asynq.ProcessIn(30*time.Minute),  // 延迟 30 分钟
        asynq.Queue("default"),
        asynq.MaxRetry(3),
    )
    return err
}

常用 Enqueue 选项

选项	说明
`asynq.ProcessIn(d)`	延迟 `d` 时间后执行
`asynq.ProcessAt(t)`	在指定时间点执行
`asynq.Queue(name)`	指定队列
`asynq.MaxRetry(n)`	最大重试次数
`asynq.Timeout(d)`	任务超时时间
`asynq.Retention(d)`	完成后保留时间
`asynq.TaskID(id)`	自定义任务 ID（用于去重）

定时任务调度器（Scheduler）

// cmd/scheduler/main.go
package main

import (
    "encoding/json"
    "log"
    "github.com/hibiken/asynq"
)

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

    // 每天凌晨 2 点生成日报
    payload, _ := json.Marshal(ReportPayload{Date: "auto"})
    scheduler.Register("0 2 * * *", asynq.NewTask("report:daily", payload))

    // 每分钟同步配置（简便写法）
    scheduler.Register("*/1 * * * *", asynq.NewTask("config:sync", nil))

    // 每小时清理过期数据
    scheduler.Register("@every 1h", asynq.NewTask("clean:expired", nil))

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

分布式定时任务防重复

// 使用 Redis 分布式锁，确保多实例下只执行一次
func (h *DailyReportHandler) ProcessTask(ctx context.Context, t *asynq.Task) error {
    lockKey := fmt.Sprintf("lock:daily_report:%s", time.Now().Format("2006-01-02"))
    acquired, err := h.svcCtx.Redis.SetNxEx(ctx, lockKey, "1", 3600)
    if err != nil {
        return err  // Redis 故障，重试
    }
    if !acquired {
        return nil  // 已有其他实例在执行，幂等跳过
    }
    defer h.svcCtx.Redis.Del(ctx, lockKey)

    // 执行报表生成...
    return nil
}

go-zero 内置执行器（轻量场景）

go-zero 自带三种高性能执行器，适合不需要分布式队列的轻量场景：

ChunkExecutor — 批量执行

import "github.com/zeromicro/go-zero/core/executors"

executor := executors.NewChunkExecutor(
    func(tasks []executors.Task) {
        logs := make([]string, 0, len(tasks))
        for _, task := range tasks {
            logs = append(logs, task.(string))
        }
        batchWriteLogs(logs)
    },
    executors.WithChunkBytes(100),             // 每 100 条触发
    executors.WithFlushInterval(time.Second),  // 或每 1 秒触发
)

// 添加任务
executor.Add("log message")

双触发机制：任务数量达到阈值或时间间隔到期，满足任一条件即批量执行。

PeriodicalExecutor — 周期执行

executor := executors.NewPeriodicalExecutor(
    time.Second*5,
    func(tasks []executors.Task) {
        refreshCache(tasks)  // 每 5 秒刷新一次缓存
    },
)

// 优雅关闭
defer func() {
    executor.Flush()
    executor.Wait()
}()

DelayExecutor — 延迟执行（防抖）

// 搜索框防抖：300ms 内的多次输入只执行最后一次
debouncer := executors.NewDelayExecutor(
    func() { search(keyword) },
    300*time.Millisecond,
)
debouncer.Trigger()  // 每次输入都调用，但只有最后一次真正执行

asynq Web UI

import "github.com/hibiken/asynqmon"

func main() {
    h := asynqmon.New(asynqmon.Options{
        RootPath:     "/monitoring",
        RedisConnOpt: asynq.RedisClientOpt{Addr: "127.0.0.1:6379"},
    })
    // h 是一个 http.Handler，可挂载到 go-zero 路由上
}

访问 http://127.0.0.1:8080/monitoring 查看：

队列状态（待处理/处理中/已完成/已归档）
任务重试 / 失败原因
服务器负载
手动取消 / 归档 / 重新执行

ServiceGroup 统一管理

// main.go
func main() {
    flag.Parse()
    var c config.Config
    conf.MustLoad(*configFile, &c)

    svcCtx := svc.NewServiceContext(c)
    serviceGroup := service.NewServiceGroup()
    defer serviceGroup.Stop()

    // 统一管理多种异步服务
    for _, mq := range listen.Mqs(c, svcCtx) {
        serviceGroup.Add(mq)
    }
    serviceGroup.Start()
}

// internal/listen/listen.go
func Mqs(c config.Config, svcCtx *svc.ServiceContext) []service.Service {
    ctx := context.Background()
    var services []service.Service

    // Kafka 消费者
    services = append(services, KqMqs(c, ctx, svcCtx)...)
    // asynq 延迟队列 + 定时任务
    services = append(services, AsynqMqs(c, ctx, svcCtx)...)

    return services
}

最佳实践

实践	说明
幂等设计	每条任务处理前检查业务状态，防止重复执行
分布式锁	定时任务在多实例环境用 Redis 锁保证单次执行
队列优先级	`critical:6, default:3, low:1`，关键任务优先
任务 ID 化	用 `asynq.TaskID` 防止同一任务重复入队
超时设置	每个任务设置合理的 `asynq.Timeout`
失败重试	配置 `MaxRetry`，业务处理返回 error 触发重试
监控告警	通过 asynqmon Web UI 或 Prometheus 监控积压量

常见陷阱

陷阱	说明
🚨 忘记幂等处理	消费者重试会导致业务重复执行
🚨 定时任务多实例重复	Scheduler 多实例时，同一 cron 会多次入队，需要分布式锁
🚨 asynq 独占 Redis DB	asynq 默认使用 Redis DB 0，与业务缓存可能冲突，建议指定 DB
🚨 任务 payload 过大	Redis 作为存储，大 payload 影响性能
🚨 不设超时的长任务	Worker 被一个任务长期占用，导致队列堆积
🚨 硬删除 vs 归档	任务处理完需要排查时找不到记录，用 `Retention` 保留

错误处理与统一响应测试与部署