如何在Go语言微服务中利用Redis实现实时通知-Golang-PHP中文网

redis pub/sub易丢消息，因其内存级广播、无持久化和确认机制，服务重启或网络抖动即丢失；而redis stream支持持久化、消费者组、ack及消息回溯，可保障可靠通知。

如何在go语言微服务中利用redis实现实时通知

为什么直接用 Redis Pub/Sub 做通知容易丢消息

Redis 的 PUB/SUB 是内存级广播，不持久化、无确认机制。服务重启、网络抖动、消费者订阅延迟都会导致消息丢失——这在订单状态变更、支付结果推送等场景里是不可接受的。

真正可用的方案得靠 Redis Stream：它支持消费者组、消息持久化、ACK 机制，且 Go 官方 redis 库（github.com/redis/go-redis/v9）已原生支持。

Stream 中每条消息有唯一 ID，可回溯消费
消费者组（XGROUP）让多个微服务实例能协作消费同一流，避免重复处理
未 ACK 的消息会留在 PENDING 列表，崩溃后可重新分配

如何用 go-redis/v9 创建带消费者组的 Stream

初始化时必须显式创建消费者组，否则 XREADGROUP 会报错 NOGROUP No such key or consumer group。

关键点：组名需全局唯一；起始 ID 用 0 表示从头读，用 $ 表示只读新消息；autoclaim 不要依赖，手动处理 pending 更可控。

ctx := context.Background()
rdb := redis.NewClient(&redis.Options{Addr: "localhost:6379"})

// 创建 stream（自动完成），再建消费者组
err := rdb.XGroupCreateMkStream(ctx, "notify:stream", "order-group", "0").Err()
if err != nil && !errors.Is(err, redis.Nil) {
    log.Fatal(err)
}

// 生产者发消息（结构体序列化为 JSON）
msg := map[string]interface{}{"event": "order_paid", "order_id": "20240511001", "user_id": 123}
_, err = rdb.XAdd(ctx, &redis.XAddArgs{
    Stream: "notify:stream",
    Values: msg,
}).Result()

微服务实例如何安全消费 Stream 消息

单个 Go 服务启动时应启动一个 goroutine 负责监听，但不能简单用 XReadGroup 轮询——空响应会持续占用连接，还可能漏掉刚写入的消息。

Redis Skill - 高性能缓存管理

Redis 缓存和数据结构管理技能。通过自然语言操作 Redis，支持 String、Hash、List、Set、ZSet、Stream 等数据结构操作。当用户提到 Redis、缓存、消息队列、会话存储时使用此技能。

下载

正确做法是设 Block 参数（单位毫秒），配合 Count 控制每次最多取几条；消费完必须显式 XAck，否则消息永远卡在 pending。

使用 redis.XReadGroupArgs 时，Group 和 Consumer 必须非空，否则报错 ERR Unbalanced XREADGROUP
Streams 字段传 map[string]string{"notify:stream": ">"}，> 表示只读新消息
若处理失败，不要 Ack，让其他实例稍后重试；或调用 XClaim 主动接管 pending 消息

go func() {
    for {
        resp, err := rdb.XReadGroup(ctx, &redis.XReadGroupArgs{
            Group:    "order-group",
            Consumer: "svc-order-1",
            Streams:  map[string]string{"notify:stream": ">"},
            Count:    10,
            Block:    5000, // 阻塞 5s，避免空轮询
        }).Result()
        if err != nil {
            if !errors.Is(err, redis.Nil) {
                log.Printf("read err: %v", err)
            }
            continue
        }
        for _, msg := range resp[0].Messages {
            process(msg.Values) // 处理业务逻辑
            rdb.XAck(ctx, "notify:stream", "order-group", msg.ID) // 必须 ACK
        }
    }
}()