Redis 消息队列：Pub/Sub 广播与 Streams 持久化消息实战

# 客户端 A 订阅频道
SUBSCRIBE orders

# 客户端 B 订阅匹配模式（通配符）
PSUBSCRIBE orders.*

# 发布者推送消息
PUBLISH orders '{"order_id":"20250813001","status":"paid","amount":299.00}'

# 查看当前被订阅的频道（服务端）
PUBSUB CHANNELS
PUBSUB NUMSUB orders

# subscriber 断开连接期间发布的消息全部丢失
# Redis 不保存消息历史，新上线的 subscriber 无法收到离线期间的消息

# Redis 配置中的 client-output-buffer-limit pubsub
# 默认：client-output-buffer-limit pubsub 32mb 8mb 60
# 当 subscriber 消费慢于生产速度，超过 32MB 硬限制时
# Redis 会强制断开该 subscriber 连接，期间消息全部丢失

# 典型缓存失效广播用法
PUBLISH cache:invalidate "user:profile:10086"
# 所有缓存节点同时收到，立即删除本地缓存

# 自动分配 ID（推荐）
XADD orders:events * order_id 001 status paid amount 299.00
# 返回: 1723512000000-0

# 自定义 ID（必须递增）
XADD orders:events 1723512000100-0 order_id 002 status shipped

# 限制 Stream 长度（近似裁剪，类似 Kafka retention）
XADD orders:events MAXLEN ~ 10000 * order_id 003 status refund

# 严格限制长度（精确裁剪，性能略低）
XADD orders:events MAXLEN 10000 * order_id 004 status completed

# 阻塞读取新消息（$ 表示只接收执行后产生的新消息）
XREAD BLOCK 5000 STREAMS orders:events $
# 阻塞 5000ms，等待新消息到达

# 读取历史消息（从指定 ID 之后开始）
XREAD COUNT 10 STREAMS orders:events 1723512000000-0

# 同时监听多个 Stream
XREAD BLOCK 0 STREAMS orders:events payments:events $ $
# BLOCK 0 表示永久阻塞

# 作为消费者组中的独立消费者读取（见第三节）
XREADGROUP GROUP order_consumers consumer-1 COUNT 5 BLOCK 3000 \
  STREAMS orders:events >

# 按 ID 范围查询
XRANGE orders:events 1723512000000-0 1723512000891-0

# 分页查询（- 表示最小 ID，+ 表示最大 ID）
XRANGE orders:events - + COUNT 10

# 反向查询（从最新开始）
XREVRANGE orders:events + - COUNT 5

# 获取 Stream 长度
XLEN orders:events

# 删除特定消息
XDEL orders:events 1723512000000-0

# 查看 Stream 信息
XINFO STREAM orders:events
XINFO GROUPS orders:events
XINFO CONSUMERS orders:events order_consumers

# 手动裁剪到指定长度
XTRIM orders:events MAXLEN ~ 5000

# 删除整个 Stream
DEL orders:events

# 基于 ID 范围删除（比如删除 7 天前的消息）
# 先找到 7 天前的毫秒时间戳，然后删除区间
XRANGE orders:events - 1722907200000-0 COUNT 1000
# 遍历结果并用 XDEL 逐个删除

# 1. 创建消费者组（从 Stream 起始位置开始）
XGROUP CREATE orders:events order_consumers 0
# 0 表示从头消费，$ 表示只消费创建后的新消息

# 2. 读取消息（> 表示读取未分配给任何消费者的新消息）
XREADGROUP GROUP order_consumers consumer-1 COUNT 5 BLOCK 3000 \
  STREAMS orders:events >

# 3. 确认消息已处理
XACK orders:events order_consumers 1723512000000-0 1723512000523-0

# 4. 查看待处理消息（PEL）
XPENDING orders:events order_consumers - + 10

# 5. 查看某个消费者的待处理消息
XPENDING orders:events order_consumers - + 10 consumer-1

# 6. 转移未确认消息（consumer-1 宕机后，consumer-2 接管）
XCLAIM orders:events order_consumers consumer-2 60000 \
  1723512000000-0
# 60000 是 idle time，只转移空闲超过 60 秒的消息

# 完整消费循环示例

