8.3 消息队列

8.3 消息队列

Redis 应用案例：消息队列 (8.3) - 代码实践与详解

消息队列（Message Queue，简称 MQ）在现代分布式系统中扮演着至关重要的角色。它允许不同的服务或组件之间通过异步的方式传递消息，从而实现解耦、提高系统吞吐量、增强系统稳定性。Redis，作为一款高性能的键值存储数据库，凭借其丰富的数据结构和强大的功能，同样可以胜任消息队列的角色。

1. Redis 作为消息队列的优势与适用场景

2. 基于 Redis List 的简单消息队列

* 2.1 LPUSH/RPOP 实现队列

* 2.2 RPUSH/LPOP 实现队列

* 2.3 BLPOP/BRPOP 实现阻塞队列

* 2.4 代码实践：Python + redis-py

* 2.5 优缺点分析
​

3. 基于 Redis Pub/Sub 的发布订阅模式

* 3.1 PUBLISH/SUBSCRIBE 实现

* 3.2 PSUBSCRIBE 实现模式匹配订阅

* 3.3 代码实践：Python + redis-py

* 3.4 优缺点分析
​

4. 基于 Redis Stream 的持久化消息队列

* 4.1 Stream 数据结构详解

* 4.2 XADD 添加消息

* 4.3 XREAD 读取消息 (消费者组与非消费者组)

* 4.4 XGROUP 创建消费者组

* 4.5 XACK 消息确认

* 4.6 XPENDING/XCLAIM 处理Pending消息

* 4.7 代码实践：Python + redis-py

* 4.8 优缺点分析
​

5. 消息队列选型：List, Pub/Sub, Stream 的对比与选择

6. 高级主题：消息持久化、消息可靠性、消息顺序性

7. 总结

1. Redis 作为消息队列的优势与适用场景

Redis 作为消息队列，具有以下显著优势：

• 高性能: Redis 基于内存操作，读写速度极快，可以处理高并发的消息生产和消费场景。

• 简单易用: Redis 的 API 简洁明了，易于上手和集成。

• 丰富的数据结构: Redis 提供了 List, Pub/Sub, Stream 等多种数据结构，可以满足不同类型的消息队列需求。

• 功能丰富: 除了基本的消息队列功能，Redis 还提供了事务、持久化、集群等高级特性，可以构建更复杂和可靠的消息系统。

• 成熟稳定: Redis 经过长时间的生产环境验证，具有很高的稳定性和可靠性。

适用场景:

• 异步任务处理: 将耗时的任务放入队列，异步处理，提高响应速度。例如：发送邮件、处理订单、数据分析等。

• 服务解耦: 不同服务之间通过消息队列通信，降低服务之间的耦合度，提高系统的可维护性和可扩展性。

• 流量削峰: 在高并发场景下，消息队列可以缓冲请求，平滑流量，避免系统过载。

• 日志收集: 将日志数据写入消息队列，异步处理和分析。

• 实时消息推送: 使用 Pub/Sub 或 Stream 实现实时消息推送功能，例如：聊天室、实时通知等。

需要注意的是，Redis 作为消息队列，其可靠性和持久性相对专业的消息队列系统（如 Kafka, RabbitMQ）会弱一些。 Redis 默认数据存储在内存中，虽然可以通过 RDB 和 AOF 进行持久化，但在极端情况下（例如 Redis 实例宕机且未及时持久化），可能会丢失部分消息。因此，对于对消息可靠性要求极高的场景，可能需要结合 Redis 的持久化机制，或者考虑使用更专业的消息队列系统。 然而，对于大部分中小型应用，Redis 仍然是一个非常优秀且高效的消息队列解决方案。

2. 基于 Redis List 的简单消息队列

Redis List 是一种有序的字符串列表，非常适合用来实现简单的消息队列。List 提供了 LPUSH, RPUSH, LPOP, RPOP, BLPOP, BRPOP 等命令，可以方便地进行消息的生产和消费。

2.1 LPUSH/RPOP 实现队列 (先进后出 - Stack 栈)

• 生产者 (Producer): 使用 LPUSH 命令将消息从 List 的左侧（头部）推入。

