SeaTunnel MQTT Sink Connector 使用指南配置、源码原理与最佳实践【免费下载链接】seatunnelSeaTunnel is a multimodal, high-performance, distributed, massive data integration tool.项目地址: https://gitcode.com/GitHub_Trending/se/seatunnel本指南以 docs/en/connectors/sink/Mqtt.md 为骨架结合connector-mqtt模块的源码与测试展开面向需要将 SeaTunnel 管道数据发布到 MQTT Broker如 EMQX、Mosquitto、HiveMQ 等的开发者。读完本文你将掌握 MQTT Sink 的全部参数语义、投递语义边界、源码级实现原理并可直接套用文中两个完整任务配置将管道数据以 JSON 或文本格式写入指定 Topic。支持引擎与适用场景MQTT Sink Connector 用于将数据写入 MQTT Broker兼容 MQTT 3.1.1 协议底层基于 Eclipse Paho 客户端库实现pom 中固定依赖org.eclipse.paho.client.mqttv31.2.5见 connector-mqtt/pom.xml。该连接器支持以下引擎SeaTunnel ZetaFlinkSpark典型使用场景包括将 SeaTunnel 管道产出的数据发布到 IoT 端点对接轻量级消息代理作为设备侧数据下发的出口与 MQTT Source 或边缘采集链路配合形成数据中继闭环。消息默认以 JSON 序列化也可以按分隔符文本text格式发布到可配置的 MQTT Topic 中。核心特性与投递语义边界连接器在 connector-v2-features 维度上的能力标注为不支持 exactly-once、不支持 CDC、不支持 timer flush。在此基础上文档明确给出了投递语义说明这是选型时最需要关注的约束QoS0提供at-most-once最多一次投递QoS1提供best-effort at-least-once尽力而为的至少一次投递。由于clean_sessiontrue是默认值也是无状态运行所必需的客户端断连期间未确认的消息可能丢失。若设置clean_sessionfalseBroker 会在 writer 运行期间保留会话状态但当前 Sink 为每个 writer 动态生成唯一 client id并未暴露client_id配置项因此任务重启后无法复用稳定 client id。文档明确建议作业重启的恢复应依赖上游重放upstream replay与 MQTT QoS而非稳定的 Sink 客户端 id。Sink 参数详解参数总览表名称类型必填默认值说明urlstring是-MQTT Broker 连接地址必须包含协议、主机与端口如tcp://broker.example.com:1883topicstring是-要发布消息的 MQTT Topic如iot/sensors/temperatureusernamestring否-MQTT Broker 认证用户名匿名访问时留空passwordstring否-MQTT Broker 认证密码匿名访问时留空qosint否1MQTT 服务质量等级0为最多一次1为至少一次formatstring否json序列化格式json将每行序列化为 JSON 对象text将每行序列化为分隔文本field_delimiterstring否,format text时使用的字段分隔符如,、|、\tbatch_sizeint否1发送前缓冲的消息条数checkpoint 时也会强制冲刷缓冲区retry_timeoutint否5000出现瞬时网络故障时重试发布的毫秒上限超过后任务失败connection_timeoutint否30MQTT 建连超时时间秒clean_sessionboolean否true是否使用干净会话true丢弃历史会话状态false保留会话状态common-optionsconfig否-Sink 插件通用参数见 Sink Common Options以上默认值与必填关系均可在 MqttSinkOptions.java 和 MqttSinkFactory.java 中得到印证——工厂的optionRule()将url、topic声明为必填其余全部为可选。必填参数url [string]MQTT Broker 连接地址必须包含协议、主机和端口。例如tcp://broker.example.com:1883。连接配置在 writer 构造阶段完成连接失败会抛出带MQTT-01CONNECTION_FAILED错误码的异常。topic [string]要发布消息的 MQTT Topic例如iot/sensors/temperature。所有写入该 Sink 的行都会发布到这个 Topic。认证参数username / password [string]MQTT Broker 的认证凭据匿名访问时两者均可留空。从 MqttSinkWriter.java 的buildConnectOptions可以看到实现细节仅当 username/password 非空时才调用setUserName/setPassword密码以字符数组传入。投递质量与序列化qos [int]发布消息的 MQTT 服务质量等级取值范围为 0 或 10— 最多一次fire and forget1— 至少一次broker 确认投递默认值。注意源码在 writer 构造时会校验取值范围qos若小于 0 或大于 1 会直接抛出IllegalArgumentExceptionMQTT QoS must be 0 (at-most-once) or 1 (at-least-once)对应测试见 MqttSinkWriterTest.java 的testInvalidQosThrowsException。MQTT 3.1.1 的 QoS2恰好一次在当前实现中不支持。format [string]输出消息的序列化格式支持json— 每行序列化为一个 JSON 对象默认text— 每行序列化为分隔符纯文本分隔符由field_delimiter控制。源码通过createSerializationSchema按 format 分派json使用JsonSerializationSchema依赖 seatunnel-format-jsontext使用TextSerializationSchema依赖 seatunnel-format-text。格式非法如xml会抛出IllegalArgumentException对应测试testInvalidFormatThrowsException。field_delimiter [string]format为text时的字段分隔符默认,。示例,、|、\t。自定义分隔符的序列化行为由测试testCustomFieldDelimiter覆盖验证。缓冲、重试与连接batch_size [int]发送前缓冲的消息条数默认1每条消息立即发送。调大该值可减少单条消息开销、提升吞吐。缓冲的消息会在每次 checkpointprepareCommit调用flushBuffer以及writer 关闭时自动冲刷。测试testBatchWriteFlushesOnThreshold验证了写满阈值才发布的行为batch_size3 时写入 2 条不发布第 3 条写入触发全部 3 条发布。retry_timeout [int]出现瞬时网络故障时重试发布的毫秒时间上限默认5000。源码在publishWithRetry中实现在超时窗口内轮询连接状态每次重试间隔固定退避 200msRETRY_BACKOFF_MS期间若连接恢复则立即发布成功返回窗口耗尽仍失败则抛出IOException错误码为MQTT-02PUBLISH_FAILED。测试testWriteWithRetrySuccess验证了首次发布失败、重试后成功的路径testWriteTimeoutAfterRetries验证了超时后抛错。connection_timeout [int]MQTT 建连超时时间单位秒默认30透传给 Paho 的MqttConnectOptions.setConnectionTimeout。clean_session [boolean]是否使用干净 MQTT 会话默认truetrue— Broker 丢弃任何历史会话状态适合无状态运行大多数场景推荐false— Broker 在 writer 运行期间为生成的 client id 保留会话状态有助于应对瞬时断连但可能造成 Broker 侧状态堆积且由于 client id 不固定跨作业重启无法提供稳定恢复。源码层面clean_sessionfalse时会打 WARN 日志提示状态堆积风险clean_sessionfalse may cause broker-side state accumulation且无论何种取值Paho 都会开启setAutomaticReconnect(true)断连后由 Paho 自动重连恢复。通用参数Sink 插件通用参数详见 Sink Common Options其中与本连接器直接相关的是plugin_input指定当前插件处理的上游数据集合未指定时处理配置文件中的前一个插件输出旧名source_table_name已废弃请迁移到plugin_input需要与上游插件配合设置plugin_output示例任务中可见该用法。源码级原理Writer 如何工作从 MqttSinkWriter.java 可以看出整个写入链路的几个关键设计动态 Client ID 防冲突每个并行子任务subtask以seatunnel_mqtt_sink_task_为前缀拼接子任务索引 随机 UUID 生成全局唯一 client id避免并行作业间相互踢线connection hijacking。内存持久化使用 Paho 的MemoryPersistence而非文件持久化避免容器化部署下的磁盘 I/O。同步发布保证顺序flushBuffer对缓冲内消息逐条同步调用publish保证消息发布顺序。无状态 SinkabortPrepare为空实现Stateless sink — nothing to roll backprepareCommit仅冲刷缓冲不涉及跨 Broker 的事务提交——这也从机制上解释了为何连接器无法提供 exactly-once。插件注册MqttSink的getPluginName()返回MQTT即配置文件中 Sink 块使用的插件名MqttSinkFactory通过AutoService(Factory.class)注册并声明url、topic为必填项。错误处理方面连接器定义了统一的错误码枚举MqttConnectorErrorCode.javaMQTT-01连接失败、MQTT-02发布失败、MQTT-03配置非法、MQTT-04接收失败供 Source 使用。另外该连接器需在 config/plugin_config 中启用connector-mqtt才会被打包加载构建发布包时请确认该条目未被注释。性能考量当前实现为同步发送以保证发布顺序官方文档给出的典型吞吐参考值本地网络QoS 0约 10,000 条/秒QoS 1约 5,000 条/秒需要 Broker ACK。提升吞吐的推荐手段调大batch_size降低单条消息开销例如batch_size 100在可接受最多一次投递时将qos降为0提高 SeaTunnel 并行度把负载分摊到多个 MQTT 客户端每个并行子任务独立连接超高吞吐场景可评估改用 Kafka Sink。以上吞吐数字为文档给出的经验参考值实际表现取决于 Broker 部署、网络与消息大小建议在目标环境自行压测验证。任务示例示例一以 JSON 格式写入 MQTTenv { parallelism 1 job.mode BATCH job.name SeaTunnel_MQTT_Sink } source { FakeSource { row.num 16 schema { fields { id bigint name string age int } } plugin_output fake } } sink { MQTT { plugin_input fake url tcp://mqtt-broker:1883 topic test/seatunnel/sink qos 1 format json } }该作业使用 FakeSource 生成 16 行数据写入test/seatunnel/sinkTopic。由于format json每行会被序列化为一条独立的 JSON 消息发布。示例二认证 Broker 文本格式env { parallelism 1 job.mode BATCH } source { FakeSource { row.num 10 schema { fields { id bigint content string } } } } sink { MQTT { url tcp://secure-broker.example.com:1883 topic data/pipeline/output username seatunnel_user password secret qos 1 format text field_delimiter | retry_timeout 10000 connection_timeout 60 } }当format text时每行序列化为以field_delimiter分隔的纯文本行用|作为分隔符retry_timeout 10000将瞬时故障重试窗口放宽到 10 秒connection_timeout 60将建连超时放宽到 60 秒适合网络不稳定的远端 Broker。版本与变更记录MQTT Sink Connector 是新增的 Sink 插件首个版本在 docs/en/connectors/changelog/connector-mqtt.md 的变更日志中记录Add MQTT Sink Connector。使用时请以当前仓库connector-mqtt模块的配置为准。常见问题速查现象排查方向作业启动即失败且错误码 MQTT-01Broker 地址不可达、协议/端口书写错误或认证失败检查url是否包含完整协议主机端口发布阶段失败且错误码 MQTT-02网络瞬时故障超过retry_timeout窗口可适当调大该参数配置校验报 MQTT QoS must be 0 or 1qos传入了 0/1 之外的值当前实现不支持 QoS 2配置校验报 unsupported formatformat只接受json与text任务重启后消息重复或丢失属预期行为依赖上游重放与 MQTT QoS 恢复而非 Sink 客户端会话需要恰好一次语义时请评估其他具备事务能力的 Sink并行度较高时出现互踢每个子任务已生成唯一 client id正常不会互踢若自行部署多个作业连同一 Broker注意 Broker 侧 client id 冲突策略【免费下载链接】seatunnelSeaTunnel is a multimodal, high-performance, distributed, massive data integration tool.项目地址: https://gitcode.com/GitHub_Trending/se/seatunnel创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 SeaTunnel MQTT Sink Connector 使用指南：配置、源码原理与最佳实践
SeaTunnel MQTT Sink Connector 使用指南配置、源码原理与最佳实践【免费下载链接】seatunnelSeaTunnel is a multimodal, high-performance, distributed, massive data integration tool.项目地址: https://gitcode.com/GitHub_Trending/se/seatunnel本指南以 docs/en/connectors/sink/Mqtt.md 为骨架结合connector-mqtt模块的源码与测试展开面向需要将 SeaTunnel 管道数据发布到 MQTT Broker如 EMQX、Mosquitto、HiveMQ 等的开发者。读完本文你将掌握 MQTT Sink 的全部参数语义、投递语义边界、源码级实现原理并可直接套用文中两个完整任务配置将管道数据以 JSON 或文本格式写入指定 Topic。支持引擎与适用场景MQTT Sink Connector 用于将数据写入 MQTT Broker兼容 MQTT 3.1.1 协议底层基于 Eclipse Paho 客户端库实现pom 中固定依赖org.eclipse.paho.client.mqttv31.2.5见 connector-mqtt/pom.xml。该连接器支持以下引擎SeaTunnel ZetaFlinkSpark典型使用场景包括将 SeaTunnel 管道产出的数据发布到 IoT 端点对接轻量级消息代理作为设备侧数据下发的出口与 MQTT Source 或边缘采集链路配合形成数据中继闭环。消息默认以 JSON 序列化也可以按分隔符文本text格式发布到可配置的 MQTT Topic 中。核心特性与投递语义边界连接器在 connector-v2-features 维度上的能力标注为不支持 exactly-once、不支持 CDC、不支持 timer flush。在此基础上文档明确给出了投递语义说明这是选型时最需要关注的约束QoS0提供at-most-once最多一次投递QoS1提供best-effort at-least-once尽力而为的至少一次投递。由于clean_sessiontrue是默认值也是无状态运行所必需的客户端断连期间未确认的消息可能丢失。若设置clean_sessionfalseBroker 会在 writer 运行期间保留会话状态但当前 Sink 为每个 writer 动态生成唯一 client id并未暴露client_id配置项因此任务重启后无法复用稳定 client id。文档明确建议作业重启的恢复应依赖上游重放upstream replay与 MQTT QoS而非稳定的 Sink 客户端 id。Sink 参数详解参数总览表名称类型必填默认值说明urlstring是-MQTT Broker 连接地址必须包含协议、主机与端口如tcp://broker.example.com:1883topicstring是-要发布消息的 MQTT Topic如iot/sensors/temperatureusernamestring否-MQTT Broker 认证用户名匿名访问时留空passwordstring否-MQTT Broker 认证密码匿名访问时留空qosint否1MQTT 服务质量等级0为最多一次1为至少一次formatstring否json序列化格式json将每行序列化为 JSON 对象text将每行序列化为分隔文本field_delimiterstring否,format text时使用的字段分隔符如,、|、\tbatch_sizeint否1发送前缓冲的消息条数checkpoint 时也会强制冲刷缓冲区retry_timeoutint否5000出现瞬时网络故障时重试发布的毫秒上限超过后任务失败connection_timeoutint否30MQTT 建连超时时间秒clean_sessionboolean否true是否使用干净会话true丢弃历史会话状态false保留会话状态common-optionsconfig否-Sink 插件通用参数见 Sink Common Options以上默认值与必填关系均可在 MqttSinkOptions.java 和 MqttSinkFactory.java 中得到印证——工厂的optionRule()将url、topic声明为必填其余全部为可选。必填参数url [string]MQTT Broker 连接地址必须包含协议、主机和端口。例如tcp://broker.example.com:1883。连接配置在 writer 构造阶段完成连接失败会抛出带MQTT-01CONNECTION_FAILED错误码的异常。topic [string]要发布消息的 MQTT Topic例如iot/sensors/temperature。所有写入该 Sink 的行都会发布到这个 Topic。认证参数username / password [string]MQTT Broker 的认证凭据匿名访问时两者均可留空。从 MqttSinkWriter.java 的buildConnectOptions可以看到实现细节仅当 username/password 非空时才调用setUserName/setPassword密码以字符数组传入。投递质量与序列化qos [int]发布消息的 MQTT 服务质量等级取值范围为 0 或 10— 最多一次fire and forget1— 至少一次broker 确认投递默认值。注意源码在 writer 构造时会校验取值范围qos若小于 0 或大于 1 会直接抛出IllegalArgumentExceptionMQTT QoS must be 0 (at-most-once) or 1 (at-least-once)对应测试见 MqttSinkWriterTest.java 的testInvalidQosThrowsException。MQTT 3.1.1 的 QoS2恰好一次在当前实现中不支持。format [string]输出消息的序列化格式支持json— 每行序列化为一个 JSON 对象默认text— 每行序列化为分隔符纯文本分隔符由field_delimiter控制。源码通过createSerializationSchema按 format 分派json使用JsonSerializationSchema依赖 seatunnel-format-jsontext使用TextSerializationSchema依赖 seatunnel-format-text。格式非法如xml会抛出IllegalArgumentException对应测试testInvalidFormatThrowsException。field_delimiter [string]format为text时的字段分隔符默认,。示例,、|、\t。自定义分隔符的序列化行为由测试testCustomFieldDelimiter覆盖验证。缓冲、重试与连接batch_size [int]发送前缓冲的消息条数默认1每条消息立即发送。调大该值可减少单条消息开销、提升吞吐。缓冲的消息会在每次 checkpointprepareCommit调用flushBuffer以及writer 关闭时自动冲刷。测试testBatchWriteFlushesOnThreshold验证了写满阈值才发布的行为batch_size3 时写入 2 条不发布第 3 条写入触发全部 3 条发布。retry_timeout [int]出现瞬时网络故障时重试发布的毫秒时间上限默认5000。源码在publishWithRetry中实现在超时窗口内轮询连接状态每次重试间隔固定退避 200msRETRY_BACKOFF_MS期间若连接恢复则立即发布成功返回窗口耗尽仍失败则抛出IOException错误码为MQTT-02PUBLISH_FAILED。测试testWriteWithRetrySuccess验证了首次发布失败、重试后成功的路径testWriteTimeoutAfterRetries验证了超时后抛错。connection_timeout [int]MQTT 建连超时时间单位秒默认30透传给 Paho 的MqttConnectOptions.setConnectionTimeout。clean_session [boolean]是否使用干净 MQTT 会话默认truetrue— Broker 丢弃任何历史会话状态适合无状态运行大多数场景推荐false— Broker 在 writer 运行期间为生成的 client id 保留会话状态有助于应对瞬时断连但可能造成 Broker 侧状态堆积且由于 client id 不固定跨作业重启无法提供稳定恢复。源码层面clean_sessionfalse时会打 WARN 日志提示状态堆积风险clean_sessionfalse may cause broker-side state accumulation且无论何种取值Paho 都会开启setAutomaticReconnect(true)断连后由 Paho 自动重连恢复。通用参数Sink 插件通用参数详见 Sink Common Options其中与本连接器直接相关的是plugin_input指定当前插件处理的上游数据集合未指定时处理配置文件中的前一个插件输出旧名source_table_name已废弃请迁移到plugin_input需要与上游插件配合设置plugin_output示例任务中可见该用法。源码级原理Writer 如何工作从 MqttSinkWriter.java 可以看出整个写入链路的几个关键设计动态 Client ID 防冲突每个并行子任务subtask以seatunnel_mqtt_sink_task_为前缀拼接子任务索引 随机 UUID 生成全局唯一 client id避免并行作业间相互踢线connection hijacking。内存持久化使用 Paho 的MemoryPersistence而非文件持久化避免容器化部署下的磁盘 I/O。同步发布保证顺序flushBuffer对缓冲内消息逐条同步调用publish保证消息发布顺序。无状态 SinkabortPrepare为空实现Stateless sink — nothing to roll backprepareCommit仅冲刷缓冲不涉及跨 Broker 的事务提交——这也从机制上解释了为何连接器无法提供 exactly-once。插件注册MqttSink的getPluginName()返回MQTT即配置文件中 Sink 块使用的插件名MqttSinkFactory通过AutoService(Factory.class)注册并声明url、topic为必填项。错误处理方面连接器定义了统一的错误码枚举MqttConnectorErrorCode.javaMQTT-01连接失败、MQTT-02发布失败、MQTT-03配置非法、MQTT-04接收失败供 Source 使用。另外该连接器需在 config/plugin_config 中启用connector-mqtt才会被打包加载构建发布包时请确认该条目未被注释。性能考量当前实现为同步发送以保证发布顺序官方文档给出的典型吞吐参考值本地网络QoS 0约 10,000 条/秒QoS 1约 5,000 条/秒需要 Broker ACK。提升吞吐的推荐手段调大batch_size降低单条消息开销例如batch_size 100在可接受最多一次投递时将qos降为0提高 SeaTunnel 并行度把负载分摊到多个 MQTT 客户端每个并行子任务独立连接超高吞吐场景可评估改用 Kafka Sink。以上吞吐数字为文档给出的经验参考值实际表现取决于 Broker 部署、网络与消息大小建议在目标环境自行压测验证。任务示例示例一以 JSON 格式写入 MQTTenv { parallelism 1 job.mode BATCH job.name SeaTunnel_MQTT_Sink } source { FakeSource { row.num 16 schema { fields { id bigint name string age int } } plugin_output fake } } sink { MQTT { plugin_input fake url tcp://mqtt-broker:1883 topic test/seatunnel/sink qos 1 format json } }该作业使用 FakeSource 生成 16 行数据写入test/seatunnel/sinkTopic。由于format json每行会被序列化为一条独立的 JSON 消息发布。示例二认证 Broker 文本格式env { parallelism 1 job.mode BATCH } source { FakeSource { row.num 10 schema { fields { id bigint content string } } } } sink { MQTT { url tcp://secure-broker.example.com:1883 topic data/pipeline/output username seatunnel_user password secret qos 1 format text field_delimiter | retry_timeout 10000 connection_timeout 60 } }当format text时每行序列化为以field_delimiter分隔的纯文本行用|作为分隔符retry_timeout 10000将瞬时故障重试窗口放宽到 10 秒connection_timeout 60将建连超时放宽到 60 秒适合网络不稳定的远端 Broker。版本与变更记录MQTT Sink Connector 是新增的 Sink 插件首个版本在 docs/en/connectors/changelog/connector-mqtt.md 的变更日志中记录Add MQTT Sink Connector。使用时请以当前仓库connector-mqtt模块的配置为准。常见问题速查现象排查方向作业启动即失败且错误码 MQTT-01Broker 地址不可达、协议/端口书写错误或认证失败检查url是否包含完整协议主机端口发布阶段失败且错误码 MQTT-02网络瞬时故障超过retry_timeout窗口可适当调大该参数配置校验报 MQTT QoS must be 0 or 1qos传入了 0/1 之外的值当前实现不支持 QoS 2配置校验报 unsupported formatformat只接受json与text任务重启后消息重复或丢失属预期行为依赖上游重放与 MQTT QoS 恢复而非 Sink 客户端会话需要恰好一次语义时请评估其他具备事务能力的 Sink并行度较高时出现互踢每个子任务已生成唯一 client id正常不会互踢若自行部署多个作业连同一 Broker注意 Broker 侧 client id 冲突策略【免费下载链接】seatunnelSeaTunnel is a multimodal, high-performance, distributed, massive data integration tool.项目地址: https://gitcode.com/GitHub_Trending/se/seatunnel创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考