Redis_7_Streams与高可用集群实战

Redis 7.0 Streams与高可用集群部署实战从消息队列到分布式架构，全面掌握Redis核心能力前言Redis不只是一个缓存数据库。Redis 5.0引入的Streams让它具备了消息队列的能力，Redis 7.0进一步增强了Streams的稳定性和性能。很多团队在用Kafka/RabbitMQ处理消息队列时，其实Redis Streams就能满足大部分场景。本文将深入讲解Redis Streams的原理和使用，以及Redis Cluster集群和Sentinel高可用方案的实战部署。一、Redis Streams核心概念1.1 什么是StreamsRedis Streams是Redis 5.0引入的数据结构，类似于Apache Kafka的消息日志：Stream: orders ├── 1699000000000-0: {product: "iPhone", quantity: 1} ├── 1699000000001-0: {product: "MacBook", quantity: 2} ├── 1699000000002-0: {product: "iPad", quantity: 1} └── ...（只追加，不可修改）核心特性：只追加（Append-only）：消息只能追加到末尾，不能修改持久化：消息存储在内存，可持久化到RDB/AOF消费者组：支持多个消费者协作消费消息确认：支持ACK机制，防止消息丢失1.2 消息ID格式毫秒时间戳-序列号 例如：1699000000000-0 - 毫秒时间戳：消息创建的时间 - 序列号：同一毫秒内的序号，从0开始 - 可以自定义ID（必须递增） - 自动生成：XADD stream * field value（*表示自动生成）二、Stream命令详解2.1 XADD - 添加消息# 自动生成IDXADD orders * product"iPhone"quantity1price999# 返回: "1699000000000-0"# 指定ID（必须大于当前最大ID）XADD orders1699000000001-0 product"MacBook"quantity2# 限制Stream最大长度XADD orders MAXLEN1000* product"iPad"quantity1# 保留最新1000条，自动删除旧消息# 近似裁减（更高效）XADD orders MAXLEN ~1000* product"Watch"quantity3# ~表示允许实际长度略超过10002.2 XREAD - 读取消息# 从指定ID开始读取（阻塞模式）XREAD COUNT10BLOCK5000STREAMS orders0# COUNT 10: 最多读取10条# BLOCK 5000: 阻塞5秒等待新消息# 0: 从头开始读取# 读取新消息（只读取比指定ID大的消息）XREAD COUNT10BLOCK0STREAMS orders1699000000000-0# 1699000000000-0: 只读取ID大于此值的消息# BLOCK 0: 永久阻塞直到有新消息# 同时监听多个StreamXREAD COUNT5BLOCK0STREAMS orders notifications002.3 XREADGROUP - 消费者组读取# 创建消费者组XGROUP CREATE orders mygroup $ MKSTREAM# $: 从最新消息开始消费# MKSTREAM: 如果Stream不存在则创建# 消费者读取消息XREADGROUP GROUP mygroup consumer1 COUNT10BLOCK0STREAMS orders# : 读取未被分配过的新消息# 读取待处理消息（重新分配）XREADGROUP GROUP mygroup consumer1 COUNT10STREAMS orders0# 0: 读取该消费者的待处理消息2.4 XACK - 确认消息# 确认消息已处理XACK orders mygroup1699000000000-0# 查看待处理消息列表（PEL）XPENDING orders mygroup# 查看待处理消息详情XPENDING orders mygroup - +10# 转移待处理消息给其他消费者XCLAIM orders mygroup consumer236000001699000000000-0# 3600000: 消息空闲超过1小时才转移2.5 其他常用命令# 查看Stream长度XLEN orders# 查看Stream信息XINFO STREAM orders# 查看消费者组信息XINFOGROUPSorders# 查看消费者信息XINFO CONSUMERS orders mygroup# 裁剪StreamXTRIM orders MAXLEN1000# 删除消息XDEL orders1699000000000-0# 反向范围查询XREVRANGE orders + - COUNT10三、消费者组深度解析3.1 消费者组的工作原理Stream: orders ├── msg-1 → 分配给 consumer1 (待处理) ├── msg-2 → 分配给 consumer2 (待处理) ├── msg-3 → 分配给 consumer1 (已确认) ├── msg-4 → 未分配 (新消息) └── msg-5 → 未分配 (新消息) 消费者组 mygroup: ├── consumer1: [msg-1(待处理), msg-3(已确认)] └── consumer2: [msg-2(待处理)] 待处理消息列表 (PEL): ├── msg-1: consumer1, 空闲30秒 └── msg-2: consumer2, 空闲5秒3.2 消费者组 vs 发布订阅特性消费者组发布订阅 (Pub/Sub)消息持久化✅ 存储在Stream中❌ 不存储消息确认✅ 支持ACK❌ 不支持消息回放✅ 可以重新消费❌ 消费后消失负载均衡✅ 自动分配给消费者❌ 每个订阅者都收到消息丢失不会（有PEL）会（离线时收不到）适用场景任务队列、事件处理实时通知、广播3.3 消息确认与重试机制importredisimporttime r=redis.Redis(host='localhost',port=6379,decode_responses=True)# 创建消费者组try:r.xgroup_create('orders','mygroup',id='0',mkstream=True)exceptredis.exceptions.ResponseError:pass# 组已存在defprocess_message(message):"""处理消息"""print(f"Processing:{message}")# 模拟处理time.sleep(1)returnTruedefconsume_messages():"""消费消息的主循环"""whileTrue:# 读取新消息messages=r.xreadgroup(groupname='mygroup',consumername='consumer1',count=1,block=5000,streams={'orders':''})ifnotmessages:continueforstream,msgsinmessages:formsg_id,datainmsgs:try:ifprocess_message(data):# 处理成功，确认消息r.xack('orders','mygroup',msg_id)print(f"Acknowledged:{msg_id}")exceptExceptionase:print(f"Failed to process{msg_id}:{e}")# 不确认消息，等待重试defretry_pending_messages():"""重试超时的待处理消息"""whileTrue:# 获取超时的待处理消息pending=r.xpending_range('orders','mygroup',min='-',max='+',count=10)foriteminpending:ifitem['time_since_delivered']60000:# 超过60秒# 转移给当前消费者r.xclaim('orders','mygroup','consumer