• 消费者 (Consumer): 使用 RPOP 命令从 List 的右侧（尾部）弹出消息。

这种方式实际上实现的是一个 栈 (Stack) 的结构，消息是 先进后出 (FILO) 的顺序。虽然不太符合通常消息队列的 先进先出 (FIFO) 语义，但在某些特定场景下也可能适用。

2.2 RPUSH/LPOP 实现队列 (先进先出 - Queue 队列)

• 生产者 (Producer): 使用 RPUSH 命令将消息从 List 的右侧（尾部）推入。

• 消费者 (Consumer): 使用 LPOP 命令从 List 的左侧（头部）弹出消息。

这种方式实现的是一个标准的 队列 (Queue) 结构，消息是 先进先出 (FIFO) 的顺序，符合大部分消息队列的应用场景。

2.3 BLPOP/BRPOP 实现阻塞队列

LPOP 和 RPOP 是非阻塞的，如果 List 为空，会立即返回 nil。 在消费者端，通常需要轮询 List 是否有新消息，这会浪费 CPU 资源。

Redis 提供了 阻塞版本的弹出命令 BLPOP 和 BRPOP (Blocking List Pop)。

• BLPOP key [key ...] timeout: 阻塞式地从左侧弹出一个或多个 List 中的元素。如果所有 List 都是空的，连接将被阻塞，直到等待超时或有元素可以弹出。

• BRPOP key [key ...] timeout: 阻塞式地从右侧弹出一个或多个 List 中的元素。如果所有 List 都是空的，连接将被阻塞，直到等待超时或有元素可以弹出。

timeout 参数指定阻塞的超时时间，单位为秒。如果 timeout 设置为 0，则表示永久阻塞，直到有元素可以弹出。

使用 BLPOP 和 BRPOP，消费者可以无需轮询，当 List 中没有消息时，消费者会被阻塞，直到有新消息到达或者超时，从而有效地节省 CPU 资源。

2.4 代码实践：Python + redis-py

以下代码示例演示了使用 Python 和 redis-py 库，基于 Redis List 实现一个简单的消息队列 (FIFO - 先进先出)。

生产者 (producer.py):

import redis
import time
import json
# Redis 连接配置
redis_host = "localhost"
redis_port = 6379
redis_db = 0
queue_name = "my_queue"  # 队列名称
# 创建 Redis 连接
r = redis.Redis(host=redis_host, port=redis_port, db=redis_db)
def produce_message(message_data):
    """生产消息并推入队列"""
    message_json = json.dumps(message_data)
    r.rpush(queue_name, message_json)  # 使用 RPUSH 推入队列尾部
    print(f"生产者：消息 '{message_json}' 已推入队列 '{queue_name}'")
if __name__ == "__main__":
    for i in range(5):
        message = {"task_id": i, "payload": f"任务内容 {i}"}
        produce_message(message)
        time.sleep(1)  # 模拟生产间隔
​

消费者 (consumer.py):

import redis
import json
# Redis 连接配置 (与生产者相同)
redis_host = "localhost"
redis_port = 6379
redis_db = 0
queue_name = "my_queue"  # 队列名称
# 创建 Redis 连接
r = redis.Redis(host=redis_host, port=redis_port, db=redis_db)
def consume_message():
    """从队列中消费消息 (阻塞式)"""
    while True:
        try:
            # 使用 BLPOP 阻塞式地从队列头部弹出消息，超时时间设置为 5 秒 (0 表示永久阻塞)
            _, message_json = r.blpop(queue_name, timeout=5)
            if message_json:
                message_data = json.loads(message_json.decode('utf-8'))
                print(f"消费者：收到消息 '{message_data}' 来自队列 '{queue_name}'")
                # 在这里处理消息，例如执行任务
                process_message(message_data)
            else:
                print("消费者：队列为空，等待新消息...")
        except redis.exceptions.ConnectionError as e:
            print(f"消费者：Redis 连接错误: {e}")
            break  # 退出循环，可以根据实际情况进行重试或处理
        except Exception as e:
            print(f"消费者：处理消息时发生错误: {e}")
