核心内容摘要
申鹤脚法：技惊四座的优雅与力量
简简单单 Online zuozuo 本心、输入输出、结果文章目录TBMQ 如何用 Redis 实现持久化消息存储前言
为什么选择 Redis
迁移到 Redis
Redis 集群约束与 Lua 脚本原子操作
选择合适的 Redis 数据结构
动态管理有序集合大小
消息检索与清理
从 Jedis 迁移到 Lettuce
8、
总结TBMQ 如何用 Redis 实现持久化消息存储编辑 | 简简单单 Online zuozuo地址 | https://blog.csdn.net/qq_15071263如果觉得本文对你有帮助欢迎关注、点赞、收藏、评论谢谢前言TBMQThingsBoard 开源 MQTT broker主要设计用于从物联网设备聚合数据并可靠地投递到后端应用。
应用会订阅来自数万甚至数十万设备的数据并消费。
为应对不同场景TBMQ 将 MQTT 客户端分为应用客户端与标准物联网设备应用客户端始终持久化依赖 Kafka 做会话持久化与消息投递设备端DEVICE则使用另一套持久化方案。
本文从技术角度介绍 TBMQ 如何用 Redis 管理 DEVICE 客户端的持久化 MQTT 会话旨在为正在设计可扩展、会话感知系统的软件工程师提供可落地的参考。
#TBMQ #Redis #MQTT #持久化会话 #物联网 #消息存储 #Lua脚本 #ThingsBoard
为什么选择 Redis在 TBMQ
x 中DEVICE 客户端的消息持久化与拉取依赖 PostgreSQL以保证客户端重连时仍能收到未投递的消息。
PostgreSQL 在初期表现尚可但随着持久化 MQTT 会话数量增长我们预判其架构会成为瓶颈若想深入了解当时基于 PostgreSQL 的用法与架构取舍可参考 TBMQ 官方博客中的相关文章。
为应对这一问题我们评估了能随负载更好扩展的替代方案并很快选定 Redis其水平扩展能力、原生集群支持与高性能特性更适合高吞吐、会话型工作负载。
迁移到 Redis确定方向后我们启动了迁移先评估能在保持 PostgreSQL 方案功能的前提下、又符合 Redis Cluster 约束的数据结构。
迁移过程既要保留原有语义消息顺序、重连后可检索未投递消息等又要利用 Redis 集群的分片与高并发能力因此数据模型与键的设计需要重新梳理。
Redis 集群约束与 Lua 脚本原子操作迁移时我们意识到要高效实现消息持久化与顺序需要多种 Redis 数据结构配合而这会带来多键操作与 Redis Cluster 的槽位约束问题。
Redis Cluster 要求相关键落在同一槽位通常通过 hash tag将键中{xxx}部分用于计算槽位保证同一客户端的数据在同一节点从而避免跨槽错误。
但在高吞吐场景下同一 MQTT 客户端可能并发收到大量消息仅靠 hash tag 不够还必须保证多步操作的原子性。
Redis 单条命令是原子的而我们则需要对每个 MQTT 客户端的多种结构在一次逻辑操作中一起更新若顺序执行多条命令会出现竞态与不一致。
因此我们决定凡是涉及保存消息或重连时拉取未投递消息等操作都通过独立的 Lua 脚本执行从而在一次脚本执行中完成所有相关读写保证一致性。
选择合适的 Redis 数据结构持久化会话的一个核心需求是在客户端多次断线重连后仍能按顺序投递消息。
在对比多种 Redis 结构后我们选用有序集合sorted set来维护消息引用顺序有序集合按 score 天然有序便于按升序或降序快速取出一批消息。
消息体本身则用字符串存储通过 SET 命令并设置过期时间EX实现 O(
写入与 TTL 管理取回或删除消息体也是 O(
且不影响有序集合结构。
键的命名采用 hash tag 形式例如{client_id}_messages其中{client_id}为持久化 MQTT 客户端的唯一 ID 作为 hash tag_messages为固定后缀。
MQTT 报文 ID 范围为 0–65535会回绕我们使用单调递增的 score 表示“逻辑顺序”这样即使 packet ID 从 65535 回到 1score 仍继续递增例如下一跳为 65536从而在有序集合中始终保持正确顺序。
此外我们用一条字符串键保存“已处理的最大 MQTT packet ID”与 PostgreSQL 方案中的用途一致客户端重连时服务端据此确定下一个要保存到 Redis 的消息应使用的 packet ID。
消息载荷的写入与 TTL 示例概念上如下SET key EX ttl_seconds payload按需获取或删除载荷GET key DEL key存储“最后处理的 packet ID”的字符串键可类比为SET {client_id}_last_packet_id value
动态管理有序集合大小在采用有序集合 字符串、并为每条消息设置 TTL 后我们不再依赖按时间周期做批量清理同时参考 PostgreSQL 时代的策略对每个持久化客户端限制最大保留消息数以控制和预测单客户端内存占用。
为贴合 MQTT 协议本身的限制单客户端持久化消息数上限设为 65535。
在 Redis 侧我们实现了有序集合大小的动态管理每次新消息写入后对有序集合做裁剪使消息总数不超过该上限从而在保证顺序与 TTL 的前提下控制内存与行为。
消息检索与清理设计不仅在新消息持久化时做动态大小管理还在“消息检索”时做清理——即设备重连并拉取未投递消息时在 Lua 脚本中一并完成已投递或过期消息的删除与有序集合的维护。
这样通过 Redis 有序集合与字符串配合再加上 Lua 脚本的原子性我们实现了高效的消息持久化、按序检索以及写入与检索时的动态清理避免无序膨胀。
从 Jedis 迁移到 Lettuce为验证基于 Redis 的持久化消息架构的可扩展性我们选择了点对点P2PMQTT 通信模式做性能测试。
在大规模测试前先用 PostgreSQL 做持久化的原型测试其吞吐约在 30k msg/s 达到上限迁到 Redis 后我们发现 Jedis 成为新的瓶颈。
Jedis 稳定但为同步客户端每条 Redis 命令顺序执行在高并发下会限制 Netty 等异步 I/O 的利用率。
RedisInsight 显示单节点约 66k 命令/秒与 TBMQ 约 40k msg/s 对应因每条消息会触发多次 Redis 操作。
为解决这一问题我们改用基于 Netty 的异步 Redis 客户端 Lettuce。
迁移后吞吐提升到约 60k msg/s单节点命令数约 100k/s与预期一致Lettuce 支持多命令并行发送与处理能更好发挥 Redis 的并发能力最终达到了我们期望的性能提升。
更详细的性能测试可参考 TBMQ 官方发布的性能测试专题文章。
8、
总结在分布式系统中当使用传统数据库等垂直扩展组件来承载高并发、会话型工作负载时往往会出现可扩展性瓶颈。
TBMQ 从 PostgreSQL 迁到 Redis 的实践表明将会话存储卸载到 Redis并在迁移过程中做好数据结构选择、集群约束、原子操作与客户端选型可以在 TBMQ
x 中构建出支持大量并发持久化会话、且无单点故障的持久化层。
希望本文能为在设计可扩展、会话感知分布式系统的工程师提供一些可借鉴的思路。
生如逆旅一苇以航欢迎关注、欢迎联系交流、欢迎沟通想法、欢迎交换意见、欢迎合作咨询感谢亲的关注、点赞、收藏、评论一键三连支持谢谢