阿里云云市场网站建设,网站建设详情页,医药公司网站建设,上海松江建设工程开发有限公司网站一、概念
Redis Streams是Redis5.0提供的一种消息队列机制#xff0c;支持多播的可持久化的消息队列#xff0c;用户实现发布订阅的功能#xff0c;借鉴了kafka设计。 二、常用命令
命令名称描述XADD key ID field value [field value ...]添加一条消息 key#xff1a;St…一、概念
Redis Streams是Redis5.0提供的一种消息队列机制支持多播的可持久化的消息队列用户实现发布订阅的功能借鉴了kafka设计。 二、常用命令
命令名称描述XADD key ID field value [field value ...]添加一条消息 keyStream 的名称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 创建一个新的消费者组。 keyStream 的名称groupname消费者组名称ID指定消费者组从哪个消息 ID 开始消费$ 表示从最新消息开始。 XGROUP DESTROY key groupname删除一个消费者组XGROUP SETID key groupname ID修改消费者组的起始消费位置XACK key groupname ID [ID ...] 确认消息已被成功处理。 keyStream 的名称。groupname消费者组名称。ID需要确认的消息 ID。 XPENDING key groupname查看消费者组中未确认的消息统计信息XPENDING key groupname start end count [consumer]查看具体的未确认消息列表XDEL key ID [ID ...] 删除指定的消息。 keyStream 的名称。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/LPOPRedis Streams资源消耗高频空轮询消耗CPU/网络阻塞读取无消息时挂起降低30% CPU消息可靠性无ACK机制消息可能丢失支持ACKPending List确保至少一次投递吞吐量单消费者约5万条/秒消费者组多节点并行可达20万条/秒功能扩展仅基础队列支持多消费者组、消息回溯、阻塞读取
