消息轨迹的查询与可视化

RocketMQ 提供了 Dashboard 可视化工具来查询消息轨迹展示内容生产阶段发送时间、IP、耗时、状态存储阶段Broker、CommitLog 位置消费阶段消费组、消费时间、结果查询方式按 MessageId 查询按 Key 查询按 Topic 时间范围查询RocketMQ Dashboard在 Dashboard 中查询轨迹登录 RocketMQ Dashboard进入“消息轨迹”页面输入 MessageId 或 Key点击查询展示完整的生产 → 存储 → 消费链路API 查询示例// 通过 MessageId 查询轨迹QueryResult queryResult admin.queryMessageByMessageId(“topic”, “messageId”);消息查询的三种方式按 MessageId、按 Key、按时间查询方式 原理 适用场景按 MessageId 直接通过 IndexFile 定位 已知消息 ID查询单条消息详情按 Key 通过 IndexFile 哈希索引查询 根据业务 Key如订单 ID查询一批消息按时间 遍历 ConsumeQueue 中的时间戳 查询某段时间内的所有消息适合批量分析消息轨迹的日志解读与问题定位常见问题通过轨迹定位消息未收到否是否是消费失败问题现象通过轨迹查看查看生产轨迹消息是否成功发送到 Broker发送成功Producer 发送失败检查网络/重试查看消费轨迹是否有 Consumer 订阅有消费记录检查消费组是否在线/订阅是否正确查看消费状态是否消费失败查看消费轨迹中的错误信息和耗时定位业务逻辑或下游依赖问题十二、消息过滤Tag 标签过滤的原理与使用Tag 是 Topic 下的二级分类用于在 Broker 侧对消息进行轻量级过滤。发送时打 TagMessage msg1 new Message(“order_topic”, “order_create”, “内容”.getBytes());Message msg2 new Message(“order_topic”, “order_pay”, “内容”.getBytes());消费时按 Tag 过滤// 只消费 order_create 和 order_pay 两种 Tagconsumer.subscribe(“order_topic”, “order_create || order_pay”);Tag 过滤的原理Tag 过滤流程匹配不匹配消息携带 TagBroker 构建 ConsumeQueue 时将 Tag 的哈希值存入索引Consumer 订阅时指定 Tag 表达式Broker 拉取消息时根据哈希码快速匹配✅ 投递给 Consumer❌ 丢弃不网络传输Tag 过滤的极致性能ConsumeQueue 中存储了 Tag 的 哈希码8 字节而非完整 Tag过滤时只需对比哈希码无需读取消息体完全在 Broker 的内存中完成几乎无性能损耗SQL92 表达式过滤的原理与使用当过滤逻辑复杂时如基于消息属性做条件判断需要使用 SQL92 过滤。发送时携带属性Message msg new Message(“order_topic”, “内容”.getBytes());msg.putUserProperty(“amount”, “1500”);msg.putUserProperty(“region”, “SH”);msg.putUserProperty(“vip”, “true”);消费时使用 SQL92 表达式// 只消费上海地区、金额大于 1000、VIP 用户的消息consumer.subscribe(“order_topic”,MessageSelector.bySql(“region ‘SH’ AND amount 1000 AND vip true”));SQL92 过滤的原理SQL92 过滤流程TRUEFALSE消息携带用户属性Broker 收到拉取请求解析 SQL 表达式读取消息的所有用户属性在 Broker 中执行 SQL 计算✅ 投递给 Consumer❌ 丢弃⚠️ 需要读取完整消息属性消耗更多 CPU 和内存不适合超高吞吐场景两种过滤方式的性能对比对比维度 Tag 过滤 SQL92 过滤过滤依据 Tag 字符串 用户自定义属性Broker 开销 极低仅哈希比对 较高解析 SQL 读取属性网络传输 不匹配的消息不传输 不匹配的消息不传输灵活性 低只能精确匹配 Tag 高复杂条件组合支持操作符 ||或 LIKE IN AND OR 等性能损耗 ~1%-2% ~10%-20%适用场景 简单的消息分类 复杂的业务规则筛选