Redis 消息队列深度对比：Pub/Sub、Streams 与 Kafka/RabbitMQ 选型指南

# 订阅指定频道
SUBSCRIBE orders

# 通过 Glob 模式批量订阅
PSUBSCRIBE orders.*

# 发布消息到频道
PUBLISH orders '{"order_id":"20260816001","status":"paid"}'

# 服务端查看订阅统计
PUBSUB CHANNELS
PUBSUB NUMSUB orders

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

import redis
import json
import threading

r = redis.Redis(host='localhost', port=6379, decode_responses=True)

# 订阅者
pubsub = r.pubsub()
pubsub.subscribe('orders')

def listener():
    for message in pubsub.listen():
        if message['type'] == 'message':
            data = json.loads(message['data'])
            print(f"Received order: {data['order_id']}")

threading.Thread(target=listener, daemon=True).start()

# 发布者
r.publish('orders', json.dumps({"order_id": "ORD-001", "status": "paid"}))

package main

import (
    "context"
    "encoding/json"
    "fmt"
    "github.com/redis/go-redis/v9"
)

type OrderEvent struct {
    OrderID string `json:"order_id"`
    Status  string `json:"status"`
}

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

    // 发布
    event := OrderEvent{OrderID: "ORD-001", Status: "paid"}
    data, _ := json.Marshal(event)
    rdb.Publish(ctx, "orders", data)

    // 订阅
    pubsub := rdb.Subscribe(ctx, "orders")
    defer pubsub.Close()

    ch := pubsub.Channel()
    for msg := range ch {
        fmt.Printf("Received: %s\n", msg.Payload)
    }
}

import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPubSub;

public class PubSubDemo {
    public static void main(String[] args) {
        Jedis jedis = new Jedis("localhost", 6379);

        // 订阅（需在独立线程执行）
        new Thread(() -> {
            jedis.subscribe(new JedisPubSub() {
                @Override
                public void onMessage(String channel, String message) {
                    System.out.println("Channel: " + channel + ", Message: " + message);
                }
            }, "orders");
        }).start();

        // 发布
        jedis.publish("orders", "{\"order_id\":\"ORD-001\"}");
    }
}

# XADD：发布消息，自动分配 ID，限制 Stream 长度
XADD events:orders MAXLEN ~ 10000 * order_id ORD-001 status paid amount 299.00

# XREAD：阻塞读取新消息
XREAD BLOCK 5000 STREAMS events:orders $

# XRANGE：按 ID 范围查询历史
XRANGE events:orders - + COUNT 10

# XLEN：获取 Stream 长度
XLEN events:orders

# 创建消费者组（从最新消息开始）
XGROUP CREATE events:orders order_group $ MKSTREAM

# 消费者读取消息（> 表示读取未分配的新消息）
XREADGROUP GROUP order_group worker-1 COUNT 5 BLOCK 3000 \
  STREAMS events:orders >

# 确认消息已处理
XACK events:orders order_group 1723779600000-0

# 查看待处理消息
XPENDING events:orders order_group - + 10

# 自动转移超时消息（Redis 6.2+）
XAUTOCLAIM events:orders order_group worker-2 60000 - COUNT 100

package main

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

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

type Event struct {
    OrderID string  `json:"order_id"`
    Status  string  `json:"status"`
    Amount  float64 `json:"amount"`
}

