Redis 作为消息队列：Pub/Sub 与 Streams 的选择与对比

Redis 消息队列：Pub/Sub 与 Streams 的华山论剑 ⚔️

各位好！我是你们的老朋友，江湖人称“码农一刀”的程序员老李。今天咱们不聊风花雪月，也不谈人生理想，就来聊聊Redis这个武林高手，在消息队列这个江湖里，它有两种成名绝技：Pub/Sub 和 Streams。这两位“选手”各有千秋，想要在实际项目中选择哪一个？嗯，这可就得好好掂量掂量了。

好比说，你要开一家快递公司，是选择“广播式派送”（Pub/Sub）还是“流水线式作业”（Streams）？选错了，那可是要影响效率，甚至赔本的！

所以，今天我就要化身技术解说员，用最通俗易懂的语言，最生动形象的比喻，来为大家剖析这两大绝技的特点，优劣，以及适用场景，帮助大家在实战中做出最佳选择。

一、开场：Redis “内力”深厚，消息队列只是小试牛刀 💪

Redis，全称 Remote Dictionary Server，直译过来就是“远程字典服务器”。听起来是不是有点枯燥？别怕，咱们换个说法：它就像一个身怀绝技的武林高手，拥有强大的“内力”（高性能的内存存储），精通各种“招式”（丰富的数据结构）。

它最擅长的，当然是做缓存，速度快如闪电⚡️，让你的应用体验飞一般的感觉🚀。但它可不甘心只做缓存，还想在消息队列领域也闯出一片天地。

为什么 Redis 能做消息队列？因为它可以快速地存储和检索数据，而且还支持一些消息队列常用的操作，比如发布、订阅、消费等等。

二、第一位选手登场：Pub/Sub – 广播式的浪漫，但也藏着风险 📢

首先登场的是 Pub/Sub，全称 Publish/Subscribe，翻译过来就是“发布/订阅”。它的核心思想很简单：

发布者 (Publisher)： 负责发布消息，就像一个大喇叭，把消息广播出去。
订阅者 (Subscriber)： 负责订阅感兴趣的频道 (Channel)，就像一群听众，只听自己想听的内容。

1. Pub/Sub 的工作原理：

想象一下，你是一个电台台长，想要向听众播放歌曲。你只需要把歌曲“发布”到某个频道，比如“流行音乐频道”。所有订阅了“流行音乐频道”的听众，就能立即听到这首歌曲。

这就是 Pub/Sub 的工作原理：

发布者将消息发布到特定的频道。
所有订阅了该频道的订阅者，都会收到消息。
消息的发送是“fire-and-forget”（发完就忘）式的，发布者不会关心订阅者是否成功接收。

2. Pub/Sub 的优势：

简单易用： API 简单，上手容易，几行代码就能实现消息发布和订阅。
实时性高： 消息实时推送，延迟非常低，适合对实时性要求高的场景。
解耦性好： 发布者和订阅者之间完全解耦，互不依赖，可以独立开发和部署。

3. Pub/Sub 的劣势：

消息丢失： 这是 Pub/Sub 最大的问题。如果订阅者离线，或者处理消息的速度跟不上发布者的速度，那么消息就会丢失。没办法，谁让它是广播式的呢，发出去就收不回来了。
消息无持久化： 消息不会持久化存储，一旦 Redis 服务器宕机，所有未被消费的消息都会丢失。
不支持消息确认： 发布者无法知道消息是否被成功消费，也没有重试机制。
无消费者组概念： 所有订阅者都会收到相同的消息，无法实现负载均衡。

4. Pub/Sub 的适用场景：

虽然 Pub/Sub 有一些缺点，但它在某些场景下仍然非常适用：

实时聊天： 比如在线聊天室，用户发送的消息可以实时广播给所有在线用户。
实时通知： 比如股票行情推送，服务器可以实时推送最新的股票价格给所有订阅者。
日志广播： 比如将应用程序的日志实时广播给多个监控系统。

5. Pub/Sub 的代码示例 (Python)：

import redis

# 连接 Redis
r = redis.Redis(host='localhost', port=6379, db=0)

# 发布者
def publisher():
    pub = r.pubsub()
    while True:
        message = input("请输入要发布的消息：")
        r.publish('news', message)  # 发布到 'news' 频道
        print(f"已发布消息：{message}")

# 订阅者
def subscriber():
    pub = r.pubsub()
    pub.subscribe('news')  # 订阅 'news' 频道
    for message in pub.listen():
        if message['type'] == 'message':
            print(f"收到消息：{message['data'].decode()}")

# 启动发布者和订阅者 (你需要分别运行这两个函数)
# publisher()
# subscriber()

三、第二位选手登场：Streams – 流水线式的严谨，但也稍显复杂 ⚙️

接下来登场的是 Streams，这是 Redis 5.0 引入的全新数据结构，专门为消息队列而生。它的核心思想是：

消息流 (Stream)： 一个持久化的消息队列，消息按照时间顺序存储。
消费者组 (Consumer Group)： 一组消费者，共同消费同一个消息流，实现负载均衡。
消息确认 (Acknowledgment)： 消费者成功处理消息后，需要向 Streams 确认，确保消息不会丢失。

1. Streams 的工作原理：

想象一下，你是一家工厂的厂长，需要管理一条生产线。这条生产线就是 Streams，每个产品就是一条消息。

产品源源不断地进入生产线。
多个工人 (消费者) 在不同的工位上负责不同的工序。
每个工人完成自己的工序后，需要“确认”一下，表示这个产品已经处理完毕。
如果某个工人突然罢工，那么他负责的产品会被重新分配给其他工人。

这就是 Streams 的工作原理：

