跳到主要内容
版本：Next

RabbitMQ

RabbitMQ Sink 连接器

引擎支持​

Spark
Flink
SeaTunnel Zeta

描述​

用于将数据写入 RabbitMQ 队列。

主要特性​

接收器选项​

名称类型是否必须默认值
hoststring是-
portint是-
virtual_hoststring是-
usernamestring否-
passwordstring否-
queue_namestring是-
formatstring否json
protobuf_schemastring否-
protobuf_message_namestring否-
urlstring否-
uristring否-
sslboolean否false
routing_keystring否-
exchangestring否-
network_recovery_intervalint否-
topology_recovery_enabledboolean否-
AUTOMATIC_RECOVERY_ENABLEDboolean否-
connection_timeoutint否-
rabbitmq.configmap否-
durableboolean否true
exclusiveboolean否false
auto_deleteboolean否false
passiveboolean否false
common-options否-

host [string]​

RabbitMQ 服务器地址

port [int]​

RabbitMQ 服务器端口

virtual_host [string]​

virtual host，连接 broker 使用的 vhost

username [string]​

连接 broker 时使用的用户名

password [string]​

连接 broker 时使用的密码

username 和 password 需要一起配置。

url [string]​

设置host、port、username、password和virtual host的简便方式。

uri [string]​

url 的兼容别名。url 和 uri 只能配置一个。

ssl [boolean]​

使用 host 和 port 配置连接时启用 SSL/TLS。若 URI 本身提供连接信息，请使用 amqps:// 开头的 url。

当 url 使用 amqps:// 时，将按 JVM 信任库校验 Broker 证书并启用主机名校验。此前依赖隐式信任所有证书、使用自签名或私有 CA 证书的连接，需要将 Broker 证书导入信任库，否则将无法建立连接。

queue_name [string]​

数据写入的队列名。如果没有配置 routing_key，连接器会通过默认 exchange 将消息直接写入该队列。

format [string]​

消息体格式，支持 json 和 protobuf，默认值为 json。

protobuf_schema [string]​

当 format 为 protobuf 时生效，定义用于序列化 RabbitMQ 消息体的 Protobuf Schema。

protobuf_message_name [string]​

当 format 为 protobuf 时生效，指定要序列化的 Protobuf Message 名称。

routing_key [string]​

发布消息时使用的路由键。如果希望通过指定 exchange 发布消息，而不是直接写入 queue_name，请同时配置 routing_key 和 exchange。

exchange [string]​

配置 routing_key 时使用的 exchange。

durable [boolean]​

  • true：队列将在服务器重启时保留。
  • false：队列将在服务器重启时删除。

exclusive [boolean]​

  • true：队列仅由当前连接使用，连接关闭时将删除。
  • false：队列可以由多个连接使用。

auto_delete [boolean]​

  • true：队列将在最后一个消费者取消订阅时自动删除。
  • false：队列不会自动删除。

passive [boolean]​

  • false：按已配置的 durable、exclusive 和 auto_delete 参数声明队列。
  • true：只校验队列已存在，不创建或修改队列。适用于可发布但没有队列声明权限的账号。

network_recovery_interval [int]​

自动恢复需等待多长时间才尝试重连，单位为毫秒。

topology_recovery_enabled [boolean]​

设置为true，表示启用拓扑恢复。

AUTOMATIC_RECOVERY_ENABLED [boolean]​

设置为 true，表示启用连接恢复。

当前连接器配置项名称使用大写形式。请写成 AUTOMATIC_RECOVERY_ENABLED，不要写成 automatic_recovery_enabled。

connection_timeout [int]​

TCP连接建立的超时时间，单位为毫秒；0代表不限制。

rabbitmq.config [map]​

除了上面提及必须设置的RabbitMQ客户端参数，你也还可以为客户端指定多个非强制参数，参见 RabbitMQ官方文档参数设置。

common options​

Sink插件常用参数，请参考Sink常用选项获取更多细节信息。

配置说明​

  • 如果配置了 username，也必须配置 password，反过来也一样。
  • url 和 uri 只能配置一个。uri 为兼容已有配置保留，新配置请使用 url。
  • 使用 host 和 port 连接 AMQPS 端点时，请设置 ssl = true。
  • host、port、virtual_host 和 queue_name 是连接器必填项。url 可额外提供 RabbitMQ 客户端使用的 AMQP URI。
  • durable、exclusive 和 auto_delete 用于连接器声明目标队列。
  • 当 format 为 protobuf 时，需要同时配置 protobuf_schema 和 protobuf_message_name。

示例​

写入队列​

env {
    parallelism = 1
    job.mode = "STREAMING"
}

source {
    FakeSource {
        row.num = 10
        schema = {
            fields {
                id = bigint
                c_string = string
            }
        }
    }
}

sink {
      RabbitMQ {
          host = "rabbitmq-e2e"
          port = 5672
          virtual_host = "/"
          username = "guest"
          password = "guest"
          queue_name = "test1"
          rabbitmq.config = {
            requested-heartbeat = 10
            connection-timeout = 10
          }
      }
}

