消息队列集成

Go-Zero

消息队列集成

go-zero 通过官方组件 go-queue 提供多消息队列的支持，涵盖 Kafka（kq）、Beanstalkd（dq）、RabbitMQ 等。同时社区推荐使用 asynq 处理延迟队列和定时任务。

技术选型速查

场景	推荐方案	底层依赖
高吞吐消息、流式处理	kq（Kafka）	Kafka
延迟队列	dq（Beanstalkd）或 asynq	Beanstalkd / Redis
定时任务调度	asynq	Redis
可靠投递、复杂路由	RabbitMQ	AMQP
轻量级消息	natsmq	NATS

Kafka（kq）集成

安装

go get github.com/zeromicro/go-queue/kq

配置

KafkaConf:
  Brokers:
    - 127.0.0.1:9092
  Group: order-consumer-group
  Topic: order-events
  Offset: first             # first | last
  Conns: 1                  # Kafka 连接数
  Consumers: 1              # 消费 goroutine 数
  Processors: 1             # 处理 goroutine 数
  MinBytes: 1024            # 最小拉取字节
  MaxBytes: 1048576         # 最大拉取字节
  Username: ""              # SASL 认证
  Password: ""

生产者（Pusher）

// ServiceContext 中初始化
type ServiceContext struct {
    Config    config.Config
    KqPusher  *kq.Pusher
}

func NewServiceContext(c config.Config) *ServiceContext {
    return &ServiceContext{
        Config:   c,
        KqPusher: kq.NewPusher(c.KafkaConf.Brokers, c.KafkaConf.Topic),
    }
}

// Logic 中发送消息
func (l *CreateOrderLogic) CreateOrder(req *types.CreateOrderReq) error {
    // 1. 创建订单
    // ...

    // 2. 发送消息
    msg := map[string]interface{}{
        "order_id": orderId,
        "user_id":  req.UserId,
        "amount":   req.Amount,
    }
    msgBytes, _ := json.Marshal(msg)

    if err := l.svcCtx.KqPusher.Push(string(msgBytes)); err != nil {
        return err
    }
    return nil
}

可选参数：

// chunkSize: 批量提交字节大小（默认 1MB）
// flushInterval: 定时提交间隔
pusher := kq.NewPusher(brokers, topic,
    kq.WithChunkSize(1024*1024),
    kq.WithFlushInterval(time.Second),
)

消费者（Consumer）

目录结构：

app/order/cmd/mq/
├── internal/
│   ├── config/
│   │   └── config.go
│   ├── svc/
│   │   └── servicecontext.go
│   ├── listen/
│   │   ├── listen.go        # 统一管理所有 MQ 消费者
│   │   └── kqMqs.go         # Kafka 消费者注册
│   └── mqs/                 # 各队列业务逻辑
│       └── kq/
│           └── paymentStatus.go
└── main.go

消费者业务逻辑 — 实现 Consume 接口：

// internal/mqs/kq/paymentStatus.go
type PaymentStatusMq struct {
    ctx    context.Context
    svcCtx *svc.ServiceContext
}

func (l *PaymentStatusMq) Consume(_, val string) error {
    var msg PaymentMessage
    if err := json.Unmarshal([]byte(val), &msg); err != nil {
        return err
    }
    // 处理业务逻辑
    return l.updateOrderStatus(msg)
}

注册消费者：

// internal/listen/kqMqs.go
func KqMqs(c config.Config, ctx context.Context, svcCtx *svc.ServiceContext) []service.Service {
    return []service.Service{
        kq.MustNewQueue(c.PaymentUpdateConf, kqMq.NewPaymentStatusMq(ctx, svcCtx)),
    }
}

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()

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

// 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
}

RabbitMQ 集成

go-queue 仓库提供 RabbitMQ 支持，底层使用 rabbitmq/amqp091-go。

// ServiceContext 中初始化
conn, _ := amqp.Dial("amqp://guest:guest@localhost:5672/")
channel, _ := conn.Channel()

// 声明交换机
channel.ExchangeDeclare("order.exchange", "topic", true, false, false, false, nil)

// 声明队列
queue, _ := channel.QueueDeclare("order.queue", true, false, false, false, nil)

// 绑定
channel.QueueBind(queue.Name, "order.*", "order.exchange", false, nil)

