---
title: MQTT Source 集成
---

# MQTT Source 集成

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

## 适用场景

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

## 前提条件

- 已了解管道结构与创建流程，参考 [数据集成](data-integration.md)；Source 连接器总览见 [Source](data-integration-input.md)
- 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](data-integration-processors-expressions.md)