def process_message(message):
    """消息处理函数 (示例)"""
    print(f"消费者：开始处理任务 ID: {message['task_id']}, 内容: {message['payload']}")
    # 模拟任务处理时间
    # time.sleep(2)
    print(f"消费者：任务 ID: {message['task_id']} 处理完成")
if __name__ == "__main__":
    print("消费者启动，等待消息...")
    consume_message()
​

运行示例:

1. 确保 Redis 服务已启动。

2. 运行 producer.py 生产消息。

3. 运行 consumer.py 消费消息。

你会看到生产者将消息推入队列，消费者阻塞等待并消费消息。当队列为空时，消费者会等待一段时间后再次检查。

2.5 优缺点分析 (List 队列)

优点:

• 实现简单: 基于 List 命令即可快速实现。

• 性能高: Redis List 操作性能非常高。

• 支持阻塞: BLPOP/BRPOP 支持阻塞式消费，节省 CPU 资源。

• 有序性: List 保证消息的顺序性 (FIFO 或 FILO)。

缺点:

• 消息可靠性较弱: 消息存储在内存中，持久化需要依赖 Redis 的 RDB 或 AOF 机制，极端情况下可能丢失消息。

• 不支持消息确认机制 (ACK): 消费者从 List 中弹出消息后，消息就从 List 中移除了。如果消费者处理消息失败或崩溃，消息会丢失。需要业务逻辑自行实现消息重试或补偿机制。

• 不支持消息分组和广播: List 队列是点对点模式，一个消息只能被一个消费者消费。不支持发布订阅模式。

• 功能相对简单: 相比专业消息队列系统，功能较为基础。

3. 基于 Redis Pub/Sub 的发布订阅模式

Redis Pub/Sub (Publish/Subscribe) 是一种消息通信模式，它允许消息的发布者 (Publisher) 将消息发布到指定的频道 (Channel)，而订阅了该频道的订阅者 (Subscriber) 都会收到该消息。 Pub/Sub 实现了消息的广播，一个消息可以被多个订阅者消费。

3.1 PUBLISH/SUBSCRIBE 实现

• 发布者 (Publisher): 使用 PUBLISH channel message 命令将消息发布到指定的频道 channel。

• 订阅者 (Subscriber): 使用 SUBSCRIBE channel [channel ...] 命令订阅一个或多个频道。一旦订阅成功，订阅者会一直监听该频道上的消息。

3.2 PSUBSCRIBE 实现模式匹配订阅

PSUBSCRIBE pattern [pattern ...] 命令允许订阅者使用 模式匹配 的方式订阅频道。可以使用通配符 * 和 ? 来匹配频道名称。

• * 匹配任意多个字符。

• ? 匹配任意一个字符。

例如，PSUBSCRIBE news.* 会订阅所有以 news. 开头的频道，例如 news.sports, news.finance, news.tech 等。

3.3 代码实践：Python + redis-py

发布者 (publisher.py):

import redis
import time
# Redis 连接配置
redis_host = "localhost"
redis_port = 6379
redis_db = 0
channel_name = "news.tech"  # 频道名称
# 创建 Redis 连接
r = redis.Redis(host=redis_host, port=redis_port, db=redis_db)
def publish_message(channel, message):
    """发布消息到指定频道"""
    r.publish(channel, message)
    print(f"发布者：消息 '{message}' 已发布到频道 '{channel}'")
if __name__ == "__main__":
    for i in range(5):
        message = f"科技快讯 {i}"
        publish_message(channel_name, message)
        time.sleep(1)
​

订阅者 (subscriber.py):

import redis
# Redis 连接配置 (与发布者相同)
redis_host = "localhost"
redis_port = 6379
redis_db = 0
channel_name = "news.tech"  # 频道名称
# 创建 Redis 连接
r = redis.Redis(host=redis_host, port=redis_port, db=redis_db)
def subscribe_channel(channel):
    """订阅频道并接收消息"""
    pubsub = r.pubsub()
    pubsub.subscribe(channel)  # 订阅频道
    print(f"订阅者：已订阅频道 '{channel}'，等待消息...")
    for message in pubsub.listen():
        if message['type'] == 'message':
            channel_received = message['channel'].decode('utf-8')
            data = message['data'].decode('utf-8')
            print(f"订阅者：收到消息 '{data}' 来自频道 '{channel_received}'")
            # 在这里处理消息