// 发布消息
channel.Publish("order.exchange", "order.created", false, false, amqp.Publishing{
    ContentType: "application/json",
    Body:        msgBytes,
})

💡 RabbitMQ 适合需要复杂路由和可靠投递保证的场景（如支付通知、邮件发送）。

延迟队列与定时任务（asynq）

asynq 是基于 Redis 的任务队列，同时支持消息队列、延迟队列、定时任务。详见 13-定时任务与异步处理.md。

发送延迟任务

// 在 API/RPC logic 中
client := asynq.NewClient(asynq.RedisClientOpt{Addr: "127.0.0.1:6379"})

payload, _ := json.Marshal(CloseOrderPayload{OrderId: 123})
task := asynq.NewTask("order:close", payload)

// 延迟 30 分钟后执行
client.Enqueue(task, asynq.ProcessIn(30*time.Minute))

消费者

srv := asynq.NewServer(
    asynq.RedisClientOpt{Addr: "127.0.0.1:6379"},
    asynq.Config{Concurrency: 10},
)

mux := asynq.NewServeMux()
mux.HandleFunc("order:close", closeOrderHandler)
srv.Run(mux)

定时任务

scheduler := asynq.NewScheduler(
    asynq.RedisClientOpt{Addr: "127.0.0.1:6379"}, nil,
)

// 每天凌晨 2 点执行
scheduler.Register("0 2 * * *", asynq.NewTask("report:daily", payload))

scheduler.Run()

消息可靠性保证

生产者端

策略	说明
本地事务 + 消息表	业务数据和消息写入同一 DB 事务，后续异步发送
同步发送 + 确认	`kq.Pusher` 推送后检查返回的 error
重试机制	发送失败后指数退避重试

消费者端

策略	说明
幂等处理	消息分配唯一 ID，消费端根据 ID 去重
手动确认	处理成功后再 commit offset
死信队列	处理失败的消息进入死信队列，人工处理
限流消费	`Processors: 1` 控制并发处理数

幂等实现

func (l *OrderMq) Consume(_, val string) error {
    var msg OrderMessage
    json.Unmarshal([]byte(val), &msg)

    // 1. 检查消息是否已处理
    exists, _ := l.svcCtx.Redis.Exists(l.ctx, "msg:"+msg.MessageId)
    if exists {
        return nil  // 已处理，幂等返回
    }

    // 2. 处理业务
    if err := l.processOrder(msg); err != nil {
        return err
    }

    // 3. 标记已处理
    l.svcCtx.Redis.SetWithExpire(l.ctx, "msg:"+msg.MessageId, "1", 24*time.Hour)
    return nil
}

ServiceGroup 统一管理

go-zero 的 ServiceGroup 可以同时管理 API、RPC、MQ 消费者：

func main() {
    group := service.NewServiceGroup()
    defer group.Stop()

    // API 服务（可选）
    // group.Add(apiServer)

    // RPC 服务（可选）
    // group.Add(rpcServer)

    // Kafka 消费者
    for _, mq := range listen.Mqs(c) {
        group.Add(mq)
    }

    group.Start()
}

最佳实践总结

实践	说明
消息 ID 化	每条消息带唯一 ID，消费端幂等去重
削峰填谷	配合限流和熔断保护下游服务
死信队列	失败消息不丢弃，进死信队列人工处理
监控告警	监控消息积压量和消费延迟
批量处理	能用批量就不用逐条，用 `ChunkExecutor`
优先 asynq	中小规模项目优先 asynq（只依赖 Redis）
优先 kq	高吞吐场景用 kq（Kafka）

常见陷阱

陷阱	说明
🚨 消息消费无幂等	消费者重试导致重复处理，必须有去重机制
🚨 消费者处理失败不返回 error	kq 框架依赖 error 判断是否重试，吞掉 error 会导致消息丢失
🚨 `Processors` 设置过大	并发过高可能压垮下游 DB，需要根据 DB 连接池大小合理设置
🚨 忘记设置 `Group`	多实例消费同一 Topic 时需要不同的 Group 做负载均衡
🚨 Kafka topic 不存在	`kq.NewPusher` 默认不会自动创建 topic
🚨 延迟队列用 Kafka	Kafka 不支持原生延迟队列，用 asynq 或 Beanstalkd

数据库与缓存链路追踪与可观测性