# Step 1: 创建组（仅需执行一次）
XGROUP CREATE orders:events order_group $ MKSTREAM
# MKSTREAM: 如果 Stream 不存在则自动创建

# Step 2: 消费者读取消息
XREADGROUP GROUP order_group worker-1 COUNT 1 BLOCK 5000 \
  STREAMS orders:events >
# 返回:
# 1) 1) "orders:events"
#    2) 1) 1) "1723512000000-0"
#          2) 1) "order_id"
#             2) "001"
#             3) "status"
#             4) "paid"

# Step 3: 业务处理...
# 处理完成后确认
XACK orders:events order_group 1723512000000-0

# Step 4: 如果处理失败不发送 XACK，消息留在 PEL
# 通过监控 PEL 发现未处理消息并重试

# 查看 PEL 概览
XPENDING orders:events order_group
# 返回: 1) (integer) 3           -- 待处理消息数
#       2) "1723512000000-0"     -- 最小 ID
#       3) "1723512000891-0"     -- 最大 ID
#       4) 1) 1) "worker-1"      -- 消费者
#             2) "2"              -- 该消费者持有的待处理数

# 查看具体待处理消息详情
XPENDING orders:events order_group - + 10
# 返回每个消息的 ID、消费者、空闲时间、投递次数

# 自动转移长时间未处理的消息（idel 超过 60 秒）
XPENDING orders:events order_group - + 10
# 拿到待处理 ID 列表后
XCLAIM orders:events order_group worker-backup 60000 \
  1723512000000-0 1723512000523-0

# 更现代的方式：XAUTOCLAIM（Redis 6.2+）
XAUTOCLAIM orders:events order_group worker-backup 60000 - \
  COUNT 100
# 自动转移所有 idle 超过 60 秒的待处理消息

# 反模式 1：用 Streams 替代 Kafka 处理海量日志
# 问题：Redis 内存成本远高于磁盘，MAXLEN 频繁裁剪影响性能
# 替代：Kafka + 冷存（S3/对象存储）

# 反模式 2：一个 Stream 配过多 Consumer Group
# 问题：每个 Group 独立维护 PEL 和游标，内存开销线性增长
# 替代：控制 Group 数量（通常 < 10），或拆分 Stream

# 反模式 3：从不发送 XACK
# 问题：PEL 无限增长，消费者重启后重复消费大量旧消息
# 修正：处理成功立即 XACK，失败时记录日志后也要 XACK（避免阻塞）

# 反模式 4：BLOCK 0 永久阻塞不设超时
# 问题：连接断开检测延迟，影响故障恢复
# 修正：BLOCK 3000-5000，应用层循环重连

module order-stream

go 1.22

require github.com/redis/go-redis/v9 v9.5.1

require (
	github.com/cespare/xxhash/v2 v2.2.0 // indirect
	github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect
)

package main

import (
	"context"
	"encoding/json"
	"fmt"
	"log"
	"time"

	"github.com/redis/go-redis/v9"
)

// OrderEvent 订单事件
type OrderEvent struct {
	OrderID   string    `json:"order_id"`
	UserID    int64     `json:"user_id"`
	Status    string    `json:"status"` // created, paid, shipped, delivered, refunded
	Amount    float64   `json:"amount"`
	Timestamp time.Time `json:"timestamp"`
}

var ctx = context.Background()

// Producer 订单事件生产者
type Producer struct {
	client *redis.Client
	stream string
}

// NewProducer 创建生产者
func NewProducer(addr string) *Producer {
	rdb := redis.NewClient(&redis.Options{
		Addr:     addr,
		Password: "",
		DB:       0,
		PoolSize: 10,
	})
	return &Producer{
		client: rdb,
		stream: "orders:events",
	}
}

// PublishEvent 发布订单事件到 Stream
func (p *Producer) PublishEvent(event OrderEvent) (string, error) {
	data, err := json.Marshal(event)
	if err != nil {
		return "", fmt.Errorf("marshal event failed: %w", err)
	}

	// XADD 自动分配 ID，限制 Stream 最大长度约 10000 条
	id, err := p.client.XAdd(ctx, &redis.XAddArgs{
		Stream: p.stream,
		MaxLen: 10000,
		Approx: true,
		Values: map[string]interface{}{
			"data": string(data),
		},
	}).Result()

	if err != nil {
		return "", fmt.Errorf("xadd failed: %w", err)
	}

	log.Printf("[Producer] Event published, id=%s, order=%s, status=%s",
		id, event.OrderID, event.Status)
	return id, nil
}

