Redis Stream 教程：快速入门与实战应用

“`markdown

在现代分布式系统中，实时数据处理和事件驱动架构变得越来越重要。Redis Stream 作为 Redis 5.0 引入的强大数据结构，为这些需求提供了高效且灵活的解决方案。它不仅是一个简单的消息队列，更是一个支持消费者组、持久化和历史查询的强大工具。本文将详细介绍 Redis Stream 的核心概念、基本操作以及在实际应用中的场景。

什么是 Redis Stream？

Redis Stream 是一种专门为日志记录、事件存储和消息队列设计的数据结构。它可以被看作是一个仅追加（append-only）的日志，其中每个条目都带有一个唯一的 ID，并由一组字段-值对组成。与传统的 Redis Pub/Sub 机制不同，Stream 允许消费者即使在离线后也能重新消费历史消息，并支持多个消费者协同工作，共同处理消息。

核心概念

理解以下核心概念是掌握 Redis Stream 的关键：

仅追加日志 (Append-Only Log)：Stream 的基本特性是新条目总是被添加到末尾，数据一旦写入就不可修改。这保证了事件的顺序性和不可变性。
唯一 ID (Unique IDs)：Stream 中的每个条目都有一个唯一的 ID。默认情况下，这个 ID 由时间戳（毫秒）和序列号组成，确保了即使在同一毫秒内添加的多个条目也能有唯一的标识。例如：1678886400000-0。
字段-值对 (Field-Value Pairs)：每个 Stream 条目实际上是一个小型的 Hash 结构，包含一个或多个字段-值对。这使得单个事件可以携带丰富的结构化数据。
消费者组 (Consumer Groups)：这是 Redis Stream 最强大的特性之一，灵感来源于 Apache Kafka。消费者组允许一组消费者共同消费同一个 Stream。Stream 会记住每个消费者组的消费进度，并确保组内的每个消息只被一个消费者处理一次。这使得构建可伸缩、高可用的消息处理系统变得非常容易。
与 Pub/Sub 的区别 (Difference from Pub/Sub)：
- 持久性：Stream 消息是持久化的，即使消费者离线，重新上线后也能从上次中断的地方继续消费。Pub/Sub 消息则不会持久化，如果消费者离线，就会错过消息。
- 消费者组：Stream 支持消费者组，允许多个消费者协同处理消息，实现负载均衡和故障恢复。Pub/Sub 是广播模式，所有订阅者都会收到所有消息。
- 历史查询：Stream 允许查询历史消息，而 Pub/Sub 只能处理实时消息。

快速入门

让我们通过一些基本的 Redis 命令来快速体验 Redis Stream。

设置 Redis

首先，你需要一个运行中的 Redis 实例。你可以通过 Docker、Homebrew 或直接下载 Redis 官网的二进制文件来启动它。

基本操作

我们将使用 redis-cli 进行操作。

1. 添加条目 (XADD)

XADD 命令用于向 Stream 添加新条目。如果指定的 Stream 不存在，它会自动创建。使用 * 作为 ID，Redis 会自动生成一个唯一的 ID。

bash XADD mystream * sensor_id 1234 temperature 22.5

这会在 mystream 中添加一个新条目，包含 sensor_id 和 temperature 两个字段。返回的将是新条目的唯一 ID。

2. 读取条目 (XREAD)

XREAD 命令用于从一个或多个 Stream 中读取条目。它可以非阻塞读取，也可以阻塞等待新数据。

从头开始读取指定数量的条目：

bash XREAD COUNT 2 STREAMS mystream 0

0 表示从 Stream 的第一个条目开始读取。
阻塞读取新条目：

bash XREAD COUNT 1 BLOCK 0 STREAMS mystream $

BLOCK 0 表示永远阻塞直到有新数据。$ 表示只读取 Stream 中最新添加的条目（即在 XREAD 命令执行之后添加的）。

3. 读取范围内的条目 (XRANGE)

XRANGE 命令用于获取 Stream 中某个 ID 范围内的所有条目。

bash XRANGE mystream - + COUNT 10

- 和 + 分别代表 Stream 的最小和最大 ID，此命令将返回 mystream 中最早的 10 个条目。

4. 创建消费者组 (XGROUP CREATE)

在进行更复杂的消费模式之前，我们需要创建一个消费者组。

bash XGROUP CREATE mystream mygroup 0-0

这会为 mystream 创建一个名为 mygroup 的消费者组，0-0 表示该组将从 Stream 的开头开始消费。

5. 从消费者组读取消息 (XREADGROUP)

XREADGROUP 命令允许消费者从其所属的消费者组中读取消息。Stream 会确保组内的每个消息只被一个消费者实例获取。

bash XREADGROUP GROUP mygroup consumer1 COUNT 1 STREAMS mystream >

这里，consumer1 是 mygroup 中的一个消费者实例。> 表示只读取该消费者从未消费过的新消息。

6. 确认消息 (XACK)

当一个消费者成功处理完消息后，需要使用 XACK 命令向 Stream 确认。这会告诉 Stream 该消息已被处理，避免重复投递。

bash XACK mystream mygroup 1678886400000-0

1678886400000-0 是之前消费到的消息 ID。

实战应用

Redis Stream 在以下场景中表现出色：

事件溯源 (Event Sourcing)：将所有业务操作记录为一系列不可变的事件，存储在 Stream 中。这些事件可以重建应用状态，提供完整的审计追踪能力。例如，记录用户的每一次点击、订单状态的每一次变化。
实时数据分析 (Real-time Analytics)：收集传感器数据、用户行为日志、交易数据等，并实时将其流入 Stream 进行处理。可以用于构建实时仪表盘、欺诈检测系统等。
消息队列 (Message Queues)：作为轻量级的消息队列，解耦生产者和消费者。生产者将消息发布到 Stream，消费者从 Stream 读取并处理消息。支持消费者组使得消息处理具有高可用性和可伸缩性。
日志聚合 (Log Aggregation)：从多个应用和服务器收集日志，统一写入到 Redis Stream。后端服务可以消费这些日志进行实时分析、存储到长期存储或触发告警。
聊天应用 (Chat Applications)：每个聊天室或用户都可以有一个 Stream 来存储消息。消费者可以从 Stream 中读取最新的消息，也可以查询历史记录。
监控系统 (Monitoring Systems)：将系统指标、错误日志等实时数据推送到 Stream，监控服务可以订阅这些 Stream，进行实时健康检查和异常检测。
用户活动追踪 (User Activity Tracking)：记录用户在应用中的所有交互行为，例如页面访问、按钮点击、搜索查询等，以便进行用户行为分析和个性化推荐。
通知系统 (Notifications)：为每个用户维护一个单独的通知 Stream，存储待发送或已发送的通知。

总结

Redis Stream 是一个功能丰富、性能卓越的数据结构，为构建现代事件驱动和实时数据处理系统提供了强大的基础。通过其仅追加日志、唯一 ID、字段-值对和消费者组等特性，开发者可以轻松实现事件溯源、实时分析、消息队列等复杂功能。如果您正在寻找一个轻量级、高性能的流处理解决方案，Redis Stream 绝对值得您的探索和应用。
“`