---
title: MQTT Sink 集成
---

# MQTT Sink 集成

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

## 前提条件

- 已了解管道结构与创建流程，参考 [数据集成](data-integration.md)；Sink 连接器总览见 [Sink](data-integration-output.md)
- 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 · 示例场景](data-integration-output.md#示例场景)。

可参考以下参数示例：

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

## 必要表单参数

参数名后带 `*` 表示控制台必填。「支持表达式」为「是」时，可使用 `${! ... }` 按消息动态取值，详见 [Processors](data-integration-processors-expressions.md)。

| 参数 | 支持表达式 | 说明 |
|------|------------|------|
| `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 可能降低吞吐