func main() {
    ctx := context.Background()
    rdb := redis.NewClient(&redis.Options{Addr: "localhost:6379", PoolSize: 10})
    defer rdb.Close()

    stream := "events:orders"
    group := "order_processors"
    consumer := os.Getenv("CONSUMER_NAME")
    if consumer == "" {
        consumer = "worker-1"
    }

    // 创建消费者组（幂等）
    err := rdb.XGroupCreateMkStream(ctx, stream, group, "$").Err()
    if err != nil && err.Error() != "BUSYGROUP Consumer Group name already exists" {
        log.Fatal(err)
    }

    ctx, cancel := context.WithCancel(ctx)
    defer cancel()

    // 信号处理
    sig := make(chan os.Signal, 1)
    signal.Notify(sig, syscall.SIGINT, syscall.SIGTERM)

    // 消费循环
    go func() {
        for {
            select {
            case <-ctx.Done():
                return
            default:
            }

            streams, err := rdb.XReadGroup(ctx, &redis.XReadGroupArgs{
                Group:    group,
                Consumer: consumer,
                Streams:  []string{stream, ">"},
                Count:    10,
                Block:    5 * time.Second,
            }).Result()

            if err == redis.Nil {
                continue
            }
            if err != nil {
                log.Printf("Read error: %v", err)
                time.Sleep(time.Second)
                continue
            }

            for _, s := range streams {
                for _, msg := range s.Messages {
                    processMessage(ctx, rdb, stream, group, msg)
                }
            }
        }
    }()

    // 定期清理超时消息
    go func() {
        ticker := time.NewTicker(30 * time.Second)
        defer ticker.Stop()
        for {
            select {
            case <-ctx.Done():
                return
            case <-ticker.C:
                result, err := rdb.XAutoClaim(ctx, &redis.XAutoClaimArgs{
                    Stream:   stream,
                    Group:    group,
                    Consumer: consumer,
                    MinIdle:  60 * time.Second,
                    Start:    "0-0",
                    Count:    100,
                }).Result()
                if err != nil {
                    log.Printf("AutoClaim error: %v", err)
                    continue
                }
                for _, msg := range result.Messages {
                    log.Printf("Claimed message %s", msg.ID)
                    processMessage(ctx, rdb, stream, group, msg)
                }
            }
        }
    }()

    <-sig
    cancel()
}

func processMessage(ctx context.Context, rdb *redis.Client, stream, group string, msg redis.XMessage) {
    data, ok := msg.Values["data"].(string)
    if !ok {
        rdb.XAck(ctx, stream, group, msg.ID)
        return
    }

    var event Event
    if err := json.Unmarshal([]byte(data), &event); err != nil {
        rdb.XAck(ctx, stream, group, msg.ID)
        return
    }

    fmt.Printf("Processing order=%s status=%s\n", event.OrderID, event.Status)
    time.Sleep(50 * time.Millisecond)

    if err := rdb.XAck(ctx, stream, group, msg.ID).Err(); err != nil {
        log.Printf("ACK failed: %v", err)
    }
}

from kafka import KafkaProducer
import json

# Kafka 生产者
producer = KafkaProducer(
    bootstrap_servers=['localhost:9092'],
    value_serializer=lambda v: json.dumps(v).encode('utf-8'),
    compression_type='snappy'
)

producer.send('orders', {'order_id': 'ORD-001', 'status': 'paid'})
producer.flush()

# Redis Streams 生产者（对比）
import redis
r = redis.Redis()
r.xadd('events:orders', {'data': json.dumps({'order_id': 'ORD-001', 'status': 'paid'})}, maxlen=10000, approximate=True)

import com.rabbitmq.client.*;

public class RabbitMQConsumer {
    private static final String QUEUE_NAME = "orders";

    public static void main(String[] args) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();

        channel.queueDeclare(QUEUE_NAME, true, false, false, null);
        channel.basicQos(10); // prefetch count

        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody(), "UTF-8");
            try {
                processMessage(message);
                channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
            } catch (Exception e) {
                channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false, true);
            }
        };

        channel.basicConsume(QUEUE_NAME, false, deliverCallback, consumerTag -> {});
    }

    static void processMessage(String message) {
        System.out.println("Received: " + message);
    }
}

import redis
import json
import time
from datetime import datetime, timedelta

r = redis.Redis(host='localhost', port=6379, decode_responses=True)

DELAYED_QUEUE = 'delayed:queue'
EVENT_STREAM = 'events:delayed'

def add_delayed_event(event: dict, delay_seconds: int) -> str:
    """添加延迟事件"""
    execute_at = time.time() + delay_seconds
    msg_id = r.xadd(EVENT_STREAM, {'data': json.dumps(event)})
    r.zadd(DELAYED_QUEUE, {msg_id: execute_at})
    return msg_id

