MQTT Source 集成

本文介绍如何在管道的 Source 步骤中使用 MQTT 作为数据来源。

适用场景

• 接入设备遥测数据
• 汇聚边缘网关上传数据

前提条件

• 已了解管道结构与创建流程，参考 数据集成；Source 连接器总览见 Source
• MQTT 服务可访问
• 已规划订阅主题与客户端认证方式

配置步骤

1. 在“数据管道”中创建管道，完成基础信息配置。
2. 在 Source 步骤选择 MQTT。
3. 配置 urls、client_id、topics、qos 等参数。
4. 点击“测试”验证连接。
5. 继续完成 Processors、Sink 与确认步骤。

配置示例

可参考以下参数示例：

参数	建议值
urls	tcp://127.0.0.1:1883
client_id	source-docs-client
topics	flowmq/mqtt/kafka
qos	0

必要表单参数

以表单中带 * 的字段为必填：

参数	说明
urls*	MQTT broker 地址列表；须带协议前缀（如 tcp://127.0.0.1:1883 或 mqtt://127.0.0.1:1883），不可只写 host:port
client_id*	客户端标识
connect_timeout*	建连超时时间
keepalive*	空闲多久发送 keepalive（秒）
topics*	订阅主题列表
qos*	投递等级（0 / 1 / 2）
clean_session*	是否使用非持久会话
auto_replay_nacks*	下游拒绝（nack）时是否自动重放

参数建议

• 主题命名应体现设备域与业务域
• QoS 与吞吐目标保持平衡；高吞吐场景优先评估 QoS 0/1
• client_id 在集群多实例场景下需保证唯一

注意事项

• 注意保留消息与会话状态影响
• 设备侧时钟偏差建议在 Processors 侧处理，详见 Processors