当前位置：首页 > news >正文

7、Redis 队列与 Stream

news 文章来源：https://blog.csdn.net/qq_29434541/article/details/140249372 2025/5/5 20:09:20

引言

Redis 自 5.0 版本起引入了一种新的数据结构——Stream。这种数据结构不仅增加了 Redis 的数据处理能力，还使其在消息队列和数据流处理方面更具竞争力。Stream 提供了持久化、多播、消费组等功能，可以满足多种复杂的数据处理需求。

1. Redis Stream 概述

Stream 是 Redis 5.0 引入的一种新的数据结构，类似于 Kafka 的设计。它是一个强大的支持多播的可持久化消息队列，适用于实时数据处理和消息传递。

基本结构

Stream 由消息链表组成，每个消息都有一个唯一的 ID 和对应的内容。消息 ID 是由时间戳和序号组成的，例如 1527846880572-5，表示在时间戳 1527846880572 时生成的第 5 条消息。每个 Stream 在 Redis 中被存储为一个键值对，其中键是 Stream 的名称，值是消息链表。

持久化

Stream 支持持久化，这意味着即使 Redis 服务器重启，Stream 中的消息仍然会保留。这通过 Redis 的持久化机制 RDB（快照）和 AOF（日志）实现，确保数据的高可用性和可靠性。

多播支持

Stream 支持多个消费组（Consumer Group），每个消费组可以有多个消费者（Consumer）。消费组通过 XGROUP CREATE 命令创建，每个消费组都有一个独立的游标 last_delivered_id，记录已消费的最后一条消息。多个消费者之间是竞争关系，即同一个消费组内的多个消费者会争夺消息的处理权。

2. Stream 的基本操作命令

生产端命令

XADD：XADD 命令用于向 Stream 追加消息。可以指定消息的 ID，或使用 * 让 Redis 自动生成 ID。消息内容是键值对形式。
XDEL：XDEL 命令用于删除 Stream 中的消息。它并不会立即删除消息，而是标记为删除状态。
XRANGE：XRANGE 命令用于获取 Stream 中指定范围的消息。可以通过起始 ID 和结束 ID 指定范围，- 表示最小值，+ 表示最大值。
XLEN：XLEN 命令用于获取 Stream 的长度，即消息数量。

消费端命令

XREAD：XREAD 命令用于读取消息，可以阻塞等待新消息，适用于单消费者模式。
XGROUP：XGROUP 命令用于管理消费组。可以创建、删除消费组，或删除消费组中的消费者。
XREADGROUP：XREADGROUP 命令用于读取消费组的消息。可以指定消费组和消费者，并阻塞等待新消息。
XACK：XACK 命令用于确认消息已被处理，防止消息重复处理。

3. Redis 队列几种实现总结

基于 List 的 LPUSH+BRPOP 实现

这种实现方式简单高效，通过 LPUSH 添加消息，BRPOP 阻塞读取消息。然而，需要处理空闲连接问题，因为长时间阻塞的连接会被服务器断开。此外，ACK 机制较为麻烦，无法保证消息处理成功。

基于 Sorted-Set 的实现

这种实现方式常用于延迟队列，通过 ZADD 添加带有分数的消息，ZRANGEBYSCORE 获取消息。然而，无法阻塞获取消息，需要轮询操作，效率较低。

PUB/SUB，订阅/发布模式

这种实现方式适合即时通讯，通过 PUBLISH 发布消息，SUBSCRIBE 订阅消息。优点是广播模式，一个消息可以同时发送给多个消费者。缺点是消息不可持久化，消费者离线期间的消息会丢失。

基于 Stream 类型的实现

这种实现方式类似消息中间件，适用于中小项目。通过 XADD 添加消息，XREAD 或 XREADGROUP 读取消息。Stream 支持消费组和持久化，具备较高的可靠性。然而，需要管理和监控，确保消息处理的顺利进行。

4. 消息队列问题

Stream 消息太多

当 Stream 中的消息过多时，可以通过 XADD 命令的 MAXLEN 参数设置 Stream 的最大长度，自动删除旧消息，防止内存占用过大。

