Redis Stream基本使用及应用场景

一、概念

Redis Streams是Redis5.0提供的一种消息队列机制，支持多播的可持久化的消息队列，用户实现发布订阅的功能，借鉴了kafka设计。

二、常用命令

命令名称	描述
`XADD key ID field value [field value ...]`	添加一条消息 `key`：Stream 的名称； `ID`：消息的唯一标识符，可以使用 `*` 自动生成； `field value`：消息的内容，以键值对的形式存储。
`XREAD [COUNT count] STREAMS key [ID]`	从一个或多个 Stream 中读取消息。 `COUNT`：限制返回的消息数量（可选）； `STREAMS`：指定要读取的 Stream 键； `ID`：指定从哪个消息 ID 开始读取（`$` 表示只读取新消息，`0` 表示从头开始读取）。
`XREADGROUP GROUP group consumer [COUNT count] STREAMS key [ID]`	从消费者组中读取消息。 `GROUP`：指定消费者组名称； `consumer`：指定消费者名称； `ID`：通常使用 `>` 表示只读取未消费的消息。
`XGROUP CREATE key groupname ID`	创建一个新的消费者组。 `key`：Stream 的名称； `groupname`：消费者组名称； `ID`：指定消费者组从哪个消息 ID 开始消费（`$` 表示从最新消息开始）。
`XGROUP DESTROY key groupname`	删除一个消费者组
`XGROUP SETID key groupname ID`	修改消费者组的起始消费位置
`XACK key groupname ID [ID ...]`	确认消息已被成功处理。 `key`：Stream 的名称。 `groupname`：消费者组名称。 `ID`：需要确认的消息 ID。
`XPENDING key groupname`	查看消费者组中未确认的消息统计信息
`XPENDING key groupname start end count [consumer]`	查看具体的未确认消息列表
`XDEL key ID [ID ...]`	删除指定的消息。 `key`：Stream 的名称。 `ID`：需要删除的消息 ID。
`XTRIM key MAXLEN [=\|~] threshold`	控制 Stream 的长度，删除旧消息。 `MAXLEN`：设置最大长度。 `=`：严格模式，保留指定数量的消息。 `~`：近似模式，可能保留比指定数量稍多的消息。
`XINFO STREAM key`	获取 Stream 的基本信息
`XINFO GROUPS key`	获取 Stream 的所有消费者组信息
`XINFO CONSUMERS key groupname`	获取消费者组中的所有消费者信息
XLEN sms_stream	查看Stream长度

三、应用场景

采用Redis Streams 实现短信发送队列的异步处理，替代传统List列表无数据轮询模式，降低系统资源消耗。

角色定义

生产者：短信网关服务，将待发送短信封装为消息写入Redis Stream。
消费者组：多个短信处理节点（Consumer Group），每个节点负责消费部分消息。
消息格式：使用Protobuf序列化消息体，包含手机号、内容、模板ID等字段。

1.Redis命令实现

生产者通过XADD命令写入Stream

XADD sms_stream * mobile "13800138000" content "您的验证码是1234" template_id "T1001"

消费者通过XREADGROUP阻塞读取消息

XREADGROUP GROUP sms_group consumer1 COUNT 1 BLOCK 5000 STREAMS sms_stream >

消息处理成功后发送XACK确认

XACK sms_stream sms_group "1678905140000-0"

消费者组初始化

XGROUP CREATE sms_stream sms_group $ MKSTREAM

消息重试机制，Pending List处理：定时扫描未ACK的消息（超过30秒未确认），重新投递

local pending = XPENDING sms_stream sms_group - + 10
for each message in pending doXCLAIM sms_stream sms_group consumer2 30000 <message-id>
end

限制Stream最大长度（避免内存溢出）

四、与传统轮询模式对比

维度	传统轮询（List/LPOP）	Redis Streams
资源消耗	高频空轮询消耗CPU/网络	阻塞读取，无消息时挂起，降低30% CPU
消息可靠性	无ACK机制，消息可能丢失	支持ACK+Pending List，确保至少一次投递
吞吐量	单消费者约5万条/秒	消费者组多节点并行，可达20万条/秒
功能扩展	仅基础队列	支持多消费者组、消息回溯、阻塞读取