消息按照时间顺序存储在消息流中。
消费者组中的多个消费者共同消费消息流。
每个消费者只消费自己负责的消息。
消费者需要确认消息，确保消息不会丢失。
如果消费者处理消息失败，或者超时未确认，那么消息会被重新分配给其他消费者。

2. Streams 的优势：

消息持久化： 消息会被持久化存储，即使 Redis 服务器宕机，消息也不会丢失。
消息确认： 消费者需要确认消息，确保消息不会丢失。
消费者组： 支持消费者组，可以实现负载均衡，提高消费能力。
消息回溯： 可以根据消息 ID 回溯到之前的消息，方便调试和分析。
消息优先级： 可以设置消息的优先级，让重要的消息优先被消费。

3. Streams 的劣势：

复杂性较高： API 相对复杂，学习成本较高。
性能稍低： 相比 Pub/Sub，Streams 的性能稍低，因为需要进行消息持久化和确认。

4. Streams 的适用场景：

Streams 适用于对消息可靠性要求非常高的场景：

订单处理： 确保每个订单都被正确处理，即使系统发生故障，也不会丢失订单。
金融交易： 确保每笔交易都被正确记录，防止资金损失。
日志收集： 确保所有日志都被收集到，方便分析和排错。

5. Streams 的代码示例 (Python)：

import redis

# 连接 Redis
r = redis.Redis(host='localhost', port=6379, db=0)

# 创建消息流
stream_name = 'my_stream'
group_name = 'my_group'

try:
    r.xgroup_create(stream_name, group_name, id='0', mkstream=True)
except redis.exceptions.ResponseError as e:
    if str(e) == 'BUSYGROUP Consumer Group name already exists':
        pass  # 忽略已存在的消费者组错误
    else:
        raise

# 生产者
def producer():
    while True:
        message = input("请输入要发布的消息：")
        r.xadd(stream_name, {'data': message})  # 添加消息到消息流
        print(f"已发布消息：{message}")

# 消费者
def consumer():
    consumer_name = 'consumer_1'  # 每个消费者需要一个唯一的名字
    while True:
        # 从消费者组中读取消息，block=0 表示不阻塞，count=1 表示每次读取一条消息
        response = r.xreadgroup(groupname=group_name, consumername=consumer_name, streams={stream_name: '>'}, count=1, block=0)

        if response:
            stream, messages = response[0]
            for message_id, message_data in messages:
                data = message_data[b'data'].decode()
                print(f"收到消息：{data}, ID: {message_id.decode()}")

                # 确认消息
                r.xack(stream_name, group_name, message_id)
                print(f"消息 {message_id.decode()} 已确认")

# 启动生产者和消费者 (你需要分别运行这两个函数)
# producer()
# consumer()

四、华山论剑：Pub/Sub vs. Streams – 谁更胜一筹？ 🏆

好了，两位选手都展示了自己的绝技，现在到了华山论剑的时刻了！

特性	Pub/Sub	Streams
消息持久化	不支持	支持
消息确认	不支持	支持
消费者组	不支持	支持
消息丢失	可能丢失	不会丢失
复杂性	简单	复杂
性能	高	稍低
适用场景	实时性要求高，消息丢失不敏感的场景	可靠性要求高，不允许消息丢失的场景
学习曲线	容易	稍难
是否需要额外配置	否	是 (需要创建消费者组)
容错性	订阅者宕机期间的消息会丢失	消费者宕机，消息会被其他消费者接管
使用案例	实时聊天，广播通知，日志广播	订单处理，金融交易，日志收集
难度系数	⭐️	⭐️⭐️⭐️
推荐指数	⭐️⭐️⭐️	⭐️⭐️⭐️⭐️

总结：

如果你需要一个简单、快速、实时的消息队列，而且可以容忍消息丢失，那么 Pub/Sub 是一个不错的选择。 它就像一个轻量级的“飞鸽传书”，速度快，但容易被风吹走。
如果你需要一个可靠、持久、可扩展的消息队列，而且不允许消息丢失，那么 Streams 是更好的选择。 它就像一个严谨的“银行金库”，安全可靠，但稍微复杂一些。

五、选择的艺术：根据场景选择合适的“武器” 🎯

选择 Pub/Sub 还是 Streams，没有绝对的答案，关键在于你的应用场景。

如果你的应用场景是“锦上添花”型的， 比如实时聊天、实时通知，那么 Pub/Sub 足够了。即使丢失一些消息，也不会造成太大的损失。
如果你的应用场景是“雪中送炭”型的， 比如订单处理、金融交易，那么 Streams 是必不可少的。任何消息丢失都可能造成严重的后果。

六、进阶之路：Redis 消息队列的更多可能性 🚀

除了 Pub/Sub 和 Streams，Redis 还有一些其他的消息队列方案，比如：

List： 使用 List 的 LPUSH 和 BRPOP 命令可以实现一个简单的消息队列。但是 List 不支持消息确认和消费者组，可靠性较低。
Sorted Set： 使用 Sorted Set 可以实现一个优先级队列，让重要的消息优先被消费。

当然，这些方案相对来说比较简单，功能也比较有限，不如 Streams 强大。

七、尾声：Redis 武林，精彩纷呈 🎭

好了，今天的 Redis 消息队列之旅就到此结束了。希望通过今天的讲解，大家对 Pub/Sub 和 Streams 有了更深入的了解。

记住，选择合适的“武器”，才能在 Redis 武林中闯出一片天地！ 💪

最后，祝大家编码愉快，Bug 远离！ 🍻

Redis 消息队列：Pub/Sub 与 Streams 的华山论剑 ⚔️

发表回复 取消回复

发表回复取消回复