// PublishBatch 批量发布（减少 RTT）
func (p *Producer) PublishBatch(events []OrderEvent) error {
	pipe := p.client.Pipeline()
	for _, event := range events {
		data, _ := json.Marshal(event)
		pipe.XAdd(ctx, &redis.XAddArgs{
			Stream: p.stream,
			MaxLen: 10000,
			Approx: true,
			Values: map[string]interface{}{
				"data": string(data),
			},
		})
	}
	_, err := pipe.Exec(ctx)
	return err
}

// Close 关闭连接
func (p *Producer) Close() error {
	return p.client.Close()
}

// 演示：模拟订单事件流
func main() {
	producer := NewProducer("localhost:6379")
	defer producer.Close()

	statuses := []string{"created", "paid", "shipped", "delivered"}

	for i := 0; i < 100; i++ {
		event := OrderEvent{
			OrderID:   fmt.Sprintf("ORD-%06d", i+1),
			UserID:    int64(10000 + i%100),
			Status:    statuses[i%len(statuses)],
			Amount:    99.00 + float64(i)*10,
			Timestamp: time.Now(),
		}
		if _, err := producer.PublishEvent(event); err != nil {
			log.Printf("publish failed: %v", err)
		}
		time.Sleep(100 * time.Millisecond)
	}

	log.Println("All events published")
}

package main

import (
	"context"
	"encoding/json"
	"fmt"
	"log"
	"os"
	"os/signal"
	"syscall"
	"time"

	"github.com/redis/go-redis/v9"
)

// Consumer 消费者组 Worker
type Consumer struct {
	client       *redis.Client
	stream       string
	group        string
	consumerName string
	batchSize    int64
	blockMS      int64
}

// NewConsumer 创建消费者
func NewConsumer(addr, group, consumerName string) *Consumer {
	rdb := redis.NewClient(&redis.Options{
		Addr:     addr,
		PoolSize: 10,
	})
	return &Consumer{
		client:       rdb,
		stream:       "orders:events",
		group:        group,
		consumerName: consumerName,
		batchSize:    10,
		blockMS:      5000,
	}
}

// Setup 初始化消费者组（幂等操作）
func (c *Consumer) Setup(ctx context.Context) error {
	// 尝试创建消费者组，从最新消息开始（$）
	// MKSTREAM 确保 Stream 存在时自动创建
	err := c.client.XGroupCreateMkStream(ctx, c.stream, c.group, "$").Err()
	if err != nil && err.Error() != "BUSYGROUP Consumer Group name already exists" {
		return fmt.Errorf("create consumer group failed: %w", err)
	}
	log.Printf("[Consumer-%s] Group ready: %s", c.consumerName, c.group)
	return nil
}

// ProcessEvents 消息处理主循环
func (c *Consumer) ProcessEvents(ctx context.Context) {
	for {
		select {
		case <-ctx.Done():
			log.Printf("[Consumer-%s] Shutting down...", c.consumerName)
			return
		default:
		}

		// XREADGROUP 读取消息
		// ">" 表示只读取从未分配给任何消费者的新消息
		streams, err := c.client.XReadGroup(ctx, &redis.XReadGroupArgs{
			Group:    c.group,
			Consumer: c.consumerName,
			Streams:  []string{c.stream, ">"},
			Count:    c.batchSize,
			Block:    time.Duration(c.blockMS) * time.Millisecond,
		}).Result()

		if err == redis.Nil {
			// 阻塞超时，继续下一轮
			continue
		}
		if err != nil {
			log.Printf("[Consumer-%s] Read error: %v", c.consumerName, err)
			time.Sleep(time.Second)
			continue
		}

		for _, stream := range streams {
			for _, message := range stream.Messages {
				c.handleMessage(ctx, message)
			}
		}
	}
}