if __name__ == "__main__":
    subscribe_channel(channel_name)
​

运行示例:

1. 确保 Redis 服务已启动。

2. 运行 publisher.py 发布消息。

3. 运行 subscriber.py 订阅频道并接收消息。

你可以启动多个 subscriber.py 实例，它们都会收到发布者发布到 news.tech 频道的消息。

3.4 优缺点分析 (Pub/Sub)

优点:

• 实现简单: 基于 Pub/Sub 命令即可快速实现。

• 高性能: Pub/Sub 性能很高，适合实时消息推送场景。

• 广播模式: 支持消息广播，一个消息可以被多个订阅者消费。

• 解耦性强: 发布者和订阅者之间完全解耦，无需知道对方的存在。

缺点:

• 消息不可靠: 消息是 Fire-and-Forget (即发即弃) 的，不保证消息的持久性和可靠性。 如果订阅者在消息发布时离线，或者处理消息失败，消息会丢失。没有消息队列的存储机制，消息不会被持久化和重试。

• 无消息确认机制 (ACK): 发布者无法知道消息是否被成功接收和处理。

• 无消息持久化: 消息只存在于内存中，Redis 重启后消息会丢失。

• 功能相对简单: 功能较为基础，不适合复杂的业务场景。

适用场景:

• 实时消息推送: 例如：聊天室、实时通知、在线游戏等。

• 事件通知: 系统内部组件之间的事件广播。

• 配置更新广播: 将配置更新广播到所有订阅者。

4. 基于 Redis Stream 的持久化消息队列

Redis Stream 是 Redis 5.0 版本引入的一种新的数据结构，它专门为消息队列场景设计，提供了更强大、更可靠的消息队列功能，弥补了 List 和 Pub/Sub 在消息可靠性和功能性上的不足。

4.1 Stream 数据结构详解

Stream 是一种 持久化的、有序的消息日志。它可以看作是一个 可追加 (append-only) 的日志结构，新的消息会不断追加到 Stream 的末尾。

Stream 的关键特性包括：

• 消息持久化: Stream 中的消息可以持久化到磁盘，即使 Redis 重启，消息也不会丢失 (取决于 Redis 的持久化配置)。

• 消息唯一 ID: 每个消息都有一个唯一的时间戳 ID，用于保证消息的顺序性和追踪。

• 消费者组 (Consumer Groups): Stream 引入了消费者组的概念，允许多个消费者组成一个组，共同消费 Stream 中的消息。消费者组保证同一个消息只会被组内的一个消费者消费，实现了消息的负载均衡。

• 消息确认机制 (ACK): 消费者在消费消息后需要进行确认 (ACK)，Stream 会跟踪哪些消息被消费了，哪些消息还没有被确认。

• Pending 消息列表 (PEL): Stream 会维护一个 Pending Entries List (PEL)，记录所有已被消费者组中的消费者获取但尚未确认的消息。用于处理消费者崩溃或消息处理失败的情况。

• 消息回溯: Stream 允许消费者从任意位置开始读取消息，支持消息回溯和重放。

4.2 XADD 添加消息

XADD key ID field value [field value ...] 命令用于向 Stream key 中添加新的消息。

• key: Stream 的名称。

• ID: 消息 ID。可以使用 * 表示由 Redis 自动生成 ID (推荐)。也可以手动指定 ID，但必须是 timestamp-sequence 格式，且必须大于 Stream 中已有的最大 ID。

• field value [field value ...]: 消息的内容，由一个或多个键值对组成。

示例:

XADD mystream * message_type log level info message "User login successful"
​

4.3 XREAD 读取消息 (消费者组与非消费者组)

XREAD [COUNT count] [BLOCK milliseconds] [STREAMS key [key ...] ID [ID ...]] 命令用于从一个或多个 Stream 中读取消息。

• COUNT count: 指定每次读取的最大消息数量。

• BLOCK milliseconds: 指定阻塞的超时时间，单位为毫秒。如果设置为 0，则表示非阻塞读取。

