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

Redis Stream 命令

news 2025/9/14 13:20:11

一、Stream 基础命令

1. XADD - 添加消息到 Stream

bash

XADD key [MAXLEN|MINID [=|~] threshold] ID field value [field value ...]

参数说明：

key：Stream 的名称（相当于队列名称）
MAXLEN|MINID：可选，消息长度限制策略
- MAXLEN：保留最新的 N 条消息
- MINID：保留 ID 大于指定值的消息
- =|~：精确匹配或近似匹配（~ 表示可能多保留一些以提高性能）
- threshold：数量阈值
ID：消息唯一标识，格式为 "时间戳 - 序号"
- *：自动生成 ID（推荐，确保唯一性和有序性）
- 手动指定：如 "1620000000000-0"（不推荐，可能破坏顺序）
field value：键值对形式的消息内容，可多个

示例：

bash

# 自动生成ID添加消息
XADD user_events * name "张三" action "login"# 限制最多保留1000条消息
XADD user_events MAXLEN ~ 1000 * name "李四" action "logout"

2. XRANGE - 读取消息范围

bash

XRANGE key start end [COUNT count]

参数说明：

key：Stream 名称
start：起始 ID，-表示最小 ID（最早的消息）
end：结束 ID，+表示最大 ID（最新的消息）
COUNT：可选，限制返回消息数量
count：具体数量值

示例：

bash

# 读取所有消息
XRANGE user_events - +# 读取ID在1620000000000-0之后的10条消息
XRANGE user_events 1620000000000-0 + COUNT 10

3. XREAD - 读取消息（非消费者组模式）

bash

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

参数说明：

COUNT：可选，限制读取消息数量
count：具体数量值
BLOCK：可选，阻塞等待时间（毫秒），0 表示无限等待
milliseconds：阻塞毫秒数
STREAMS：固定关键字
key：一个或多个 Stream 名称
id：每个 Stream 对应的起始 ID，$表示只读取新消息

示例：

bash

# 非阻塞读取，从最新位置开始
XREAD STREAMS user_events $# 阻塞10秒读取，最多5条消息
XREAD COUNT 5 BLOCK 10000 STREAMS user_events $

二、消费者组命令

1. XGROUP CREATE - 创建消费者组

bash

XGROUP CREATE key groupname id [MKSTREAM]

参数说明：

key：Stream 名称
groupname：消费者组名称
id：消费起始位置
- $：从最新消息开始（只消费新消息）
- 0：从最早消息开始（消费所有历史消息）
- 具体 ID：从指定 ID 之后开始消费
MKSTREAM：可选，如果 Stream 不存在则自动创建

示例：

bash

# 创建消费者组，从最新消息开始消费
XGROUP CREATE user_events login_group $# 从最早消息开始消费，不存在则创建Stream
XGROUP CREATE user_events all_group 0 MKSTREAM

2. XREADGROUP - 从消费者组读取消息

bash

XREADGROUP GROUP groupname consumername [COUNT count] [BLOCK milliseconds] STREAMS key [key ...] id [id ...]

参数说明：

GROUP：固定关键字
groupname：消费者组名称
consumername：消费者名称（属于该组）
COUNT：可选，限制读取数量
BLOCK：可选，阻塞毫秒数
STREAMS：固定关键字
key：Stream 名称
id：消费起始位置
- >：只读取未分配给任何消费者的新消息（推荐）
- 具体 ID：读取指定 ID 的消息（用于重试）

示例：

bash

# 消费者consumer1从login_group读取新消息，最多5条，阻塞10秒
XREADGROUP GROUP login_group consumer1 COUNT 5 BLOCK 10000 STREAMS user_events >

3. XACK - 确认消息处理完成

bash

XACK key groupname id [id ...]

参数说明：

key：Stream 名称
groupname：消费者组名称
id：一个或多个消息 ID（需要确认的消息）

示例：

bash

# 确认ID为1620000000000-0的消息已处理
XACK user_events login_group 1620000000000-0

4. XPENDING - 查看未确认消息

bash

XPENDING key groupname [start end count] [consumer]

参数说明：

key：Stream 名称
groupname：消费者组名称
start：起始 ID，-表示最小
end：结束 ID，+表示最大
count：返回消息数量
consumer：可选，指定消费者名称（只看该消费者的未确认消息）

示例：

bash

# 查看login_group所有未确认消息
XPENDING user_events login_group# 查看consumer1的10条未确认消息
XPENDING user_events login_group - + 10 consumer1

5. XCLAIM - 抢占未确认消息

bash

XCLAIM key groupname consumername min-idle-time id [id ...] [IDLE ms] [TIME ms-unix-time] [RETRYCOUNT count] [FORCE]

参数说明：

key：Stream 名称
groupname：消费者组名称
consumername：接收消息的消费者名称
min-idle-time：消息至少需要空闲的毫秒数（未被确认的时间）
id：一个或多个要抢占的消息 ID
IDLE：可选，设置抢占后消息的初始空闲时间
TIME：可选，设置消息的时间戳
RETRYCOUNT：可选，设置重试次数
FORCE：可选，即使消息当前被活跃消费者处理也强制抢占

示例：

bash

# 将空闲超过30秒的消息抢占给consumer2
XCLAIM user_events login_group consumer2 30000 1620000000000-0

6. XDEL - 删除消息

bash

XDEL key id [id ...]

参数说明：

key：Stream 名称
id：一个或多个要删除的消息 ID

示例：

bash

# 删除指定消息
XDEL user_events 1620000000000-0

7. XGROUP 其他命令

bash

# 删除消费者
XGROUP DELCONSUMER key groupname consumername# 销毁消费者组
XGROUP DESTROY key groupname# 修改消费者组的起始消费位置
XGROUP SETID key groupname id

三、命令与参数对应关系表

命令	核心参数	作用
XADD	key, ID, field-value	添加消息到 Stream
XRANGE	key, start, end	读取指定范围消息
XREAD	key, id, BLOCK	非组模式读取消息
XGROUP CREATE	key, groupname, id	创建消费者组
XREADGROUP	groupname, consumername, >	组内读取消息
XACK	groupname, id	确认消息处理完成
XPENDING	groupname, consumer	查看未确认消息
XCLAIM	groupname, consumername, min-idle-time	抢占超时消息