// handleMessage 处理单条消息
func (c *Consumer) handleMessage(ctx context.Context, msg redis.XMessage) {
	dataStr, ok := msg.Values["data"].(string)
	if !ok {
		log.Printf("[Consumer-%s] Invalid message format: %s", c.consumerName, msg.ID)
		c.ack(ctx, msg.ID) // 格式错误也 ACK，避免阻塞
		return
	}

	var event OrderEvent
	if err := json.Unmarshal([]byte(dataStr), &event); err != nil {
		log.Printf("[Consumer-%s] Unmarshal failed: %v", c.consumerName, err)
		c.ack(ctx, msg.ID)
		return
	}

	// 模拟业务处理（发送通知、更新统计等）
	log.Printf("[Consumer-%s] Processing order=%s status=%s amount=%.2f",
		c.consumerName, event.OrderID, event.Status, event.Amount)

	// 模拟处理耗时
	time.Sleep(50 * time.Millisecond)

	// 处理成功，发送 ACK
	if err := c.ack(ctx, msg.ID); err != nil {
		log.Printf("[Consumer-%s] ACK failed for %s: %v", c.consumerName, msg.ID, err)
	}
}

// ack 确认消息
func (c *Consumer) ack(ctx context.Context, ids ...string) error {
	return c.client.XAck(ctx, c.stream, c.group, ids...).Err()
}

// ClaimPending 转移长时间未处理的待处理消息（故障恢复）
func (c *Consumer) ClaimPending(ctx context.Context, minIdleTime time.Duration) error {
	// XAUTOCLAIM 自动转移 idle 超时的消息（Redis 6.2+）
	// go-redis v9 支持 XAutoClaim
	result, err := c.client.XAutoClaim(ctx, &redis.XAutoClaimArgs{
		Stream:   c.stream,
		Group:    c.group,
		Consumer: c.consumerName,
		MinIdle:  minIdleTime,
		Start:    "0-0", // 从最小 ID 开始扫描
		Count:    100,
	}).Result()

	if err != nil {
		return err
	}

	for _, msg := range result.Messages {
		log.Printf("[Consumer-%s] Claimed message %s from dead worker",
			c.consumerName, msg.ID)
		c.handleMessage(ctx, msg)
	}

	return nil
}

// Close 关闭连接
func (c *Consumer) Close() error {
	return c.client.Close()
}

func main() {
	// 通过环境变量区分不同 worker 实例
	consumerName := os.Getenv("CONSUMER_NAME")
	if consumerName == "" {
		consumerName = "worker-1"
	}

	consumer := NewConsumer("localhost:6379", "order-processors", consumerName)
	ctx, cancel := context.WithCancel(context.Background())

	if err := consumer.Setup(ctx); err != nil {
		log.Fatal(err)
	}

	// 启动消费者主循环
	go consumer.ProcessEvents(ctx)

	// 定期执行 ClaimPending（每 30 秒检查一次超时消息）
	go func() {
		ticker := time.NewTicker(30 * time.Second)
		defer ticker.Stop()
		for {
			select {
			case <-ctx.Done():
				return
			case <-ticker.C:
				if err := consumer.ClaimPending(ctx, 60*time.Second); err != nil {
					log.Printf("[Consumer-%s] Claim failed: %v", consumerName, err)
				}
			}
		}
	}()

	// 优雅退出
	sig := make(chan os.Signal, 1)
	signal.Notify(sig, syscall.SIGINT, syscall.SIGTERM)
	<-sig
	cancel()
	consumer.Close()
	log.Println("Consumer stopped gracefully")
}

# docker-compose.yml
version: "3.8"
services:
  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"
    command: redis-server --appendonly yes

# 启动 Redis
docker-compose up -d

# 启动生产者
go run producer.go

# 启动多个消费者（不同终端）
CONSUMER_NAME=worker-1 go run consumer.go
CONSUMER_NAME=worker-2 go run consumer.go
CONSUMER_NAME=worker-3 go run consumer.go

#!/bin/bash
# monitor-pel.sh - 监控待处理消息

STREAM="orders:events"
GROUP="order-processors"

while true; do
  PENDING=$(redis-cli XPENDING $STREAM $GROUP)
  echo "[$(date '+%Y-%m-%d %H:%M:%S')] PEL Count: $PENDING"

  if [ "$PENDING" -gt "0" ]; then
    # 查看详情
    redis-cli XPENDING $STREAM $GROUP - + 10
  fi

  sleep 5
done