• STREAMS key [key ...] ID [ID ...]: 指定要读取的 Stream 和起始 ID。

读取模式:

• 非消费者组模式: 使用 XREAD 命令直接读取 Stream，每个消费者独立读取 Stream 中的所有消息。类似于 List 队列的模式。

  • ID: 可以使用 $ 表示从 Stream 的末尾开始读取新消息 (类似 tail -f)。可以使用 0 表示从 Stream 的开头开始读取所有消息。

• 消费者组模式: 使用 XREADGROUP GROUP groupname consumername [COUNT count] [BLOCK milliseconds] [STREAMS key [key ...] ID [ID ...]] 命令从消费者组中读取消息。

  • GROUP groupname: 消费者组名称。

  • consumername: 消费者名称，组内唯一。

  • ID: 可以使用 > 表示从消费者组的未消费消息的末尾开始读取新消息 (只读取新消息)。可以使用消息 ID 或 0 从特定的位置开始读取历史消息。

4.4 XGROUP 创建消费者组

XGROUP CREATE key groupname ID [$ | 0] [MKSTREAM] 命令用于创建消费者组。

• key: Stream 的名称。

• groupname: 消费者组名称。

• ID: 消费者组的起始 ID。通常设置为 $ 表示从 Stream 末尾开始消费新消息，设置为 0 表示从 Stream 开头开始消费所有消息。

• MKSTREAM: 可选参数，如果 Stream 不存在，则自动创建 Stream。

4.5 XACK 消息确认

XACK key groupname ID [ID ...] 命令用于消费者组中的消费者确认消息已被成功处理。

• key: Stream 的名称。

• groupname: 消费者组名称。

• ID [ID ...]: 要确认的消息 ID 列表。

4.6 XPENDING/XCLAIM 处理 Pending 消息

• XPENDING key groupname [IDLE min-idle-time] [[IDLE min-idle-time] COUNT count] [consumername]: 查看消费者组的 Pending Entries List (PEL)，即已分发给消费者但尚未确认的消息列表。

  • IDLE min-idle-time: 可选参数，筛选空闲时间超过 min-idle-time 毫秒的消息。

  • COUNT count: 可选参数，限制返回的消息数量。

  • consumername: 可选参数，筛选特定消费者的 Pending 消息。

• XCLAIM key groupname consumername min-idle-time ID [ID ...] [RETRYCOUNT count] [FORCE]: 用于将 Pending 消息的所有权从一个消费者转移到另一个消费者。通常用于处理消费者崩溃或消息处理超时的情况。

  • min-idle-time: 消息的最小空闲时间，毫秒。只有空闲时间超过这个值，才能被认领。

  • ID [ID ...]: 要认领的消息 ID 列表。

  • RETRYCOUNT count: 可选参数，设置消息的重试次数。

  • FORCE: 可选参数，强制认领消息，即使消息的空闲时间未达到 min-idle-time。

4.7 代码实践：Python + redis-py

生产者 (stream_producer.py):

import redis
import time
# Redis 连接配置
redis_host = "localhost"
redis_port = 6379
redis_db = 0
stream_name = "mystream"  # Stream 名称
# 创建 Redis 连接
r = redis.Redis(host=redis_host, port=redis_port, db=redis_db)
def produce_stream_message(message_data):
    """生产消息并推入 Stream"""
    message_id = r.xadd(stream_name, message_data)  # 使用 XADD 添加消息
    print(f"生产者：消息 '{message_data}' 已推入 Stream '{stream_name}'，消息 ID: {message_id}")
if __name__ == "__main__":
    for i in range(5):
        message = {"task_type": "job", "payload": f"Stream 任务内容 {i}"}
        produce_stream_message(message)
        time.sleep(1)
​

消费者 (stream_consumer.py):

