FMQ 第三方上行源
理解第三方上行源，完成 MQTT 来源配置，并按四级结果判断真实消息是否到达外部系统。
阅读方法
先确认入口、权限和目标对象，再按编号完成操作。每一步都包含预期结果；详细字段和选项收在可展开的参考表中。
截图说明
截图来自原始操作手册。公开版会隐藏真实节点地址、连接地址和凭据；示例中的状态、时间和数据不代表当前生产环境。
开始前准备
| 项目 | 说明 |
|---|---|
| 能做什么 | 把第三方 HTTP、MQTT 或 Kafka 数据接入 FMQ，并交给指定上行规则。 |
| 从哪里进入 | 集成 → 第三方上行源 |
| 需要权限 | 上行源和规则维护权限；获准的来源 Broker 与外部验证资源。 |
| 预计时间 | 35 分钟 |
功能与边界
第三方上行源用于把外部系统已有的数据接入 FMQ。第三方可以主动调用 HTTP，也可以把数据放在 MQTT 或 Kafka 中由 FMQ 持续读取；FMQ 再把消息整理成统一上行事件，交给指定规则处理。
例如供应商已经把温湿度 JSON 发布到自己的 MQTT Topic。FMQ 订阅该 Topic，按字段路径取出设备标识和 data，再由上行规则通过 RabbitMQ 发送给业务系统。
- 本模块负责接入、认证、字段映射、设备身份判断和规则限定。
- 连接器负责规则处理后的外发连接；它不是第三方上行源。
- 逻辑标识透传不会自动创建设备；需要设备治理时使用严格关联。
页面展示了什么
页面上方四项摘要用于判断当前有没有来源在运行、FMQ 是否收到数据、连接器是否确认发送以及是否出现失败。列表用于选择具体来源。
选中来源后，概览看接入地址和最近结果，配置看字段路径，规则绑定看处理范围，运行节点看版本是否生效；失败记录和版本历史只在排障或恢复时使用。
页面主要展示内容
| 配置项 | 用途与使用建议 | 生效与恢复 |
|---|---|---|
| 运行源数 | 正在运行的上行源数量与总数；怎么看：数量不符时查看具体源的发布版本和运行节点。 | 数据来源：当前节点和集群期望状态汇总；只用于查看和判断，不改变运行状态。 |
| 接入速率 / 累计接入 | FMQ 收到第三方消息的速度和累计数量；怎么看：增加只证明消息到达 FMQ。 | 数据来源：上行源运行指标；只用于查看和判断，不改变运行状态。 |
| 连接器确认速率 / 累计确认 | 规则动作通过连接器得到 Broker 确认的数量；怎么看：它仍不等于目标队列已经收到。 | 数据来源：连接器运行结果；只用于查看和判断，不改变运行状态。 |
| 失败数 | 认证、解析、身份、规则或连接器阶段记录的失败；怎么看：出现增长时进入该上行源的失败记录查看阶段和原因。 | 数据来源：最近 24 小时失败记录；只用于查看和判断，不改变运行状态。 |
| 上行源列表 | 展示来源名称、类型、草稿/发布版本、集群运行、最近消息和确认/失败；怎么看：先选中目标行，再看下方详情。 | 数据来源：上行源目录与运行快照；只用于查看和判断，不改变运行状态。 |
| 集群应用与运行状态 | APPLIED 表示目标版本已应用；RUNNING 表示当前节点正在接收；怎么看：发布后同时核对版本和状态。 | 数据来源：集群节点状态；只用于查看和判断，不改变运行状态。 |
| 详情标签 | 概览看接入地址和最近结果；配置看映射；规则绑定看规则；运行节点看版本；失败记录和版本历史用于恢复；怎么看：按问题选择标签，不必逐页查看。 | 数据来源：当前选中的上行源；只用于查看和判断，不改变运行状态。 |
创建与配置
开始前准备三样资料：第三方的接入方式和连接参数、一条真实脱敏消息样例、负责处理这批数据的已启用规则。本例使用 MQTT、JSON 字段映射和逻辑标识透传。
先按消息样例配置字段路径，再选择身份策略和规则。路径使用点号读取嵌套字段并区分大小写；data 应指向业务参数对象。
1. 打开第三方上行源
操作位置： 集成→第三方上行源
从左侧菜单进入 集成 → 第三方上行源，先查看运行源数、累计接入、连接器确认和失败数。
图：第三方上行源页面先显示运行源、接入、连接器确认和失败四类状态，再提供上行源列表。
2. 选择 MQTT 并填写来源
操作位置： 集成→第三方上行源
点击 新建上行源，名称填写 FastBee_供应商温度上行，接入类型选择 MQTT 订阅；填写 Broker 地址、Topic 和单主租约。
图：选择 MQTT 订阅后，填写 Broker 地址和 Topic；演示使用公共 Broker，生产环境应使用受控 Broker。
字段说明
| 字段 / 控件 | 说明 |
|---|---|
| 上行源名称 | 用于列表、运行状态和失败记录中识别来源；保存后进入当前上行源草稿；发布后才影响运行。必填 · 来源：由实施负责人按来源系统和数据用途命名。选择：名称要能区分来源，不要只写 MQTT。推荐：按对接资料填写 · 示例：FastBee_供应商温度上行 |
| 接入类型 | 决定由谁发起连接，以及 FMQ 要创建什么运行资源；保存后进入当前上行源草稿；发布后才影响运行。必填 · 来源：第三方当前提供数据的方式。选择：能调用接口选 HTTP；已有 MQTT Topic 选 MQTT；已有 Kafka Topic 选 Kafka。推荐：按对接资料填写 · 示例：MQTT |
| Broker 地址 | FMQ 要连接的 MQTT 地址，必须包含协议和端口；保存后进入当前上行源草稿；发布后才影响运行。必填 · 来源：MQTT 管理方提供的 FMQ 可达地址。选择：明文链路用 tcp://，TLS 用 ssl://；不要填写管理界面地址。推荐：按对接资料填写 · 示例：tcp://broker.example.com:1883 |
| 订阅 Topic | FMQ 持续监听的主题；可填写多个；保存后进入当前上行源草稿；发布后才影响运行。必填 · 来源：发送方发布约定。选择：多个 Topic 用英文逗号分隔；通配符范围要经过授权。推荐：按对接资料填写 · 示例：vendor/+/telemetry |
| 集群消费策略 | 决定一个源由单节点订阅，还是多个节点共享消费；保存后进入当前上行源草稿；发布后才影响运行。必填 · 来源：FMQ 集群规模和 Broker 能力。选择：默认用单主租约；只有 Broker 支持共享订阅且需要并行消费时再选共享订阅。推荐：按对接资料填写 · 示例：单主租约 |
3. 按真实 JSON 配置字段路径
操作位置： 集成→第三方上行源
消息模式选择 JSON 字段映射，从供应商样例中确认 clientId、deviceKey、timestamp、messageType 和 data 的实际路径。
图：JSON 字段映射从第三方消息中读取设备标识、时间、消息类型和 data 对象。
字段说明
| 字段 / 控件 | 说明 |
|---|---|
| 消息模式 | 决定直接校验标准 Alink，还是先从第三方 JSON 中提取字段；保存后进入当前上行源草稿；发布后才影响运行。必填 · 来源：第三方真实消息样例。选择：普通业务 JSON 选字段映射；只有消息已经符合 Alink 才选 Alink JSON。推荐：按对接资料填写 · 示例：JSON 字段映射 |
| 客户端标识路径 | 从第三方 JSON 读取 clientId；与设备连接标识一致时使用；保存后进入当前上行源草稿；发布后才影响运行。选填 · 来源：第三方消息字段名和层级。选择：与设备标识路径至少填写一项；路径区分大小写。推荐：按对接资料填写 · 示例：clientId |
| 设备标识路径 | 从第三方 JSON 读取业务设备 key；保存后进入当前上行源草稿；发布后才影响运行。选填 · 来源：第三方消息字段名和层级。选择：业务侧稳定使用 deviceKey 时填写。推荐：按对接资料填写 · 示例：deviceKey |
| 数据路径 | 指向要交给上行规则的业务对象；保存后进入当前上行源草稿；发布后才影响运行。必填 · 来源：第三方真实消息结构。选择：温度、湿度在 data 下时填 data；不要指向整条消息。推荐：按对接资料填写 · 示例：data |
| 时间戳路径 | 读取消息产生时间；留空时使用 FMQ 接收时间；保存后进入当前上行源草稿；发布后才影响运行。选填 · 来源：第三方消息约定。选择：来源时间可靠且为平台可识别时间戳时填写。推荐：按对接资料填写 · 示例：timestamp |
| 消息类型路径 | 区分属性、事件或状态消息；保存后进入当前上行源草稿；发布后才影响运行。选填 · 来源：第三方消息约定。选择：来源混合多类消息时填写；本例使用 PROPERTY。推荐：按对接资料填写 · 示例：messageType |
4. 选择身份策略和规则
操作位置： 集成→第三方上行源
本次试接入选择 逻辑标识透传，搜索并勾选 FastBee_第三方上行转发。
图：演示选择逻辑标识透传，并只绑定负责该来源数据的已启用规则。
字段说明
| 字段 / 控件 | 说明 |
|---|---|
| 设备身份策略 | 决定未匹配平台设备的消息能否继续进入规则；保存后进入当前上行源草稿；发布后才影响运行。必填 · 来源：平台设备治理要求。选择：正式设备数据优先严格关联；试接入或逻辑对象可用透传。两种方式都不会自动创建设备。推荐：按对接资料填写 · 示例：逻辑标识透传 |
5. 核对高级参数并保存草稿
操作位置： 集成→第三方上行源
客户端 ID 填写本来源的唯一标识，QoS 保持 1，勾选 首次发布后立即启动，然后点击 保存草稿。
图：客户端 ID 用于识别 FMQ 的订阅会话；QoS 保持 1，并勾选首次发布后立即启动。
字段说明
| 字段 / 控件 | 说明 |
|---|---|
| 客户端 ID | Broker 用它识别 FMQ 订阅会话；保存后进入当前上行源草稿；发布后才影响运行。选填 · 来源：Broker 命名规范。选择：留空可由系统生成；手工填写时必须避免与其他客户端重复。推荐：按对接资料填写 · 示例：FastBee_upstream_consumer |
| QoS | 决定订阅交付级别；保存后进入当前上行源草稿；发布后才影响运行。选填 · 来源：发送可靠性和性能要求。选择：多数业务选 1；QoS 2 需要 Broker 和发送方共同支持。推荐：1 · 示例：1 |
| 清理会话 | 断开后是否丢弃离线订阅状态；保存后进入当前上行源草稿；发布后才影响运行。选填 · 来源：Broker 会话恢复要求。选择：默认开启；需要离线积压时必须同时评估固定 clientId 和 Broker 策略。推荐：true · 示例：开启 |
来源与接入
| 配置项 | 用途与使用建议 | 生效与恢复 |
|---|---|---|
| 必填 上行源名称 | 用于列表、运行状态和失败记录中识别来源；怎么选：名称要能区分来源，不要只写 MQTT；值从哪里来：由实施负责人按来源系统和数据用途命名；建议：以部署或对接资料为准；示例：FastBee_供应商温度上行。 | 保存后进入当前上行源草稿；发布后才影响运行；影响范围：当前第三方上行源；生效要求：需发布；填错时：取值与真实来源不一致时，消息可能无法接入或无法进入规则；恢复：按来源方资料修正，重新校验并用新消息验证。 |
| 说明 | 简要说明来源系统、数据范围或负责人；怎么选：写清接收什么数据，避免重复名称；值从哪里来：对接方案；建议：以部署或对接资料为准；示例：接收供应商 MQTT 温湿度 JSON。 | 保存后进入当前上行源草稿；发布后才影响运行；影响范围：当前第三方上行源；生效要求：需发布；填错时：取值与真实来源不一致时，消息可能无法接入或无法进入规则；恢复：按来源方资料修正，重新校验并用新消息验证。 |
| 必填 接入类型 | 决定由谁发起连接，以及 FMQ 要创建什么运行资源；怎么选：能调用接口选 HTTP；已有 MQTT Topic 选 MQTT；已有 Kafka Topic 选 Kafka；值从哪里来：第三方当前提供数据的方式；建议：以部署或对接资料为准；示例：MQTT。HTTP 推送：第三方主动调用 FMQ 独立接口；一条请求对应一条上行消息。MQTT 订阅：FMQ 主动连接 Broker 并订阅 Topic；持续接收 Topic 中的消息。Kafka 消费：FMQ 加入消费组读取 Topic；按分区持续消费消息。 | 保存后进入当前上行源草稿；发布后才影响运行；影响范围：当前第三方上行源；生效要求：需发布；填错时：取值与真实来源不一致时，消息可能无法接入或无法进入规则；恢复：按来源方资料修正，重新校验并用新消息验证。 |
消息、身份与规则
| 配置项 | 用途与使用建议 | 生效与恢复 |
|---|---|---|
| 必填 消息模式 | 决定直接校验标准 Alink，还是先从第三方 JSON 中提取字段；怎么选：普通业务 JSON 选字段映射；只有消息已经符合 Alink 才选 Alink JSON；值从哪里来：第三方真实消息样例；建议：以部署或对接资料为准；示例：JSON 字段映射。JSON 字段映射：按字段路径读取设备、时间和 data；生成统一上行事件。Alink JSON：消息已含 id、version、method、sys、params；直接校验，不读取路径配置。 | 保存后进入当前上行源草稿；发布后才影响运行；影响范围：当前第三方上行源；生效要求：需发布；填错时：取值与真实来源不一致时，消息可能无法接入或无法进入规则；恢复：按来源方资料修正，重新校验并用新消息验证。 |
| 客户端标识路径 | 从第三方 JSON 读取 clientId；与设备连接标识一致时使用；怎么选：与设备标识路径至少填写一项；路径区分大小写；值从哪里来：第三方消息字段名和层级；建议：以部署或对接资料为准；示例：clientId。 | 保存后进入当前上行源草稿；发布后才影响运行；影响范围：当前第三方上行源；生效要求：需发布；填错时：取值与真实来源不一致时，消息可能无法接入或无法进入规则；恢复：按来源方资料修正，重新校验并用新消息验证。 |
| 设备标识路径 | 从第三方 JSON 读取业务设备 key；怎么选：业务侧稳定使用 deviceKey 时填写；值从哪里来：第三方消息字段名和层级；建议：以部署或对接资料为准；示例：deviceKey。 | 保存后进入当前上行源草稿；发布后才影响运行；影响范围：当前第三方上行源；生效要求：需发布；填错时：取值与真实来源不一致时，消息可能无法接入或无法进入规则；恢复：按来源方资料修正，重新校验并用新消息验证。 |
| 必填 数据路径 | 指向要交给上行规则的业务对象；怎么选：温度、湿度在 data 下时填 data；不要指向整条消息；值从哪里来：第三方真实消息结构；建议：以部署或对接资料为准；示例：data。 | 保存后进入当前上行源草稿；发布后才影响运行；影响范围：当前第三方上行源；生效要求：需发布；填错时：取值与真实来源不一致时，消息可能无法接入或无法进入规则；恢复：按来源方资料修正，重新校验并用新消息验证。 |
| 时间戳路径 | 读取消息产生时间；留空时使用 FMQ 接收时间；怎么选：来源时间可靠且为平台可识别时间戳时填写；值从哪里来：第三方消息约定；建议：以部署或对接资料为准；示例：timestamp。 | 保存后进入当前上行源草稿；发布后才影响运行；影响范围：当前第三方上行源；生效要求：需发布；填错时：取值与真实来源不一致时，消息可能无法接入或无法进入规则；恢复：按来源方资料修正，重新校验并用新消息验证。 |
| 状态路径 | 提取在线、离线等状态值；怎么选：只接属性数据时可留空；值从哪里来：第三方状态消息结构；建议：以部署或对接资料为准；示例：status。 | 保存后进入当前上行源草稿；发布后才影响运行；影响范围：当前第三方上行源；生效要求：需发布；填错时：取值与真实来源不一致时，消息可能无法接入或无法进入规则；恢复：按来源方资料修正，重新校验并用新消息验证。 |
| 消息类型路径 | 区分属性、事件或状态消息；怎么选：来源混合多类消息时填写；本例使用 PROPERTY；值从哪里来：第三方消息约定；建议：以部署或对接资料为准；示例：messageType。 | 保存后进入当前上行源草稿；发布后才影响运行；影响范围：当前第三方上行源；生效要求：需发布；填错时：取值与真实来源不一致时，消息可能无法接入或无法进入规则；恢复：按来源方资料修正，重新校验并用新消息验证。 |
| 必填 设备身份策略 | 决定未匹配平台设备的消息能否继续进入规则；怎么选：正式设备数据优先严格关联；试接入或逻辑对象可用透传。两种方式都不会自动创建设备；值从哪里来：平台设备治理要求；建议：以部署或对接资料为准；示例：逻辑标识透传。严格关联：只接收能命中本地设备缓存的标识；未命中时记录身份失败。逻辑标识透传：允许未关联标识进入规则；消息带未关联标记，但不创建设备。 | 保存后进入当前上行源草稿；发布后才影响运行；影响范围：当前第三方上行源；生效要求：需发布；填错时：取值与真实来源不一致时，消息可能无法接入或无法进入规则；恢复：按来源方资料修正，重新校验并用新消息验证。 |
| 必填 上行规则 | 限定此来源的消息交给哪些已启用规则；怎么选：只选择真正处理该来源的规则，不要为通过校验勾选无关规则；值从哪里来：规则列表；规则必须已启用且包含有效连接器动作；建议：以部署或对接资料为准；示例：FastBee_第三方上行转发。 | 保存后进入当前上行源草稿；发布后才影响运行；影响范围：当前第三方上行源；生效要求：需发布；填错时：取值与真实来源不一致时，消息可能无法接入或无法进入规则；恢复：按来源方资料修正，重新校验并用新消息验证。 |
验证、发布与真实接入
按“保存草稿 → 配置校验 → MQTT/Kafka 连接测试或 HTTP 映射预览 → 发布 → 核对运行节点 → 真实发送”的顺序操作。每一步证明的范围不同，不能跳级。
真实成功按四级判断：FMQ 接入增加、规则处理、连接器确认增加、外部系统实收。本例以 RabbitMQ 队列收到 32.6℃标准 Alink 为终点。
6. 先做配置校验
操作位置： 集成→第三方上行源
选中刚保存的上行源，点击 校验。
图：草稿已保存为 v1；校验通过只证明字段、规则和发布前约束满足。
7. 再做 MQTT 连接测试
操作位置： 集成→第三方上行源
点击 连接测试，等待平台完成连接和订阅探测。
图：连接测试通过，说明 FMQ 能连接 Broker 并订阅 Topic；还没有证明真实消息和外部实收。
字段说明
| 字段 / 控件 | 说明 |
|---|---|
| Broker 地址 | FMQ 要连接的 MQTT 地址，必须包含协议和端口；保存后进入当前上行源草稿；发布后才影响运行。必填 · 来源：MQTT 管理方提供的 FMQ 可达地址。选择：明文链路用 tcp://，TLS 用 ssl://；不要填写管理界面地址。推荐：按对接资料填写 · 示例：tcp://broker.example.com:1883 |
| 订阅 Topic | FMQ 持续监听的主题；可填写多个；保存后进入当前上行源草稿；发布后才影响运行。必填 · 来源：发送方发布约定。选择：多个 Topic 用英文逗号分隔；通配符范围要经过授权。推荐：按对接资料填写 · 示例：vendor/+/telemetry |
8. 发布并确认运行版本
操作位置： 集成→第三方上行源
点击 发布；随后打开 运行节点，核对发布版本、目标/实际版本、集群应用和运行状态。
图：发布 v2 已应用到两个节点；1 个节点运行、1 个节点待命，目标版本与实际版本一致。
9. 用新版 FMQX 发送真实 MQTT 消息
操作位置： FMQX
在 FMQX 新建 MQTT 会话，连接来源 Broker，向本例 Topic 发送 temperature=32.6 的合法 JSON。
图：新版 FMQX 连接公共 MQTT Broker，并向演示 Topic 发送 32.6℃ 的第三方 JSON。
字段说明
| 字段 / 控件 | 说明 |
|---|---|
| Broker 地址 | FMQ 要连接的 MQTT 地址，必须包含协议和端口；保存后进入当前上行源草稿；发布后才影响运行。必填 · 来源：MQTT 管理方提供的 FMQ 可达地址。选择：明文链路用 tcp://，TLS 用 ssl://；不要填写管理界面地址。推荐：按对接资料填写 · 示例：tcp://broker.example.com:1883 |
| 订阅 Topic | FMQ 持续监听的主题；可填写多个；保存后进入当前上行源草稿；发布后才影响运行。必填 · 来源：发送方发布约定。选择：多个 Topic 用英文逗号分隔；通配符范围要经过授权。推荐：按对接资料填写 · 示例：vendor/+/telemetry |
| 消息模式 | 决定直接校验标准 Alink，还是先从第三方 JSON 中提取字段；保存后进入当前上行源草稿；发布后才影响运行。必填 · 来源：第三方真实消息样例。选择：普通业务 JSON 选字段映射；只有消息已经符合 Alink 才选 Alink JSON。推荐：按对接资料填写 · 示例：JSON 字段映射 |
10. 核对接入、规则和连接器确认
操作位置： 集成→第三方上行源
回到上行源详情，刷新后比较累计接入、确认和失败；必要时查看运行节点与规则监控。
图：接入计数和确认计数已增加：FMQ 已接收消息，并由绑定规则提交到连接器。
11. 确认 RabbitMQ 队列真实收到
操作位置： RabbitMQ 管理界面
在获准的 RabbitMQ 管理界面读取本批次队列，核对标准 Alink 的 id、version、method、sys 和 params.temperature=32.6。
图：RabbitMQ 专用队列收到标准 Alink 消息，业务参数 temperature=32.6。
按钮什么时候使用
| 配置项 | 用途与使用建议 | 生效与恢复 |
|---|---|---|
| 保存草稿 | 保存下一版配置；什么时候用：参数未录完也可先保存；保存后先校验，不把草稿当成已生效。 | 草稿版本增加，当前发布版本不变；失败时：参数未录完也可先保存；保存后先校验，不把草稿当成已生效。 |
| 校验 | 检查字段、来源类型、身份策略和规则绑定；什么时候用：所有类型发布前都执行。 | 记录校验摘要；失败时：所有类型发布前都执行。 |
| 映射预览 | 用样例验证 HTTP JSON 的字段提取；什么时候用：HTTP + JSON 字段映射发布前执行；使用真实脱敏样例。 | 只解析和映射，不进入规则；失败时：HTTP + JSON 字段映射发布前执行；使用真实脱敏样例。 |
| 连接测试 | 检查 MQTT/Kafka 当前连接参数和订阅条件；什么时候用：MQTT 或 Kafka 修改连接字段后执行。 | 保存当前连接配置指纹的测试结果；失败时：MQTT 或 Kafka 修改连接字段后执行。 |
| 发布 | 把已校验草稿生成不可变运行版本并通知集群应用；什么时候用：校验及必要测试通过后使用。 | 发布版本增加；集群开始应用目标版本；失败时：校验及必要测试通过后使用。 |
| 启动 | 让已发布且暂停的上行源开始接收新消息；什么时候用：首次发布未立即启动，或维护结束后使用。 | 期望状态改为 RUNNING；失败时：首次发布未立即启动，或维护结束后使用。 |
| 暂停 | 停止接收新消息；什么时候用：维护、切换来源或归档前使用。 | 期望状态改为 STOPPED；失败时：维护、切换来源或归档前使用。 |
| 重启 | 用当前发布版本重新建立连接或消费资源；什么时候用：配置未变但连接状态异常时使用。 | 不创建新版本；失败时：配置未变但连接状态异常时使用。 |
| 回滚 | 基于选中的历史快照生成新的发布版本；什么时候用：新版本异常且旧版本仍符合当前环境时使用。 | 版本号继续递增；失败时：新版本异常且旧版本仍符合当前环境时使用。 |
| 重放 | 对可重放的失败载荷创建一条新事件；什么时候用：先修复失败原因，且页面显示可重放时使用。 | 原失败记录保留，新事件重新进入链路；失败时：先修复失败原因，且页面显示可重放时使用。 |
| 归档 | 结束已发布上行源的使用；什么时候用：先暂停并确认无新消息，再归档。 | 生命周期变为 ARCHIVED；失败时：先暂停并确认无新消息，再归档。 |
怎样判断消息真正成功
| 配置项 | 用途与使用建议 | 生效与恢复 |
|---|---|---|
| 1. FMQ 接入 | 累计接入或接入速率增加，说明来源消息到达 FMQ；怎么看：先确认收到，再检查失败和规则。 | 数据来源：上行源运行指标；只用于查看和判断，不改变运行状态。 |
| 2. 规则处理 | 消息通过映射和身份判断，并进入已绑定规则；怎么看：规则未命中时检查数据路径、消息类型和规则条件。 | 数据来源：规则运行结果与失败记录；只用于查看和判断，不改变运行状态。 |
| 3. 连接器确认 | 连接器得到目标 Broker 的发送确认；怎么看：确认增加后仍要检查目标队列或消费端。 | 数据来源：连接器确认计数；只用于查看和判断，不改变运行状态。 |
| 4. 外部系统实收 | 目标队列或消费程序真正取到并能解析消息；怎么看：这是本实战的闭环终点。 | 数据来源：RabbitMQ 管理界面或受控消费者；只用于查看和判断，不改变运行状态。 |
启停、版本和失败处理
暂停用于停止接收新消息；重启用于在配置未变时重建当前运行资源；回滚会基于旧快照生成新的发布版本。配置本身有误时，应修正草稿、重测并重新发布。
失败记录先看阶段和原因。只有载荷被安全保存且页面显示可重放时才能重放；重放会创建新事件，不修改原失败记录。已发布来源不再使用时，先暂停再归档。
12. 发送一条非法 JSON
操作位置： FMQX
在 FMQX 切换到文本模式，向同一 Topic 发送一条缺少合法 JSON 值的消息。
图：新版 FMQX 以文本模式发送一条可控的非法 JSON，用于演示解析失败。
字段说明
| 字段 / 控件 | 说明 |
|---|---|
| 消息模式 | 决定直接校验标准 Alink，还是先从第三方 JSON 中提取字段；保存后进入当前上行源草稿；发布后才影响运行。必填 · 来源：第三方真实消息样例。选择：普通业务 JSON 选字段映射；只有消息已经符合 Alink 才选 Alink JSON。推荐：按对接资料填写 · 示例：JSON 字段映射 |
13. 查看解析失败原因
操作位置： 集成→第三方上行源
回到 FMQ，打开 失败记录，查看阶段、原因和说明。
图：失败记录显示解析阶段和原因；非法 JSON 没有进入规则，也没有投递到 RabbitMQ。
14. 修正 JSON 后重新发送
操作位置： FMQX
在 FMQX 改回 JSON 模式，将 temperature 改为 33.1 后向同一 Topic 重新发送。
图：把消息修正为合法 JSON 后，在新版 FMQX 重新发送 33.1℃ 数据。
字段说明
| 字段 / 控件 | 说明 |
|---|---|
| 消息模式 | 决定直接校验标准 Alink，还是先从第三方 JSON 中提取字段；保存后进入当前上行源草稿；发布后才影响运行。必填 · 来源：第三方真实消息样例。选择：普通业务 JSON 选字段映射；只有消息已经符合 Alink 才选 Alink JSON。推荐：按对接资料填写 · 示例：JSON 字段映射 |
| 订阅 Topic | FMQ 持续监听的主题；可填写多个；保存后进入当前上行源草稿；发布后才影响运行。必填 · 来源：发送方发布约定。选择：多个 Topic 用英文逗号分隔；通配符范围要经过授权。推荐：按对接资料填写 · 示例：vendor/+/telemetry |
15. 确认修正后的消息已实收
操作位置： RabbitMQ 管理界面
在 RabbitMQ 目标队列读取新消息，核对标准 Alink 和 params.temperature=33.1。
图：RabbitMQ 专用队列收到标准 Alink 消息，业务参数 temperature=33.1。
16. 查看版本并理解回滚
操作位置： 集成→第三方上行源
打开 版本历史，确认当前版本；只有新版本异常且旧配置仍适用时，才选择旧快照回滚。
图：版本历史显示当前发布版本 v2；回滚会生成新的发布版本，版本号不会倒退。
17. 暂停来源并区分重启
操作位置： 集成→第三方上行源
维护或切换数据源前点击 暂停；配置未变但连接异常时才使用 重启。
图：暂停后不再接收新消息；已有版本和失败记录仍保留，可再次启动。
18. 停用来源后归档
操作位置： 集成→第三方上行源
确认上行源已暂停且无新消息后点击 归档。
图：上行源必须先暂停才能归档；归档后保留历史，不再作为运行来源使用。
三种来源参数与常见问题
接入类型按第三方现有能力选择，不按个人熟悉程度选择。地址、账号、Topic、消费组和认证方式由对应系统管理员提供；容量和重试参数只有在峰值、分区数或压测依据明确时才调整。
HTTP 推送参数
| 配置项 | 用途与使用建议 | 生效与恢复 |
|---|---|---|
| 必填 认证方式 | 决定调用方如何证明身份；怎么选：生产优先 HMAC-SHA256；固定 Token 选 Bearer；无认证仅限明确允许的开发环境；值从哪里来：双方接口安全约定；建议：以部署或对接资料为准；示例：HMAC-SHA256。HMAC-SHA256：校验时间戳、随机数和请求体签名；可防篡改和重放。Bearer Token：Authorization 头携带固定 Token；实现简单，但需严格保管。无认证：不校验调用身份；默认生产配置会阻断发布。 | 保存后进入当前上行源草稿；发布后才影响运行；影响范围：当前第三方上行源；生效要求：需发布；填错时：取值与真实来源不一致时，消息可能无法接入或无法进入规则；恢复：按来源方资料修正，重新校验并用新消息验证。 |
| 必填 签名密钥 / Bearer Token | 与认证方式配套的密钥；怎么选：新建时填写高强度值；编辑时留空或保留脱敏值表示不更换；值从哪里来：由双方通过受控渠道约定；建议：以部署或对接资料为准；示例：不在手册中展示。 | 保存后进入当前上行源草稿；发布后才影响运行；影响范围：当前第三方上行源；生效要求：需发布；填错时：泄露后第三方可伪造请求；不得进入截图或日志；恢复：按来源方资料修正，重新校验并用新消息验证。 |
| 单条载荷上限 | 允许的请求体最大字节数，超过返回 413；怎么选：默认 1 MiB；只在样本确实更大时调整；值从哪里来：真实最大消息尺寸和安全上限；建议：1048576。 | 保存后进入当前上行源草稿；发布后才影响运行；影响范围：当前第三方上行源；生效要求：需发布；填错时：取值与真实来源不一致时，消息可能无法接入或无法进入规则；恢复：按来源方资料修正，重新校验并用新消息验证。 |
| 每秒请求数 | 单个上行源允许的持续请求速率，超出返回 429；怎么选：默认 1000；没有压测依据时保持；值从哪里来：正常峰值与容量测试；建议：1000。 | 保存后进入当前上行源草稿；发布后才影响运行；影响范围：当前第三方上行源；生效要求：需发布；填错时：取值与真实来源不一致时，消息可能无法接入或无法进入规则；恢复：按来源方资料修正，重新校验并用新消息验证。 |
| 突发容量 | 短时间内允许的额外请求令牌；怎么选：应覆盖合理突发但不能无限放大；值从哪里来：突发流量评估；建议：2000。 | 保存后进入当前上行源草稿；发布后才影响运行；影响范围：当前第三方上行源；生效要求：需发布；填错时：取值与真实来源不一致时，消息可能无法接入或无法进入规则；恢复：按来源方资料修正，重新校验并用新消息验证。 |
| 签名时钟偏差 | HMAC 请求时间允许偏离服务器的秒数；怎么选：默认 300 秒；优先修复时钟同步，不盲目放大；值从哪里来：双方服务器时钟同步情况；建议：300；示例：300 秒。 | 保存后进入当前上行源草稿；发布后才影响运行；影响范围：当前第三方上行源；生效要求：需发布；填错时：取值与真实来源不一致时，消息可能无法接入或无法进入规则；恢复：按来源方资料修正，重新校验并用新消息验证。 |
| 接入地址 | 保存草稿后按 accessKey 生成的独立 HTTP 地址；怎么看：只交给获准调用方；发布并启动后才可接收。 | 数据来源：平台详情概览；只用于查看和判断，不改变运行状态。 |
MQTT 订阅参数
| 配置项 | 用途与使用建议 | 生效与恢复 |
|---|---|---|
| 必填 Broker 地址 | FMQ 要连接的 MQTT 地址，必须包含协议和端口；怎么选：明文链路用 tcp://，TLS 用 ssl://；不要填写管理界面地址；值从哪里来：MQTT 管理方提供的 FMQ 可达地址；建议：以部署或对接资料为准；示例：tcp://broker.example.com:1883。 | 保存后进入当前上行源草稿；发布后才影响运行；影响范围：当前第三方上行源；生效要求：需发布；填错时：取值与真实来源不一致时，消息可能无法接入或无法进入规则；恢复：按来源方资料修正，重新校验并用新消息验证。 |
| 必填 订阅 Topic | FMQ 持续监听的主题；可填写多个；怎么选：多个 Topic 用英文逗号分隔；通配符范围要经过授权；值从哪里来：发送方发布约定；建议：以部署或对接资料为准；示例：vendor/+/telemetry。 | 保存后进入当前上行源草稿；发布后才影响运行；影响范围：当前第三方上行源；生效要求：需发布；填错时：取值与真实来源不一致时，消息可能无法接入或无法进入规则；恢复：按来源方资料修正，重新校验并用新消息验证。 |
| 必填 集群消费策略 | 决定一个源由单节点订阅，还是多个节点共享消费；怎么选：默认用单主租约；只有 Broker 支持共享订阅且需要并行消费时再选共享订阅；值从哪里来：FMQ 集群规模和 Broker 能力；建议：以部署或对接资料为准；示例：单主租约。单主租约：一个节点订阅，异常时由租约接管；避免集群重复消费。共享订阅：转换为 $share 分组，由多节点并行消费；依赖 Broker 支持共享订阅。 | 保存后进入当前上行源草稿；发布后才影响运行；影响范围：当前第三方上行源；生效要求：需发布；填错时：取值与真实来源不一致时，消息可能无法接入或无法进入规则；恢复：按来源方资料修正，重新校验并用新消息验证。 |
| 用户名 | MQTT 连接身份；怎么选：允许匿名连接时用户名和密码同时留空；值从哪里来：Broker 管理员通过受控渠道提供；建议：以部署或对接资料为准；示例：按实际填写。 | 保存后进入当前上行源草稿；发布后才影响运行；影响范围：当前第三方上行源；生效要求：需发布；填错时：取值与真实来源不一致时，消息可能无法接入或无法进入规则；恢复：按来源方资料修正，重新校验并用新消息验证。 |
| 密码 | 与用户名配套的 MQTT 凭据；怎么选：编辑时留空或保留脱敏值表示不更换；值从哪里来：Broker 管理员通过受控渠道提供；建议：以部署或对接资料为准；示例：不在手册中展示。 | 保存后进入当前上行源草稿；发布后才影响运行；影响范围：当前第三方上行源；生效要求：需发布；填错时：不得进入截图、日志或手册；恢复：按来源方资料修正，重新校验并用新消息验证。 |
| 客户端 ID | Broker 用它识别 FMQ 订阅会话；怎么选：留空可由系统生成；手工填写时必须避免与其他客户端重复；值从哪里来：Broker 命名规范；建议：以部署或对接资料为准；示例：FastBee_upstream_consumer。 | 保存后进入当前上行源草稿；发布后才影响运行；影响范围：当前第三方上行源；生效要求：需发布；填错时：取值与真实来源不一致时，消息可能无法接入或无法进入规则；恢复：按来源方资料修正，重新校验并用新消息验证。 |
| QoS | 决定订阅交付级别；怎么选：多数业务选 1；QoS 2 需要 Broker 和发送方共同支持；值从哪里来：发送可靠性和性能要求；建议：1。0 · 最多一次：开销最低，允许丢失；适合可丢弃高频数据。1 · 至少一次：可能重复，可靠性与性能平衡；业务侧需按消息标识考虑幂等。2 · 仅一次：握手最多、开销最高；仅在明确需要时使用。 | 保存后进入当前上行源草稿；发布后才影响运行；影响范围：当前第三方上行源；生效要求：需发布；填错时：取值与真实来源不一致时，消息可能无法接入或无法进入规则；恢复：按来源方资料修正，重新校验并用新消息验证。 |
| 清理会话 | 断开后是否丢弃离线订阅状态；怎么选：默认开启；需要离线积压时必须同时评估固定 clientId 和 Broker 策略；值从哪里来：Broker 会话恢复要求；建议：true；示例：开启。 | 保存后进入当前上行源草稿；发布后才影响运行；影响范围：当前第三方上行源；生效要求：需发布；填错时：取值与真实来源不一致时，消息可能无法接入或无法进入规则；恢复：按来源方资料修正，重新校验并用新消息验证。 |
Kafka 消费参数
| 配置项 | 用途与使用建议 | 生效与恢复 |
|---|---|---|
| 必填 Broker 地址 | Kafka 引导地址，可填写多个；怎么选：多个端点用英文逗号分隔；值从哪里来：Kafka 管理员提供的 FMQ 可达地址；建议：以部署或对接资料为准；示例：kafka-1:9092,kafka-2:9092。 | 保存后进入当前上行源草稿；发布后才影响运行；影响范围：当前第三方上行源；生效要求：需发布；填错时：取值与真实来源不一致时，消息可能无法接入或无法进入规则；恢复：按来源方资料修正，重新校验并用新消息验证。 |
| 必填 订阅 Topic | 要消费的一个或多个主题；怎么选：多个 Topic 用英文逗号分隔；值从哪里来：数据生产方约定；建议：以部署或对接资料为准；示例：vendor-telemetry。 | 保存后进入当前上行源草稿；发布后才影响运行；影响范围：当前第三方上行源；生效要求：需发布；填错时：取值与真实来源不一致时，消息可能无法接入或无法进入规则；恢复：按来源方资料修正，重新校验并用新消息验证。 |
| 消费组 | Kafka 用消费组管理分区归属和 offset；怎么选：留空时系统按上行源生成稳定消费组；已有消费进度要复用约定组名；值从哪里来：Kafka 对接约定；建议：以部署或对接资料为准；示例：留空由系统生成。 | 保存后进入当前上行源草稿；发布后才影响运行；影响范围：当前第三方上行源；生效要求：需发布；填错时：取值与真实来源不一致时，消息可能无法接入或无法进入规则；恢复：按来源方资料修正，重新校验并用新消息验证。 |
| 必填 安全协议 | 决定是否使用 SASL 认证及 TLS 加密；怎么选：必须与集群一致；跨网络生产环境优先 SASL_SSL；值从哪里来：Kafka 集群安全配置；建议：以部署或对接资料为准；示例：SASL_SSL。SASL_SSL：认证并加密链路；需要正确证书与认证配置。SASL_PLAINTEXT：认证但不加密链路；仅用于可信网络。 | 保存后进入当前上行源草稿；发布后才影响运行；影响范围：当前第三方上行源；生效要求：需发布；填错时：取值与真实来源不一致时，消息可能无法接入或无法进入规则；恢复：按来源方资料修正，重新校验并用新消息验证。 |
| 必填 SASL 机制 | Kafka 身份认证算法；怎么选：在 PLAIN、SCRAM-SHA-256、SCRAM-SHA-512 中选择与集群一致的一项；值从哪里来：Kafka 管理员提供；建议：以部署或对接资料为准；示例：SCRAM-SHA-256。PLAIN：明文口令机制，通常配 TLS；与 Broker 配置一致。SCRAM-SHA-256：挑战响应认证；与 Broker 配置一致。SCRAM-SHA-512：更长摘要的挑战响应认证；与 Broker 配置一致。 | 保存后进入当前上行源草稿；发布后才影响运行；影响范围：当前第三方上行源；生效要求：需发布；填错时：取值与真实来源不一致时，消息可能无法接入或无法进入规则；恢复：按来源方资料修正，重新校验并用新消息验证。 |
| 必填 用户名 | Kafka SASL 身份；怎么选：与密码同时填写；值从哪里来：Kafka 管理员通过受控渠道提供；建议：以部署或对接资料为准；示例：按实际填写。 | 保存后进入当前上行源草稿；发布后才影响运行；影响范围：当前第三方上行源；生效要求：需发布；填错时：取值与真实来源不一致时，消息可能无法接入或无法进入规则；恢复：按来源方资料修正，重新校验并用新消息验证。 |
| 必填 密码 | 与 Kafka 用户名配套的凭据；怎么选：编辑时留空或保留脱敏值表示不更换；值从哪里来：Kafka 管理员通过受控渠道提供；建议：以部署或对接资料为准；示例：不在手册中展示。 | 保存后进入当前上行源草稿；发布后才影响运行；影响范围：当前第三方上行源；生效要求：需发布；填错时：不得进入截图、日志或手册；恢复：按来源方资料修正，重新校验并用新消息验证。 |
| 并发消费者 | 同一上行源同时运行的消费者数量；怎么选：不能超过可用分区数；默认 1；值从哪里来：Topic 分区数和吞吐测试；建议：1。 | 保存后进入当前上行源草稿；发布后才影响运行；影响范围：当前第三方上行源；生效要求：需发布；填错时：取值与真实来源不一致时，消息可能无法接入或无法进入规则；恢复：按来源方资料修正，重新校验并用新消息验证。 |
| 单次拉取上限 | 一次 poll 最多返回的记录数；怎么选：默认 500；避免批次过大导致处理超时或内存峰值；值从哪里来：单条处理耗时、内存和吞吐测试；建议：500。 | 保存后进入当前上行源草稿；发布后才影响运行；影响范围：当前第三方上行源；生效要求：需发布；填错时：取值与真实来源不一致时，消息可能无法接入或无法进入规则；恢复：按来源方资料修正，重新校验并用新消息验证。 |
| 毒消息重试次数 | 单条无法解析的消息在提交 offset 前最多重试几次；怎么选：默认 2；永久错误应尽快记录失败，避免阻塞分区；值从哪里来：数据质量和重试成本；建议：2。 | 保存后进入当前上行源草稿；发布后才影响运行；影响范围：当前第三方上行源；生效要求：需发布；填错时：取值与真实来源不一致时，消息可能无法接入或无法进入规则；恢复：按来源方资料修正，重新校验并用新消息验证。 |
| 无 offset 起点 | 消费组没有已提交 offset 时从哪里开始；怎么选：新上线只接新数据选 latest；明确需要补历史且已评估容量时选 earliest；值从哪里来：是否需要补历史数据；建议：以部署或对接资料为准；默认：latest。latest · 从最新开始：只消费启动后的新数据；不会自动补历史。earliest · 从最早开始：从可用最早位置补数据；可能产生大量历史流量。 | 保存后进入当前上行源草稿；发布后才影响运行；影响范围：当前第三方上行源；生效要求：需发布；填错时：取值与真实来源不一致时，消息可能无法接入或无法进入规则；恢复：按来源方资料修正，重新校验并用新消息验证。 |
参考表 4
| 看到什么现象 | 可能原因 | 先检查哪里 | 如何恢复 |
|---|---|---|---|
| 校验通过，但不能发布 | 发布阶段还会检查当前连接测试、HTTP 映射预览、规则动作和平台集成主密钥。 | 先看发布错误，再确认测试结果仍对应当前草稿。 | 补齐前置条件后重新发布；不要重复保存同一草稿。 |
| 连接测试通过，但接入一直为 0 | 发送方没有向同一 Broker/Topic 发送，或上行源未运行。 | 核对 RUNNING、Topic、发送时间和 FMQX 会话。 | 修正 Topic 或启动来源，再发送一条带唯一值的新消息。 |
| 接入增加，但连接器确认不增加 | 映射、身份、规则条件或连接器动作失败。 | 查看失败阶段、规则绑定和连接器状态。 | 从最后成功层级的下一层修正，再发新消息。 |
| 连接器确认增加，但目标队列为空 | Exchange、Routing Key 与队列 Binding 不匹配，或消息被其他消费者取走。 | 核对队列、绑定、消费者和本次消息唯一值。 | 修正路由或建立隔离观察窗口，再验证实收。 |
| 严格关联持续失败 | 消息标识没有命中当前 FMQ 本地设备缓存。 | 核对 clientId/deviceKey 路径、设备同步和缓存状态。 | 完成设备建档与同步；试接入场景可评估逻辑透传。 |
生产操作提示
截图和示例用于说明操作位置与判断方法，不代表当前生产状态。涉及发布、启停、重绑、重试或删除时，应先确认目标环境、影响范围和恢复方案，并由授权人员执行。
蜂信物联（FastBee） · FMQ 操作手册