示例 2​

配置队列的 durable、exclusive、auto_delete：

env {
    parallelism = 1
    job.mode = "STREAMING"
}

source {
    FakeSource {
        row.num = 10
        schema = {
            fields {
                id = bigint
                c_string = string
            }
        }
    }
}

sink {
      RabbitMQ {
          host = "rabbitmq-e2e"
          port = 5672
          virtual_host = "/"
          username = "guest"
          password = "guest"
          queue_name = "test1"
          durable = true
          exclusive = false
          auto_delete = false
          rabbitmq.config = {
            requested-heartbeat = 10
            connection-timeout = 10
          }
      }
}

写入 Protobuf 消息到队列​

sink {
      RabbitMQ {
          host = "rabbitmq-e2e"
          port = 5672
          virtual_host = "/"
          queue_name = "protobuf_queue"
          format = protobuf
          protobuf_message_name = Person
          protobuf_schema = """
              syntax = "proto3";
              message Person {
                int64 id = 1;
                string name = 2;
              }
          """
      }
}

常见问题​

RabbitMQ Sink 支持路由到指定的 Exchange 和 Routing Key 吗？​

支持。Sink 会根据配置的 queue_name 及路由参数将消息发布到 RabbitMQ 目标队列或路由规则中。

RabbitMQ Sink 如何处理网络重连和超时？​

可以通过 rabbitmq.config 配置块调优客户端连接参数（如 connection-timeout、requested-heartbeat 等），以应对网络短暂抖动并提高连接稳定性。

变更日志​

Change Log
ChangeCommitVersion
[Fix][connector-rabbitmq] Set default value for durable, exclusive and auto-delete (#9631)https://github.com/apache/seatunnel/commit/5f9492e62a2.3.12
[Feature][Checkpoint] Add check script for source/sink state class serialVersionUID missing (#9118)https://github.com/apache/seatunnel/commit/4f5adeb1c72.3.11
[Improve] rabbit mq options (#8740)https://github.com/apache/seatunnel/commit/4eec9be0122.3.10
[Improve] restruct connector common options (#8634)https://github.com/apache/seatunnel/commit/f3499a6eeb2.3.10
[Improve][dist]add shade check rule (#8136)https://github.com/apache/seatunnel/commit/51ef8000162.3.9
[Feature][Restapi] Allow metrics information to be associated to logical plan nodes (#7786)https://github.com/apache/seatunnel/commit/6b7c53d03c2.3.9
[Feature][Rabbitmq] Allow configuration of queue durability and deletion policy (#7365)https://github.com/apache/seatunnel/commit/aabfc8eb782.3.8
[Hotfix][connector-v2-rabbit] fix rabbit checkpoint exception in Flink mode (#7108)https://github.com/apache/seatunnel/commit/423a7b142b2.3.6
[Feature][Kafka] Support multi-table source read (#5992)https://github.com/apache/seatunnel/commit/60104602d12.3.6
[Improve] Remove use SeaTunnelSink::getConsumedType method and mark it as deprecated (#5755)https://github.com/apache/seatunnel/commit/8de74081002.3.4
Support config column/primaryKey/constraintKey in schema (#5564)https://github.com/apache/seatunnel/commit/eac76b4e502.3.4
[Bugfix][connector-v2][rabbitmq] Fix reduplicate ack msg bug and code style (#4842)https://github.com/apache/seatunnel/commit/985fb6642a2.3.2
[Hotfix][E2E] Fix RabbitmqIT (#4593)https://github.com/apache/seatunnel/commit/9bd5403d712.3.2
Merge branch 'dev' into merge/cdchttps://github.com/apache/seatunnel/commit/4324ee19122.3.1
[Improve][Project] Code format with spotless plugin.https://github.com/apache/seatunnel/commit/423b5830382.3.1
[improve][api] Refactoring schema parse (#4157)https://github.com/apache/seatunnel/commit/b2f573a13e2.3.1
[Improve][build] Give the maven module a human readable name (#4114)https://github.com/apache/seatunnel/commit/d7cd6010512.3.1
[Improve][Project] Code format with spotless plugin. (#4101)https://github.com/apache/seatunnel/commit/a2ab1665612.3.1
[Feature][Connector] add get source method to all source connector (#3846)https://github.com/apache/seatunnel/commit/417178fb842.3.1
[Improve][Connector-V2] Change Connector Custom Config Prefix To Map (#3719)https://github.com/apache/seatunnel/commit/ef1b8b1bb52.3.1
[Feature][API & Connector & Doc] add parallelism and column projection interface (#3829)https://github.com/apache/seatunnel/commit/b9164b8ba12.3.1
[Hotfix][OptionRule] Fix option rule about all connectors (#3592)https://github.com/apache/seatunnel/commit/226dc6a1192.3.0
[Feature][Connector-V2][RabbitMQ] Add RabbitMQ source & sink connector (#3312)https://github.com/apache/seatunnel/commit/4b12691a8d2.3.0