import redis
import time
# Redis 连接配置 (与生产者相同)
redis_host = "localhost"
redis_port = 6379
redis_db = 0
stream_name = "mystream"  # Stream 名称
group_name = "mygroup"     # 消费者组名称
consumer_name = "consumer1" # 消费者名称
# 创建 Redis 连接
r = redis.Redis(host=redis_host, port=redis_port, db=redis_db)
def consume_stream_message():
    """从 Stream 消费者组消费消息"""
    try:
        # 尝试创建消费者组 (如果不存在)
        r.xgroup_create(stream_name, group_name, id='0', mkstream=True) # 从 Stream 开头消费历史消息
        # r.xgroup_create(stream_name, group_name, id='$', mkstream=True) # 从 Stream 末尾消费新消息
    except redis.exceptions.ResponseError as e:
        if "BUSYGROUP Consumer Group name already exists" not in str(e):
            raise e
        print(f"消费者组 '{group_name}' 已存在，跳过创建")
    while True:
        try:
            # 使用 XREADGROUP 从消费者组读取消息 (阻塞式)
            response = r.xreadgroup(
                groupname=group_name,
                consumername=consumer_name,
                streams={stream_name: '>'},  # '>' 表示从消费者组未消费消息的末尾开始读取
                count=1,
                block=5000  # 阻塞 5 秒
            )
            if response:
                stream_data = response[0][1]  # 获取 Stream 数据列表
                for message_id, message_data in stream_data:
                    message_id_str = message_id.decode('utf-8')
                    print(f"消费者 '{consumer_name}'：收到消息 '{message_data}' 来自 Stream '{stream_name}'，消息 ID: {message_id_str}")
                    # 在这里处理消息
                    process_stream_message(message_data)
                    # 消息确认 (ACK)
                    r.xack(stream_name, group_name, message_id)
                    print(f"消费者 '{consumer_name}'：消息 ID: {message_id_str} 已确认")
            else:
                print(f"消费者 '{consumer_name}'：Stream '{stream_name}' 没有新消息，等待...")
        except redis.exceptions.ConnectionError as e:
            print(f"消费者 '{consumer_name}'：Redis 连接错误: {e}")
            break
        except Exception as e:
            print(f"消费者 '{consumer_name}'：处理消息时发生错误: {e}")
def process_stream_message(message):
    """Stream 消息处理函数 (示例)"""
    print(f"消费者 '{consumer_name}'：开始处理 Stream 任务，内容: {message}")
    # 模拟任务处理时间
    # time.sleep(2)
    print(f"消费者 '{consumer_name}'：Stream 任务处理完成")
if __name__ == "__main__":
    print(f"消费者 '{consumer_name}' 启动，加入消费者组 '{group_name}'，等待 Stream 消息...")
    consume_stream_message()
​

运行示例:

1. 确保 Redis 服务已启动。

2. 运行 stream_producer.py 生产消息。

3. 运行 stream_consumer.py 消费消息。

你可以启动多个 stream_consumer.py 实例，并使用相同的 group_name，它们会组成一个消费者组，共同消费 Stream 中的消息，实现负载均衡。

4.8 优缺点分析 (Stream)

优点:

• 消息持久化: 支持消息持久化，消息可靠性高 (取决于 Redis 持久化配置)。

• 消息确认机制 (ACK): 提供消息确认机制，保证消息至少被消费一次 (at-least-once delivery)。

• 消费者组: 支持消费者组，实现消息的负载均衡和并行消费。

• Pending 消息处理: 支持 Pending 消息列表和认领机制，处理消费者崩溃或消息处理失败的情况，提高消息的可靠性。

• 消息顺序性: 保证 Stream 内消息的顺序性。

• 功能强大: 功能丰富，适合构建复杂的、可靠的消息队列系统。

缺点:

• 实现相对复杂: 相比 List 和 Pub/Sub，Stream 的使用和概念相对复杂一些。

• 性能略低于 List: 由于功能更强大，Stream 的性能相比 List 略有下降，但仍然非常高效。

• 资源消耗略高: 维护 Stream 的数据结构和消费者组信息，资源消耗相对 List 略高。

适用场景:

• 需要高可靠性和持久性的消息队列场景。

• 需要消息确认和重试机制的场景。

• 需要消费者组和负载均衡的场景。

• 复杂的异步任务处理、订单处理、金融交易等关键业务场景。

5. 消息队列选型：List, Pub/Sub, Stream 的对比与选择