def process_delayed_events():
    """轮询并转移到期事件到活跃 Stream"""
    while True:
        now = time.time()
        # 获取到期的消息
        expired = r.zrangebyscore(DELAYED_QUEUE, 0, now, start=0, num=100)
        
        if not expired:
            time.sleep(0.1)
            continue
        
        for msg_id in expired:
            # 从延迟队列移除
            r.zrem(DELAYED_QUEUE, msg_id)
            
            # 读取原始消息并写入活跃 Stream（或直接处理）
            # 实际应用中可维护第二个活跃 Stream 供消费者组消费
            print(f"Delayed event triggered: {msg_id}")

# 使用示例
add_delayed_event(
    {'type': 'order_timeout', 'order_id': 'ORD-001'},
    delay_seconds=300  # 5 分钟后触发
)

add_delayed_event(
    {'type': 'reminder', 'user_id': 'U10086', 'message': '会议即将开始'},
    delay_seconds=3600  # 1 小时后触发
)

# 优化：使用阻塞等待 + 信号
def optimized_poll():
    while True:
        now = time.time()
        expired = r.zrangebyscore(DELAYED_QUEUE, 0, now, start=0, num=100)
        
        if expired:
            for msg_id in expired:
                r.zrem(DELAYED_QUEUE, msg_id)
                process_event(msg_id)
            continue
        
        # 获取下一个到期时间，阻塞等待
        next_item = r.zrange(DELAYED_QUEUE, 0, 0, withscores=True)
        if next_item:
            wait_time = max(0, next_item[0][1] - now)
            time.sleep(min(wait_time, 1.0))  # 最多等待 1 秒
        else:
            time.sleep(1.0)

import { Queue, Worker, Job } from 'bullmq';
import Redis from 'ioredis';

const connection = new Redis({ host: 'localhost', port: 6379, maxRetriesPerRequest: null });

// 定义队列
const emailQueue = new Queue('email', { connection });

// 添加任务（支持延迟）
await emailQueue.add('send-welcome', 
  { to: 'user@example.com', template: 'welcome' },
  { delay: 5000, attempts: 3, backoff: { type: 'exponential', delay: 2000 } }
);

// Worker 消费
const worker = new Worker('email', async (job: Job) => {
  console.log(`Processing job ${job.id}: ${job.name}`);
  await sendEmail(job.data.to, job.data.template);
}, { connection, concurrency: 5 });

// 事件监听
worker.on('completed', (job) => {
  console.log(`Job ${job.id} completed`);
});

worker.on('failed', (job, err) => {
  console.error(`Job ${job?.id} failed:`, err);
});

from redis import Redis
from rq import Queue, Worker
from rq.job import Job
import time

# 连接 Redis
redis_conn = Redis()
q = Queue(connection=redis_conn)

# 定义任务函数
def send_notification(user_id: str, message: str):
    time.sleep(2)
    print(f"Notification sent to {user_id}: {message}")
    return {"status": "sent", "user": user_id}

# 入队（支持延迟）
job = q.enqueue(send_notification, 'U10086', '订单已发货', job_timeout=30)

# 获取任务状态
print(f"Job ID: {job.id}, Status: {job.get_status()}")

# 启动 Worker（命令行）
# rq worker --with-scheduler

# 延迟任务（需要 rq-scheduler）
from rq_scheduler import Scheduler
scheduler = Scheduler(connection=redis_conn)
scheduler.enqueue_at(datetime(2026, 8, 16, 14, 0), send_notification, 'U10086', '定时提醒')

# 为每个租户创建独立 Stream
XADD tenant:123:events * action login user_id U001 ip 1.2.3.4
XADD tenant:123:events * action update_profile user_id U001 field avatar

# 按时间范围查询（最近一小时）
XRANGE tenant:123:events 1723776000000-0 +

# 审计分析：统计某用户的操作
# 通过 XRANGE 拉取后业务层过滤

# Pipeline 批量写入示例
redis-cli --pipe <<'EOF'
XADD events * field1 value1
XADD events * field2 value2
XADD events * field3 value3
EOF

// Go Pipeline 批量写入
pipe := rdb.Pipeline()
for i := 0; i < 1000; i++ {
    pipe.XAdd(ctx, &redis.XAddArgs{
        Stream: "events",
        Values: map[string]interface{}{"n": i},
    })
}
cmders, err := pipe.Exec(ctx)