MQTT

本文介绍如何将 FlowMQ 数据发布到 MQTT 系统。

前提条件

• 已了解管道结构与创建流程，参考 数据集成；Sink 连接器总览见 Sink
• MQTT 服务可访问
• 已规划发布主题与 QoS 策略

配置步骤

1. 在“数据管道”中创建管道，完成基础信息配置。
2. 在 Source 步骤选择 Kafka（Topic：flowmq.mqtt.kafka），Processors 按需配置。
3. 在 Sink 步骤选择 MQTT，配置 urls、topic、qos 等参数。
4. 点击“测试”验证连接。
5. 继续完成确认步骤。

配置示例

以下示例将 Kafka Topic flowmq.mqtt.kafka 中的样例消息写入目标系统。样例 JSON 与 Source 约定见 Sink · 示例场景。

可参考以下参数示例：

参数	建议值
urls	tcp://127.0.0.1:1883
client_id	sink-mqtt-demo
topic	flowmq/sink/demo
qos	0
retained	false

必要表单参数

参数名后带 * 表示控制台必填。「支持表达式」为「是」时，可使用 ${! ... } 按消息动态取值，详见 Processors。

参数	支持表达式	说明
urls*	否	MQTT broker 地址列表，填写完整 URI（如 tcp://127.0.0.1:1883）
client_id*	否	客户端标识
connect_timeout*	否	建连超时时间
keepalive*	否	心跳间隔（秒）
topic*	是	发布主题
qos*	否	消息 QoS 等级
write_timeout*	否	写超时时间
retained*	否	是否保留消息
max_in_flight*	否	并发写入上限

参数建议

• topic 宜便于订阅与维护；也可按消息动态指定，例如 ${! json("source") }
• 高吞吐场景宜选用较低 QoS，减轻写放大

注意事项

• 若某 MQTT 主题的数据会流入本管道读取的 Kafka Topic，Sink 请勿再发布到该主题，以免形成消息环路
• 过高的 QoS 可能降低吞吐