【redis】数据类型之Stream

Redis Stream是Redis 5.0版本引入的一种新的数据类型，它提供了一种持久化的、可查询的、可扩展的消息队列服务。

它结合了Redis高性能的特性与持久化能力，支持：

• 多消费者组模式（Consumer Groups）
• 消息回溯（历史消息查询）
• 消息阻塞等待
• 消息持久化（AOF/RDB）

与传统的PUB/SUB模式相比，Stream提供了消息持久化和可靠性投递机制；与专业消息中间件（如Kafka）相比，它更加轻量且完全兼容Redis生态系统。

Stream的基本概念与内部结构

Redis Stream是一个包含零个或任意多个流元素的有序队列。队列中的每个元素都包含一个唯一的ID和任意多个键值对。这些元素会根据ID的大小在流中有序地进行排列。通过将元素ID与时间进行关联，并强制要求新元素的ID必须大于旧元素的ID，Redis从逻辑上将流变成了一种只执行追加操作（append only）的数据结构。

元素ID

每个流元素都有一个唯一的ID，该ID由两部分组成：毫秒时间戳和序列号(<millisecondsTime)>-<sequenceNumber>），例如1717020000000-5。

• 毫秒时间戳：基于Redis实例的当前时间生成，决定ID的主要顺序。

• 序列号：同一毫秒内的递增序号（64 位无符号整数），保证唯一性。

Redis允许用户自定义元素ID，但要求新元素的ID必须比流中所有已有元素的ID都要大。为了方便执行添加操作，Redis为XADD命令的id参数设定了一个特殊值*。当使用作为id参数的值时，Redis将自动为新添加的元素生成一个可用的新ID。

消息生命周期与状态变化

• 未读状态：未被任何消费者读取的消息。

• PENDING状态：通过XREADGROUP读取但未确认的消息（记录在消费者组的PEL列表中）。

• ACKED状态：通过XACK确认后，消息从PEL移除，但仍保留在Stream中（除非手动删除或自动修剪）。

stream命令的使用

有关stream类型的命令可以通过help @stream命令来查看。有关命令的使用可以通过help 命令来查看，例如help xadd。

XADD

XADD：发送消息。

语法：

XADD key [NOMKSTREAM] [MAXLEN|MINID [=|~] threshold [LIMIT count]] *|id field value [field value ...]

选项说明：

• *：自动生成消息ID（格式：<毫秒时间戳>-<序列号>，如1717020000000-0）

• field value：消息内容（至少1组键值对，支持多个字段）

使用：

127.0.0.1:6379> xadd mystream * k1 v1 k2 v2
"1742455243729-0"

XREAD

XREAD：读取消息。

语法：

XREAD [COUNT count] [BLOCK milliseconds] STREAMS key [key ...] ID [ID ...]

选项说明：

• BLOCK：阻塞等待毫秒数（0=无限等待）
• ID：是对应流的起始消息ID（特殊值$表示只读新消息，0-0从头读取）

使用：

127.0.0.1:6379> xread count 10 streams mystream 0-0
1) 1) "mystream"2) 1) 1) "1742455243729-0"2) 1) "k1"2) "v1"3) "k2"4) "v2"

XRANGE

XRANGE：范围查询消息。

语法：

XRANGE key start end [COUNT count]

选项说明：

• start/end：消息ID范围（特殊值-表示最小值，+表示最大值）

• count：限制返回条目数（默认为全部）

使用：

# 查询所有消息
127.0.0.1:6379> xrange mystream - + count 3
1) 1) "1742455243729-0"2) 1) "k1"2) "v1"3) "k2"4) "v2"
2) 1) "1742457110169-0"2) 1) "k1"2) "v1"
3) 1) "1742457113644-0"2) 1) "k2"2) "v2"# 按指定范围查询
127.0.0.1:6379> xrange mystream 1742457110169-0 1742457113644-0
1) 1) "1742457110169-0"2) 1) "k1"2) "v1"
2) 1) "1742457113644-0"2) 1) "k2"2) "v2"

XLEN

XLEN：用于获取指定Stream中当前存储的消息总数（包含未被删除的条目）。

语法：

XLEN key

使用：

127.0.0.1:6379> xlen mystream
(integer) 7

XDEL

XDEL：根据消息ID删除消息，支持多个ID批量删除。

