Redis Streams是一个仅追加Append-Only的持久化消息日志。它不是一个普通的 List列表或 Pub/Sub发布订阅每条消息进来都会自动生成一个全局唯一的 ID时间戳序号并且消息会持久化在内存及 RDB/AOF中不会因为消费者离线而丢失。会有一个消费者组来对Streams进行消费。同一条消息只会被组内的某一个消费者拿到实现负载均衡。// // Redis 配置 // struct RedisConfig { std::string host 127.0.0.1; int port 6379; std::string password ; int db 0; // Streams 配置 std::string streamKey device:commands; std::string consumerGroup gateway-group; std::string consumerName gateway-1; // 启动时动态覆盖 // 连接池配置 int poolSize 4; // 连接池大小 int connTimeoutMs 3000; // 连接超时 int cmdTimeoutMs 5000; // 命令超时 };除此之外还有一个极其重要的隐藏概念PELPending Entries List待处理列表。当消费者取走一条消息但还没回复XACK确认时这条消息就挂在 PEL 里。这是 Redis Streams保证消息不丢失的基石。可能问题配置里有cmdTimeoutMs 5000。如果你为了实时性采用XREADGROUP BLOCK 0永久阻塞等待新消息该命令在收到下一条消息前会一直阻塞住这条 TCP 连接。如果客户端的底层cmdTimeoutMs设置为 5 秒那么每隔 5 秒这个阻塞命令就会因超时而断开网关就会频繁重连产生大量XGROUP重入日志。解决方案对于阻塞读命令必须单独设置为无限超时-1或者采用短轮询如BLOCK 1000超时重试。consumerName绝对不能重复PEL 无限膨胀如果网关处理设备命令时异常崩溃没有来得及执行XACK这些消息就会永远卡在 PEL 里。随着时间推移PEL 会越来越大导致XPENDING命令变慢甚至撑爆 Redis 内存。补救措施启动一个后台定时任务扫描XPENDING超时如处理超过 30 秒的消息执行XCLAIM认领并重新处理或者直接XACK并记录失败日志。在分布式中redis的数据库一共16个但是只能使用db0的那一个暂时还不支持