消息未 ACK

如果消费者未及时确认消息，会导致 pending_ids 列表增长，占用内存。应尽快使用 XACK 命令确认消息，减少 pending_ids 列表的长度。

PEL 防丢失

如果客户端在读取消息后断开连接，未确认的消息会保留在 pending_ids 列表中。客户端重新连接后，可以继续读取这些消息，防止消息丢失。

死信问题

当某条消息长期未被处理时，可以通过 XPENDING 命令查看消息的处理状态，设定消息处理的临界值。达到临界值后，可以将该消息标记为死信（Dead Letter），通过 XDEL 命令删除。

Stream 高可用

Stream 的高可用性基于 Redis 的主从复制机制。通过 Sentinel 或 Cluster 集群环境，Stream 可以实现高可用。然而，在故障转移时，可能会丢失少量数据。

分区 Partition

Redis 服务器本身不支持原生分区功能。如果需要分区，可以在客户端实现，使用不同的 Stream 名称，将消息分片存储在不同的 Stream 中，实现水平扩展。

5. 使用场景

日志收集系统

在分布式系统中，日志收集是一个常见需求。可以使用 Redis Stream 实现一个高效的日志收集系统，将各个服务的日志通过 Stream 传递到日志处理服务，实现实时日志收集和分析。

实时监控系统

实时监控系统需要对大量实时数据进行采集和处理。可以使用 Redis Stream 构建一个高性能的实时监控系统。各个监控点将数据发送到 Stream，监控中心从 Stream 中读取数据进行处理和展示，实现对系统的实时监控。

消息队列服务

在分布式微服务架构中，各个服务之间的通信可以通过消息队列实现。Redis Stream 可以作为消息队列服务，实现服务之间的消息传递和处理，确保消息的可靠传递和高效处理。

6. 与其他消息队列的对比

与 Kafka 的对比

设计理念：Kafka 是一个分布式流处理平台，适用于大规模数据流处理和持久化；Stream 则更轻量，适合中小规模的消息队列和数据流处理。
持久化：Kafka 的消息持久化设计更为复杂和可靠，适用于高可用性要求高的场景；Stream 的持久化通过 RDB 和 AOF 实现，相对简单。
分区：Kafka 支持分区，可以水平扩展；Stream 不支持分区，需要通过客户端实现分片。

与 RabbitMQ 的对比

协议支持：RabbitMQ 支持多种协议（如 AMQP、MQTT），适用范围更广；Stream 仅支持 Redis 协议。
持久化：RabbitMQ 的持久化机制更为灵活和可靠；Stream 的持久化通过 Redis 实现，性能和可靠性略逊。
功能特性：RabbitMQ 提供丰富的消息路由、优先级队列等特性；Stream 则更为简单和高效，适合基本的消息队列需求。

7. 实践案例

日志收集系统

在一个分布式系统中，日志收集是一个常见需求。可以使用 Redis Stream 实现一个高效的日志收集系统，将各个服务的日志通过 Stream 传递到日志处理服务，实现实时日志收集和分析。

实时监控系统

实时监控系统需要对大量实时数据进行采集和处理，可以使用 Redis Stream 构建一个高性能的实时监控系统。各个监控点将数据发送到 Stream，监控中心从 Stream 中读取数据进行处理和展示，实现对系统的实时监控。

消息队列服务

在一个分布式微服务架构中，各个服务之间的通信可以通过消息队列实现。Redis Stream 可以作为消息队列服务，实现服务之间的消息传递和处理，确保消息的可靠传递和高效处理。

8. 结论

Redis Stream 是 Redis 5.0 引入的一种强大的数据结构，具备持久化、多播、消费组等特性，适用于日志收集、实时监控、消息队列等场景。Stream 提供了灵活的消息处理能力，可以满足复杂的实时数据处理需求。通过对 Stream 的详细介绍和

操作演示，可以看到 Stream 在复杂数据流处理和消息传递中的强大功能。相比其他消息队列，Stream 更为轻量和高效，适合中小规模的应用场景。希望本文能够帮助读者深入理解和应用 Redis Stream，为实际项目提供有力支持。

引言