语法：

XDEL key ID [ID ...]

使用：

127.0.0.1:6379> xdel mystream 1742457110169-0
(integer) 1

XREVRANGE

XREVRANGE：逆序获取消息范围，与XRANGE相反。

语法：

XREVRANGE key end start [COUNT count]

使用：

127.0.0.1:6379> xrevrange mystream + - count 3
1) 1) "1742457125799-0"2) 1) "k6"2) "v6"
2) 1) "1742457122555-0"2) 1) "k5"2) "v5"
3) 1) "1742457119534-0"2) 1) "k4"2) "v4"

XGROUP

XGROUP：用于管理消费者组。

语法：

XGROUP [CREATE key groupname ID|$ [MKSTREAM]] [SETID key groupname ID|$] [DESTROY key groupname] [CREATECONSUMER key groupname consumername] [DELCONSUMER key groupname consumername]

选项说明：

• CREATE key groupname ID|$ [MKSTREAM]：显式创建消费者，

  • id/$：起始ID
  • MKSTREAM：自动创建流

• SETID key groupname ID|$：重置消费组的最后递送ID

• DESTROY key groupname：强制删除整个消费者组

• CREATECONSUMER key groupname consumername：显式创建消费者

• DELCONSUMER key groupname consumername：删除消费者

使用：

127.0.0.1:6379> xgroup create mystream mygroup $
OK

XREADGROUP

XREADGROUP：从消费者组内读取消息并管理消费状态。

语法：

XREADGROUP GROUP group consumer [COUNT count] [BLOCK milliseconds] [NOACK] STREAMS key [key ...] ID [ID ...]

选项说明：

• GROUP：消费者组名（需通过XGROUP CREATE预先创建）。
• consumer-name：消费者标识（自动创建或显式管理）。
• STREAMS：目标Stream键名及起始ID（>表示新消息，0-0表示历史消息）。
• COUNT：单次读取消息数（默认全部）。
• BLOCK：阻塞等待时间（毫秒，0表示无限等待）。

使用：

127.0.0.1:6379> xreadgroup group mygroup consumer1 count 1 streams mystream >
1) 1) "mystream"2) 1) 1) "1742459887439-0"2) 1) "kk"2) "vv"

XACK

XACK：消费者组模式下确认消息处理完成。

语法：

XACK key group ID [ID ...]

使用：

127.0.0.1:6379> xack mystream mygroup 1742459887439-0
(integer) 1

XPENDING

XPENDING：查询消费者组内未确认消息（Pending Entries List, PEL）。

语法：

XPENDING key group [[IDLE min-idle-time] start end count [consumer]]

使用：

127.0.0.1:6379> xpending mystream mygroup - + 10
1) 1) "1742460095984-0"2) "consumer1"3) (integer) 46824) (integer) 1

XCLAIM

XCLAIM：用于转移消费者组内未确认消息（Pending Entries List, PEL）所有权的核心命令，主要用于故障恢复、负载均衡和消息重试场景。

语法：

XCLAIM key group consumer min-idle-time id [id ...] [IDLE ms] [TIME unix-time-milliseconds] [RETRYCOUNT count] [FORCE] [JUSTID] [LASTID lastid]

选项说明：

• consumer：接收消息的目标消费者名称
• min_idle_time：消息空闲时间阈值（毫秒），仅空闲时间超过此阈值的消息会被转移
• ID：要转移的消息ID（支持批量操作）
• IDLE：设置消息的新空闲时间（默认重置为0）
• RETRYCOUNT：设置消息的重试次数（默认自动+1）
• JUSTID：仅返回转移成功的消息ID，不返回消息内容（用于批量操作优化性能）

使用：

127.0.0.1:6379> xclaim mystream mygroup consumer2 30 1742460095984-0
1) 1) "1742460095984-0"2) 1) "kk"2) "vv"127.0.0.1:6379> xpending mystream mygroup - + 10 consumer2
1) 1) "1742460095984-0"2) "consumer2"3) (integer) 210554) (integer) 3

XAUTOCLAIM

XAUTOCLAIM：自动化批量转移未确认消息，专为优化消费者组内消息故障恢复和负载均衡设计。它通过自动扫描并转移满足条件的 PENDING消息，显著简化了手动调用XCLAIM的复杂流程。

