数据工程大数据批处理流处理【免费下载链接】seatunnelSeaTunnel is a next-generation super high-performance, distributed, massive data integration tool.项目地址https://gitcode.com/gh_mirrors/sea/seatunnel点击查看免费下载SeaTunnel 的 RabbitMQ Sink 连接器用于将上游数据写入 RabbitMQ 队列它在内部把每一行SeaTunnelRow序列化为 JSON 消息后通过 AMQP 协议发布到指定队列。本文以官方文档 docs/en/connector-v2/sink/Rabbitmq.md 为主体结合 connector-rabbitmq 模块源码与 RabbitMQ 端到端测试 配置完整讲解全部连接器参数、可用配置写法、底层消息发布链路与常见故障排查思路帮助你在真实作业中正确、稳定地把 SeaTunnel 数据落进 RabbitMQ。概述RabbitMQ Sink 连接器是 SeaTunnel Connector V2 体系中的一个 Sink 插件插件标识为RabbitMQ可通过 plugin-mapping.properties 中的seatunnel.sink.RabbitMQ connector-rabbitmq映射确认。它用于将数据写入 RabbitMQ与同模块下的 RabbitMQ Source 配合即可完成 RabbitMQ → RabbitMQ、任意 Source → RabbitMQ 的数据管道搭建。从实现角度看该连接器基于官方com.rabbitmq:amqp-client客户端pom.xml 中声明的版本为 5.9.0并依赖seatunnel-format-json模块完成行数据的 JSON 编码。因此它面向的是以 JSON 消息形式投递结构化行数据这一典型场景。核心特性与语义保证原文档在 Key features 一节明确标注exactly-once精确一次不支持复选框为未选中状态详见 connector-v2-features.md 中关于连接器能力矩阵的说明。这意味着当前 RabbitMQ Sink 提供的是at-least-once至少一次级别的投递语义消息一旦写入即认为投递完成作业失败或重启时可能出现重复消息下游消费端需要自行做幂等或去重。若业务对重复敏感应结合消息内容设计去重逻辑。连接器参数详解原文档给出的完整参数表如下名称类型是否必填默认值hoststring是-portint是-virtual_hoststring是-usernamestring是-passwordstring是-queue_namestring是-urlstring否-network_recovery_intervalint否-topology_recovery_enabledboolean否-automatic_recovery_enabledboolean否-use_correlation_idboolean否falseconnection_timeoutint否-rabbitmq.configmap否-common-options-否-逐一说明如下host [string]连接 RabbitMQ Broker 时使用的默认主机地址例如rabbitmq-e2e或127.0.0.1。该值最终会传给ConnectionFactory.setHost()。port [int]连接时使用的默认端口RabbitMQ 默认 AMQP 端口为5672。对应源码中的ConnectionFactory.setPort()。virtual_host [string]连接 Broker 时使用的虚拟主机virtual host。RabbitMQ 通过 vhost 实现多租户隔离常用默认值/。在源码中通过ConnectionFactory.setVirtualHost()设置。username [string]连接 Broker 时使用的 AMQP 用户名对应ConnectionFactory.setUsername()。password [string]连接 Broker 时使用的密码对应ConnectionFactory.setPassword()。注意虽然原文档表格将username、password标为必填但源码层面稍有差别。RabbitmqSinkFactory.java 使用bundled(USERNAME, PASSWORD)将它们声明为捆绑参数——即要么同时提供、要么同时不提供而 RabbitmqSink.java 的prepare()阶段则通过CheckConfigUtil.checkAllExists校验host、port、virtual_host、username、password、queue_name六个字段必须全部存在。实际配置时建议六个字段全部显式给出与官方文档保持一致。queue_name [string]消息要写入的队列名称。连接器初始化时见 RabbitmqClient.java会调用channel.queueDeclare(queueName, true, false, false, null)以持久化方式声明该队列durabletrue非排他、非自动删除因此目标队列不需要预先手工创建若队列已存在则执行的是幂等的声明操作。url [string]便捷参数用于一次性设置 AMQP URI 中的多个字段host、port、username、password、virtual host。当配置了url时源码会优先走factory.setUri(config.getUri())分支见 RabbitmqClient.java此时host等字段仍然建议填写以满足必填校验但实际连接参数以 URI 为准。URI 解析失败时抛出错误码RABBITMQ-07parse uri failed。schema [Config] → fields [Config]上游数据的字段结构定义。这是可选参数用于让连接器明确上游SeaTunnelRow的 schema不配置时连接器会依据上游算子传递的行类型工作。典型写法schema { fields { id bigint name string score double } }network_recovery_interval [int]自动恢复automatic recovery在尝试重连之前的等待时长单位为毫秒ms。对应源码中的factory.setNetworkRecoveryInterval()仅在自动恢复启用时生效。topology_recovery_enabled [boolean]设为true时启用拓扑恢复topology recovery即连接恢复后自动重新声明队列、交换器等拓扑对象。对应factory.setTopologyRecoveryEnabled()。automatic_recovery_enabled [boolean]设为true时启用连接恢复connection recovery当网络异常断开后客户端会自动尝试重连。对应factory.setAutomaticRecoveryEnabled()。use_correlation_id [boolean]是否启用 correlation id 机制用于在消息接收方为消息附加唯一 ID 以在ack 失败的场景下做去重。注意该参数语义主要面向 RabbitMQ Source 消费侧Sink 侧在源码中同样保留了解析入口见 RabbitmqConfig.java默认值为false。connection_timeout [int]TCP 建连超时时间单位为毫秒设为0表示无限等待。对应factory.setConnectionTimeout()。rabbitmq.config [map]除上述显式参数外用户还可以通过该 map 指定 RabbitMQ 客户端的其他非必填参数覆盖官方 RabbitMQ 配置文档中定义的所有客户端参数。在 RabbitmqConfig.java 的parseSinkOptionProperties中map 内所有 key 会被转为小写后存入sinkOptionProps后续传递给客户端工厂使用。例如rabbitmq.config { requested-heartbeat 10 connection-timeout 10 }common optionsSink 插件公共参数详见 Sink Common Options。其中最有价值的是source_table_name当作业中存在多个 Source/Transform/Sink 时用它显式指定当前 Sink 处理哪一张上游结果表配合上游的result_table_name使用。源码中还可用的其他参数原文档参数表之外RabbitmqConfig.java 中还定义了以下可解析参数供进阶场景使用名称类型说明routing_keystring发布消息时使用的 routing key为空时消息直接发往默认交换器direct 到队列exchangestring发布消息时使用的交换器名称默认默认交换器requested_channel_maxint初始请求的最大 channel 数量对应factory.setRequestedChannelMax()requested_frame_maxint请求的最大帧大小对应factory.setRequestedFrameMax()requested_heartbeatint请求的心跳超时时间对应factory.setRequestedHeartbeat()prefetch_countlong未 ack 时最多接收的消息条数QoS在连接初始化时通过channel.basicQos(prefetchCount, true)设置delivery_timeoutint投递最大等待时间for_e2e_testingboolean用于标识 E2E 测试模式这些参数虽未出现在文档参数表中但源码解析与客户端装配均已就绪RabbitmqClient.java 对每个非空字段逐一调用对应 setter使用时请以当前仓库源码版本为准。配置示例示例一最简配置原文档给出的 simple 示例sink { RabbitMQ { host rabbitmq-e2e port 5672 virtual_host / username guest password guest queue_name test1 rabbitmq.config { requested-heartbeat 10 connection-timeout 10 } } }这是最常见的写法host/port/virtual_host/username/password/queue_name六个必填字段齐全rabbitmq.config内补充心跳与连接超时等客户端级调优参数。示例二使用 url 连接当 RabbitMQ 提供了 AMQP URI例如amqp://guest:guestlocalhost:5672/%2F时可写成sink { RabbitMQ { url amqp://guest:guestlocalhost:5672/%2F host localhost port 5672 virtual_host / username guest password guest queue_name test1 } }示例三通过交换器与 routing key 发布默认routing_key为空时源码走channel.basicPublish(, queue_name, null, msg)即通过默认交换器把消息直接投递到指定队列当配置了routing_key后会改走channel.basicPublish(exchange, routing_key, false, false, null, msg)的交换器路由模式RabbitmqClient.javasink { RabbitMQ { host rabbitmq-e2e port 5672 virtual_host / username guest password guest queue_name test1 exchange my_exchange routing_key my_routing_key } }示例四完整端到端作业来自 E2E 测试仓库自带的 RabbitMQ 端到端测试配置 rabbitmq-to-rabbitmq.conf 展示了带 schema 的完整用法——RabbitMQ Source 读取队列testSink 写入队列test1env { parallelism 1 job.mode STREAMING } source { RabbitMQ { host rabbitmq-e2e port 5672 virtual_host / username guest password guest queue_name test for_e2e_testing true schema { fields { id bigint c_map mapstring, smallint c_array arraytinyint c_string string c_boolean boolean c_tinyint tinyint c_smallint smallint c_int int c_bigint bigint c_float float c_double double c_decimal decimal(2, 1) c_bytes bytes c_date date c_timestamp timestamp } } } } transform { } sink { RabbitMQ { host rabbitmq-e2e port 5672 virtual_host / username guest password guest queue_name test1 } }其中schema.fields展示了 SeaTunnel 支持的主要数据类型整数族、浮点族、字符串、布尔、decimal、bytes、date、timestamp、map、arraySink 会依据该 schema 完成 JSON 序列化。消息写入链路从 SeaTunnelRow 到 AMQP 消息理解底层调用链有助于排查消息格式不对消息没进队列等问题。以源码为准写入链路如下插件装配作业启动时SPI 机制通过AutoService(SeaTunnelSink.class)加载 RabbitmqSink.java其prepare()校验必填配置并构建RabbitmqConfigcreateWriter()产出RabbitmqSinkWriter。客户端初始化RabbitmqClient.java 构造时依次完成构建ConnectionFactory根据是否配置url决定走 URI 还是 host/port/vhost/user/password 分支→ 建立Connection→ 创建Channel→ 设置prefetch_countQoS → 声明持久化队列。逐行序列化RabbitmqSinkWriter.java 对每一行调用JsonSerializationSchema.serialize(element)将SeaTunnelRow按 schema 编码为 JSON 字节数组。发布消息rabbitMQClient.write(bytes)根据是否配置routing_key选择默认交换器直投队列或交换器 routing key 路由两种basicPublish方式发送。发送异常时若logFailuresOnly为 true 则仅记录错误日志否则抛出错误码RABBITMQ-04send messages failed终止任务。资源回收close()依次关闭 Channel 与 Connection任一关闭失败都会抛出RABBITMQ-03close connection failed。另外值得注意RabbitmqSinkWriter的prepareCommit()返回Optional.empty()说明该 Sink 不参与两阶段提交与不支持 exactly-once的特性声明一致。错误码速查连接器定义了独立的错误码枚举见 RabbitmqConnectorErrorCode.java错误码描述RABBITMQ-01处理队列消费者关闭信号失败RABBITMQ-02创建 RabbitMQ 客户端失败RABBITMQ-03关闭连接失败RABBITMQ-04发送消息失败RABBITMQ-05创建检查点期间消息无法 ackRABBITMQ-06消息无法通过 basicReject 拒绝RABBITMQ-07解析 URI 失败RABBITMQ-08初始化 SSL 上下文失败RABBITMQ-09设置 SSL 工厂失败作业日志中出现上述错误码时可据此快速定位是建连、发送、URI 解析还是 SSL 环节出了问题。常见问题与使用建议消息未写入预期队列确认queue_name拼写、vhost 是否正确默认写路径使用默认交换器直投队列若消息被路由走应检查是否误配了routing_key/exchange。连接频繁断开可显式设置automatic_recovery_enabled true并配合network_recovery_interval控制重连间隔同时通过requested-heartbeat保持心跳。消息格式Sink 固定输出 JSON 文本消息下游消费方应按 JSON 解析字段类型与schema声明不一致可能导致序列化异常。重复投递当前连接器不提供 exactly-once 保证关键业务建议在下游按use_correlation_id或业务主键做去重。并发写入每个 Sink 任务使用独立 Channel 发布可通过parallelism控制写入并发度但需留意目标队列的吞吐承受能力。版本变更记录原文档 Changelog 记录如下next version新增 RabbitMQ Sink 连接器将连接器自定义配置前缀改为 map 形式即rabbitmq.config的引入对应 PR 3719。这意味着rabbitmq.config这种 map 风格的扩展参数写法是较新版本的行为升级插件后请使用 map 写法传递自定义客户端参数。延伸阅读连接器官方文档docs/en/connector-v2/sink/Rabbitmq.mdSink 公共参数说明docs/en/connector-v2/sink/common-options.md连接器能力特性矩阵docs/en/concept/connector-v2-features.md配置解析实现RabbitmqConfig.java客户端与发布实现RabbitmqClient.java插件工厂与参数规则RabbitmqSinkFactory.javaSink 主类与参数校验RabbitmqSink.java端到端测试配置rabbitmq-to-rabbitmq.conf插件映射关系plugin-mapping.properties赞分享数据工程大数据批处理流处理【免费下载链接】seatunnelSeaTunnel is a next-generation super high-performance, distributed, massive data integration tool.项目地址https://gitcode.com/gh_mirrors/sea/seatunnel点击查看免费下载相关推荐SeaTunnel RabbitMQ Sink 连接器完整指南参数详解、队列声明机制与消息写入实战SeaTunnel RabbitMQ Sink 连接器完整指南参数详解、队列声明机制与消息写入实战 本文基于 SeaTunnel 仓库中 docs/zh/co数据集成ETL大数据批处理流处理变更数据捕获SeaTunnel RabbitMQ Sink 连接器实战指南配置详解、消息发布与故障恢复SeaTunnel RabbitMQ Sink 连接器实战指南配置详解、消息发布与故障恢复 RabbitMQ 是业界广泛使用的开源消息中间件而 SeaTun数据集成ETL大数据批处理流处理变更数据捕获SeaTunnel ActiveMQ Sink 连接器完全指南配置、JSON 消息序列化机制与源码级解析SeaTunnel ActiveMQ Sink 连接器完全指南配置、JSON 消息序列化机制与源码级解析 本文基于 SeaTunnel 仓库中的 Active数据集成ETL大数据批处理流处理变更数据捕获上一篇FastRTC实时通信质量保障从音频处理到连接管理的全链路测试下一篇如何使用Claude Code Hooks Mastery实现自动化部署流程创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
SeaTunnel RabbitMQ Sink 连接器完全指南：参数详解、JSON 序列化与消息发布原理
数据工程大数据批处理流处理【免费下载链接】seatunnelSeaTunnel is a next-generation super high-performance, distributed, massive data integration tool.项目地址https://gitcode.com/gh_mirrors/sea/seatunnel点击查看免费下载SeaTunnel 的 RabbitMQ Sink 连接器用于将上游数据写入 RabbitMQ 队列它在内部把每一行SeaTunnelRow序列化为 JSON 消息后通过 AMQP 协议发布到指定队列。本文以官方文档 docs/en/connector-v2/sink/Rabbitmq.md 为主体结合 connector-rabbitmq 模块源码与 RabbitMQ 端到端测试 配置完整讲解全部连接器参数、可用配置写法、底层消息发布链路与常见故障排查思路帮助你在真实作业中正确、稳定地把 SeaTunnel 数据落进 RabbitMQ。概述RabbitMQ Sink 连接器是 SeaTunnel Connector V2 体系中的一个 Sink 插件插件标识为RabbitMQ可通过 plugin-mapping.properties 中的seatunnel.sink.RabbitMQ connector-rabbitmq映射确认。它用于将数据写入 RabbitMQ与同模块下的 RabbitMQ Source 配合即可完成 RabbitMQ → RabbitMQ、任意 Source → RabbitMQ 的数据管道搭建。从实现角度看该连接器基于官方com.rabbitmq:amqp-client客户端pom.xml 中声明的版本为 5.9.0并依赖seatunnel-format-json模块完成行数据的 JSON 编码。因此它面向的是以 JSON 消息形式投递结构化行数据这一典型场景。核心特性与语义保证原文档在 Key features 一节明确标注exactly-once精确一次不支持复选框为未选中状态详见 connector-v2-features.md 中关于连接器能力矩阵的说明。这意味着当前 RabbitMQ Sink 提供的是at-least-once至少一次级别的投递语义消息一旦写入即认为投递完成作业失败或重启时可能出现重复消息下游消费端需要自行做幂等或去重。若业务对重复敏感应结合消息内容设计去重逻辑。连接器参数详解原文档给出的完整参数表如下名称类型是否必填默认值hoststring是-portint是-virtual_hoststring是-usernamestring是-passwordstring是-queue_namestring是-urlstring否-network_recovery_intervalint否-topology_recovery_enabledboolean否-automatic_recovery_enabledboolean否-use_correlation_idboolean否falseconnection_timeoutint否-rabbitmq.configmap否-common-options-否-逐一说明如下host [string]连接 RabbitMQ Broker 时使用的默认主机地址例如rabbitmq-e2e或127.0.0.1。该值最终会传给ConnectionFactory.setHost()。port [int]连接时使用的默认端口RabbitMQ 默认 AMQP 端口为5672。对应源码中的ConnectionFactory.setPort()。virtual_host [string]连接 Broker 时使用的虚拟主机virtual host。RabbitMQ 通过 vhost 实现多租户隔离常用默认值/。在源码中通过ConnectionFactory.setVirtualHost()设置。username [string]连接 Broker 时使用的 AMQP 用户名对应ConnectionFactory.setUsername()。password [string]连接 Broker 时使用的密码对应ConnectionFactory.setPassword()。注意虽然原文档表格将username、password标为必填但源码层面稍有差别。RabbitmqSinkFactory.java 使用bundled(USERNAME, PASSWORD)将它们声明为捆绑参数——即要么同时提供、要么同时不提供而 RabbitmqSink.java 的prepare()阶段则通过CheckConfigUtil.checkAllExists校验host、port、virtual_host、username、password、queue_name六个字段必须全部存在。实际配置时建议六个字段全部显式给出与官方文档保持一致。queue_name [string]消息要写入的队列名称。连接器初始化时见 RabbitmqClient.java会调用channel.queueDeclare(queueName, true, false, false, null)以持久化方式声明该队列durabletrue非排他、非自动删除因此目标队列不需要预先手工创建若队列已存在则执行的是幂等的声明操作。url [string]便捷参数用于一次性设置 AMQP URI 中的多个字段host、port、username、password、virtual host。当配置了url时源码会优先走factory.setUri(config.getUri())分支见 RabbitmqClient.java此时host等字段仍然建议填写以满足必填校验但实际连接参数以 URI 为准。URI 解析失败时抛出错误码RABBITMQ-07parse uri failed。schema [Config] → fields [Config]上游数据的字段结构定义。这是可选参数用于让连接器明确上游SeaTunnelRow的 schema不配置时连接器会依据上游算子传递的行类型工作。典型写法schema { fields { id bigint name string score double } }network_recovery_interval [int]自动恢复automatic recovery在尝试重连之前的等待时长单位为毫秒ms。对应源码中的factory.setNetworkRecoveryInterval()仅在自动恢复启用时生效。topology_recovery_enabled [boolean]设为true时启用拓扑恢复topology recovery即连接恢复后自动重新声明队列、交换器等拓扑对象。对应factory.setTopologyRecoveryEnabled()。automatic_recovery_enabled [boolean]设为true时启用连接恢复connection recovery当网络异常断开后客户端会自动尝试重连。对应factory.setAutomaticRecoveryEnabled()。use_correlation_id [boolean]是否启用 correlation id 机制用于在消息接收方为消息附加唯一 ID 以在ack 失败的场景下做去重。注意该参数语义主要面向 RabbitMQ Source 消费侧Sink 侧在源码中同样保留了解析入口见 RabbitmqConfig.java默认值为false。connection_timeout [int]TCP 建连超时时间单位为毫秒设为0表示无限等待。对应factory.setConnectionTimeout()。rabbitmq.config [map]除上述显式参数外用户还可以通过该 map 指定 RabbitMQ 客户端的其他非必填参数覆盖官方 RabbitMQ 配置文档中定义的所有客户端参数。在 RabbitmqConfig.java 的parseSinkOptionProperties中map 内所有 key 会被转为小写后存入sinkOptionProps后续传递给客户端工厂使用。例如rabbitmq.config { requested-heartbeat 10 connection-timeout 10 }common optionsSink 插件公共参数详见 Sink Common Options。其中最有价值的是source_table_name当作业中存在多个 Source/Transform/Sink 时用它显式指定当前 Sink 处理哪一张上游结果表配合上游的result_table_name使用。源码中还可用的其他参数原文档参数表之外RabbitmqConfig.java 中还定义了以下可解析参数供进阶场景使用名称类型说明routing_keystring发布消息时使用的 routing key为空时消息直接发往默认交换器direct 到队列exchangestring发布消息时使用的交换器名称默认默认交换器requested_channel_maxint初始请求的最大 channel 数量对应factory.setRequestedChannelMax()requested_frame_maxint请求的最大帧大小对应factory.setRequestedFrameMax()requested_heartbeatint请求的心跳超时时间对应factory.setRequestedHeartbeat()prefetch_countlong未 ack 时最多接收的消息条数QoS在连接初始化时通过channel.basicQos(prefetchCount, true)设置delivery_timeoutint投递最大等待时间for_e2e_testingboolean用于标识 E2E 测试模式这些参数虽未出现在文档参数表中但源码解析与客户端装配均已就绪RabbitmqClient.java 对每个非空字段逐一调用对应 setter使用时请以当前仓库源码版本为准。配置示例示例一最简配置原文档给出的 simple 示例sink { RabbitMQ { host rabbitmq-e2e port 5672 virtual_host / username guest password guest queue_name test1 rabbitmq.config { requested-heartbeat 10 connection-timeout 10 } } }这是最常见的写法host/port/virtual_host/username/password/queue_name六个必填字段齐全rabbitmq.config内补充心跳与连接超时等客户端级调优参数。示例二使用 url 连接当 RabbitMQ 提供了 AMQP URI例如amqp://guest:guestlocalhost:5672/%2F时可写成sink { RabbitMQ { url amqp://guest:guestlocalhost:5672/%2F host localhost port 5672 virtual_host / username guest password guest queue_name test1 } }示例三通过交换器与 routing key 发布默认routing_key为空时源码走channel.basicPublish(, queue_name, null, msg)即通过默认交换器把消息直接投递到指定队列当配置了routing_key后会改走channel.basicPublish(exchange, routing_key, false, false, null, msg)的交换器路由模式RabbitmqClient.javasink { RabbitMQ { host rabbitmq-e2e port 5672 virtual_host / username guest password guest queue_name test1 exchange my_exchange routing_key my_routing_key } }示例四完整端到端作业来自 E2E 测试仓库自带的 RabbitMQ 端到端测试配置 rabbitmq-to-rabbitmq.conf 展示了带 schema 的完整用法——RabbitMQ Source 读取队列testSink 写入队列test1env { parallelism 1 job.mode STREAMING } source { RabbitMQ { host rabbitmq-e2e port 5672 virtual_host / username guest password guest queue_name test for_e2e_testing true schema { fields { id bigint c_map mapstring, smallint c_array arraytinyint c_string string c_boolean boolean c_tinyint tinyint c_smallint smallint c_int int c_bigint bigint c_float float c_double double c_decimal decimal(2, 1) c_bytes bytes c_date date c_timestamp timestamp } } } } transform { } sink { RabbitMQ { host rabbitmq-e2e port 5672 virtual_host / username guest password guest queue_name test1 } }其中schema.fields展示了 SeaTunnel 支持的主要数据类型整数族、浮点族、字符串、布尔、decimal、bytes、date、timestamp、map、arraySink 会依据该 schema 完成 JSON 序列化。消息写入链路从 SeaTunnelRow 到 AMQP 消息理解底层调用链有助于排查消息格式不对消息没进队列等问题。以源码为准写入链路如下插件装配作业启动时SPI 机制通过AutoService(SeaTunnelSink.class)加载 RabbitmqSink.java其prepare()校验必填配置并构建RabbitmqConfigcreateWriter()产出RabbitmqSinkWriter。客户端初始化RabbitmqClient.java 构造时依次完成构建ConnectionFactory根据是否配置url决定走 URI 还是 host/port/vhost/user/password 分支→ 建立Connection→ 创建Channel→ 设置prefetch_countQoS → 声明持久化队列。逐行序列化RabbitmqSinkWriter.java 对每一行调用JsonSerializationSchema.serialize(element)将SeaTunnelRow按 schema 编码为 JSON 字节数组。发布消息rabbitMQClient.write(bytes)根据是否配置routing_key选择默认交换器直投队列或交换器 routing key 路由两种basicPublish方式发送。发送异常时若logFailuresOnly为 true 则仅记录错误日志否则抛出错误码RABBITMQ-04send messages failed终止任务。资源回收close()依次关闭 Channel 与 Connection任一关闭失败都会抛出RABBITMQ-03close connection failed。另外值得注意RabbitmqSinkWriter的prepareCommit()返回Optional.empty()说明该 Sink 不参与两阶段提交与不支持 exactly-once的特性声明一致。错误码速查连接器定义了独立的错误码枚举见 RabbitmqConnectorErrorCode.java错误码描述RABBITMQ-01处理队列消费者关闭信号失败RABBITMQ-02创建 RabbitMQ 客户端失败RABBITMQ-03关闭连接失败RABBITMQ-04发送消息失败RABBITMQ-05创建检查点期间消息无法 ackRABBITMQ-06消息无法通过 basicReject 拒绝RABBITMQ-07解析 URI 失败RABBITMQ-08初始化 SSL 上下文失败RABBITMQ-09设置 SSL 工厂失败作业日志中出现上述错误码时可据此快速定位是建连、发送、URI 解析还是 SSL 环节出了问题。常见问题与使用建议消息未写入预期队列确认queue_name拼写、vhost 是否正确默认写路径使用默认交换器直投队列若消息被路由走应检查是否误配了routing_key/exchange。连接频繁断开可显式设置automatic_recovery_enabled true并配合network_recovery_interval控制重连间隔同时通过requested-heartbeat保持心跳。消息格式Sink 固定输出 JSON 文本消息下游消费方应按 JSON 解析字段类型与schema声明不一致可能导致序列化异常。重复投递当前连接器不提供 exactly-once 保证关键业务建议在下游按use_correlation_id或业务主键做去重。并发写入每个 Sink 任务使用独立 Channel 发布可通过parallelism控制写入并发度但需留意目标队列的吞吐承受能力。版本变更记录原文档 Changelog 记录如下next version新增 RabbitMQ Sink 连接器将连接器自定义配置前缀改为 map 形式即rabbitmq.config的引入对应 PR 3719。这意味着rabbitmq.config这种 map 风格的扩展参数写法是较新版本的行为升级插件后请使用 map 写法传递自定义客户端参数。延伸阅读连接器官方文档docs/en/connector-v2/sink/Rabbitmq.mdSink 公共参数说明docs/en/connector-v2/sink/common-options.md连接器能力特性矩阵docs/en/concept/connector-v2-features.md配置解析实现RabbitmqConfig.java客户端与发布实现RabbitmqClient.java插件工厂与参数规则RabbitmqSinkFactory.javaSink 主类与参数校验RabbitmqSink.java端到端测试配置rabbitmq-to-rabbitmq.conf插件映射关系plugin-mapping.properties赞分享数据工程大数据批处理流处理【免费下载链接】seatunnelSeaTunnel is a next-generation super high-performance, distributed, massive data integration tool.项目地址https://gitcode.com/gh_mirrors/sea/seatunnel点击查看免费下载相关推荐SeaTunnel RabbitMQ Sink 连接器完整指南参数详解、队列声明机制与消息写入实战SeaTunnel RabbitMQ Sink 连接器完整指南参数详解、队列声明机制与消息写入实战 本文基于 SeaTunnel 仓库中 docs/zh/co数据集成ETL大数据批处理流处理变更数据捕获SeaTunnel RabbitMQ Sink 连接器实战指南配置详解、消息发布与故障恢复SeaTunnel RabbitMQ Sink 连接器实战指南配置详解、消息发布与故障恢复 RabbitMQ 是业界广泛使用的开源消息中间件而 SeaTun数据集成ETL大数据批处理流处理变更数据捕获SeaTunnel ActiveMQ Sink 连接器完全指南配置、JSON 消息序列化机制与源码级解析SeaTunnel ActiveMQ Sink 连接器完全指南配置、JSON 消息序列化机制与源码级解析 本文基于 SeaTunnel 仓库中的 Active数据集成ETL大数据批处理流处理变更数据捕获上一篇FastRTC实时通信质量保障从音频处理到连接管理的全链路测试下一篇如何使用Claude Code Hooks Mastery实现自动化部署流程创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