语法：

XAUTOCLAIM key group consumer min-idle-time start [COUNT count] [JUSTID]

使用：

127.0.0.1:6379> xautoclaim mystream mygroup consumer1 30 0-0
1) "0-0"
2) 1) 1) "1742460095984-0"2) 1) "kk"2) "vv"

XSETID

XSETID：手动设置Stream的最后一个消息ID。

语法：

XSETID key arg

使用：

127.0.0.1:6379> xsetid mystream 1742460095984-0
OK

XINFO

XINFO：用于获取Stream数据结构及关联消费者组的详细信息。

语法：

XINFO [CONSUMERS key groupname] [GROUPS key] [STREAM key] [HELP]

使用：

127.0.0.1:6379> xinfo consumers mystream mygroup
1) 1) "name"2) "consumer1"3) "pending"4) (integer) 15) "idle"6) (integer) 1228633
2) 1) "name"2) "consumer2"3) "pending"4) (integer) 05) "idle"6) (integer) 2619218

XTRIM

XTRIM：动态修剪Stream中的消息数量或时间范围，以控制内存占用和保持数据时效性。

语法：

XTRIM key MAXLEN|MINID [=|~] threshold [LIMIT count]

选项说明：

• MAXLEN：保留最新threshold条消息（按消息数量修剪）
• MINID：保留ID≥threshold的消息（按时间范围修剪）
• =|~：=表示精确修剪；~表示近似修剪（效率更高但可能略微超出阈值）
• LIMIT count：可选，限制单次修剪的消息数量（避免大范围删除阻塞服务）

使用：

127.0.0.1:6379> xtrim mystream maxlen 5
(integer) 3

使用场景

消息队列：替代Redis的LIST或Pub/Sub，支持持久化和消费者组。示例：订单处理、异步任务队列。

日志收集：将日志作为事件流存储，支持多消费者并行分析。

实时数据处理：结合XREAD阻塞读取，实时处理用户行为数据（如点击、搜索）。

事件溯源：记录系统状态变化，支持按时间回放事件。

Java使用示例

生产者生产消息

package com.morris.redis.demo.stream;import org.springframework.data.redis.connection.stream.ObjectRecord;
import org.springframework.data.redis.connection.stream.StreamRecords;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;import javax.annotation.Resource;
import java.util.HashMap;
import java.util.Map;@RestController
@RequestMapping("stream")
public class StreamProducer {@Resourceprivate RedisTemplate<String, Object> redisTemplate;@RequestMapping("/produce")public String sendMessage() {Map<String, Object> message = new HashMap<>();message.put("name", "morris");message.put("age", 18);ObjectRecord<String, Map<String, Object>> record = StreamRecords.newRecord().ofObject(message).withStreamKey(StreamConfig.STREAM_KEY);redisTemplate.opsForStream().add(record);return "success";}
}

消费者组初始化

package com.morris.redis.demo.stream;import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.RedisSystemException;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;import javax.annotation.PostConstruct;
import javax.annotation.Resource;@Slf4j
@Component
public class StreamConfig {public static final String STREAM_KEY = "java-stream";public static final String GROUP = "java-group";@Resourceprivate RedisTemplate redisTemplate;@PostConstructpublic void initConsumerGroup() {try {redisTemplate.opsForStream().createGroup(STREAM_KEY, GROUP);} catch (RedisSystemException e) {log.info("Consumer group already exists");}}
}

消费者消费消息

package com.morris.redis.demo.stream;import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;import javax.annotation.Resource;
import java.util.List;@RestController
@RequestMapping("stream")
public class StreamConsumer {@Resourceprivate RedisTemplate redisTemplate;@RequestMapping("/consumer")public Object consumeMessages() {Consumer consumerConfig = Consumer.from(StreamConfig.GROUP, "java-consumer");StreamOffset<String> offset = StreamOffset.create(StreamConfig.STREAM_KEY, ReadOffset.lastConsumed());List<MapRecord<String, Object, Object>> records = redisTemplate.opsForStream().read(consumerConfig, StreamReadOptions.empty().count(10), offset);records.forEach(record -> {redisTemplate.opsForStream().acknowledge(StreamConfig.STREAM_KEY, StreamConfig.GROUP, record.getId());});return records;}
}