跳到主要内容
版本：Next

RabbitMQ

RabbitMQ 源连接器

引擎支持​

Spark
Flink
SeaTunnel Zeta

描述​

用于从 RabbitMQ 队列读取数据。

主要特性​

提示

为了实现精确一次，源必须是非并行的（并行度设置为 1）。这个限制主要是由于 RabbitMQ 从单个队列向多个消费者分派消息的方式。

选项​

参数名类型必须默认值描述
hoststring是-连接的默认主机
portint是-连接的默认端口
virtual_hoststring否-虚拟主机 – 连接到代理时使用的虚拟主机
usernamestring否-连接到代理时使用的 AMQP 用户名
passwordstring否-连接到代理时使用的密码
queue_namestring否-要消费消息的队列
schemaconfig否-上游数据的模式。更多详情请参考 Schema 特性。
tables_configsarray否-用于同时从多个队列读取消息。数组中的每个对象必须包含 queue_name 和 schema。
formatstring否json消息体格式，支持 json 和 protobuf
protobuf_schemastring否-当 format 为 protobuf 时生效，用于解析消息体的 Protobuf Schema
protobuf_message_namestring否-当 format 为 protobuf 时生效，指定要解析的 Protobuf Message 名称
urlstring否-便捷方法，用于设置 AMQP URI 中的字段：主机、端口、用户名、密码和虚拟主机
uristring否-url 的兼容别名
sslboolean否false使用 host 和 port 连接时是否启用 SSL/TLS
routing_keystring否-RabbitMQ 共享配置中的可选路由键
exchangestring否-RabbitMQ 共享配置中的可选 exchange
network_recovery_intervalint否-自动恢复在尝试重新连接之前等待多长时间（毫秒）
topology_recovery_enabledboolean否-如果为 true，启用拓扑恢复
AUTOMATIC_RECOVERY_ENABLEDboolean否-如果为 true，启用连接恢复
connection_timeoutint否-连接 tcp 建立超时（毫秒）；零表示无限
requested_channel_maxint否-最初请求的最大通道数；零表示无限制。注意：值必须在 0 到 65535 之间（AMQP 0-9-1 中的无符号短整数）。
requested_frame_maxint否-请求的最大帧大小
requested_heartbeatint否-设置请求的心跳超时。注意：值必须在 0 到 65535 之间（AMQP 0-9-1 中的无符号短整数）。
prefetch_countint否-预取计数，无需确认即可接收的最大消息数
delivery_timeoutint否-交付超时，等待下一条消息交付的最大时间（毫秒）
use_correlation_idboolean否-消息是否带有可用于去重的唯一 correlation id
durableboolean否true队列是否在服务器重启时保留
exclusiveboolean否false队列是否仅由当前连接使用
auto_deleteboolean否false队列是否在最后一个消费者取消订阅时自动删除
passiveboolean否false是否只校验已有队列而不声明或创建队列
common-options否-源插件通用参数

host [string]​

连接的默认主机

port [int]​

连接的默认端口

virtual_host [string]​

虚拟主机 – 连接到代理时使用的虚拟主机

username [string]​

连接到代理时使用的 AMQP 用户名

password [string]​

连接到代理时使用的密码

username 和 password 需要一起配置。

url [string]​

便捷方法，用于设置 AMQP URI 中的字段：主机、端口、用户名、密码和虚拟主机

uri [string]​

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

ssl [boolean]​

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

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

queue_name [string]​

要消费消息的队列。注意：如果未配置 tables_configs，则为必填项。

routing_key [string]​

RabbitMQ 共享配置中的可选路由键。普通队列消费不需要配置它。

exchange [string]​

RabbitMQ 共享配置中的可选 exchange。普通队列消费不需要配置它。

schema [Config]​

fields [Config]​

上游数据的模式字段。更多详情请参考 Schema 特性。注意：如果未配置 tables_configs，则为必填项。

tables_configs [array]​

用于同时从多个队列读取消息。数组中的每个对象必须包含 queue_name 和 schema。

format [string]​

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

protobuf_schema [string]​

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

protobuf_message_name [string]​

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

network_recovery_interval [int]​

自动恢复在尝试重新连接之前等待多长时间（毫秒）

topology_recovery_enabled [boolean]​

如果为 true，启用拓扑恢复

AUTOMATIC_RECOVERY_ENABLED [boolean]​

如果为 true，启用连接恢复。

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

connection_timeout [int]​

连接 tcp 建立超时（毫秒）；零表示无限

requested_channel_max [int]​

最初请求的最大通道数；零表示无限制。注意：值必须在 0 到 65535 之间（AMQP 0-9-1 中的无符号短整数）。

requested_frame_max [int]​

请求的最大帧大小

requested_heartbeat [int]​

设置请求的心跳超时。注意：值必须在 0 到 65535 之间（AMQP 0-9-1 中的无符号短整数）。

prefetch_count [int]​

预取计数，无需确认即可接收的最大消息数

delivery_timeout [int]​

交付超时，等待下一条消息交付的最大时间（毫秒）

use_correlation_id [boolean]​

消费到的消息是否带有唯一 correlation id。开启后，当确认消息失败时，可用这个 id 辅助去重。

common options​

源插件通用参数，详情请参考 源通用选项。

durable​

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

exclusive​

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

auto_delete​

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

passive​

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

迁移指南与配置规则​

如果您从仅支持单表读取的早期版本升级，您现有的配置无需任何更改即可正常工作。

配置优先级：

  • 不能同时配置 tables_configs 和根级别的 queue_name。它们是互斥的，同时配置会导致校验失败。
  • 使用 tables_configs 进行多表模式。
  • 使用根级别的 queue_name 和 schema 进行单队列模式。
  • 多表模式下，每个队列自己的 schema 应放在对应的 tables_configs 条目里。
  • 当 format 为 protobuf 时，需要在队列配置所在层级同时配置 protobuf_schema 和 protobuf_message_name。
  • 如果配置了 username，也必须配置 password，反过来也一样。
  • url 和 uri 只能配置一个。uri 为兼容已有配置保留，新配置请使用 url。
  • 使用 host 和 port 连接 AMQPS 端点时，请设置 ssl = true。
  • host 和 port 总是必填。virtual_host 是可选项，除非您的 RabbitMQ 环境要求使用非默认虚拟主机。

示例​

单表读取示例​

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

source {
    RabbitMQ {
        host = "rabbitmq-e2e"
        port = 5672
        virtual_host = "/"
        username = "guest"
        password = "guest"
        queue_name = "test"
        durable = true
        exclusive = false
        auto_delete = false
        schema = {
            fields {
                id = bigint
                c_map = "map<string, smallint>"
                c_array = "array<tinyint>"
                c_string = string
                c_boolean = boolean
            }
        }
    }
}

sink {
    Console {}
}

多表读取示例​

您可以使用 tables_configs 选项在一个作业中同时从多个 RabbitMQ 队列消费消息。连接器将根据消息来源的队列自动为每行数据分配正确的表标识符，允许您使用 plugin_input 将它们路由到不同的 sink。

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

source {
  RabbitMQ {
    host = "rabbitmq-e2e"
    port = 5672
    virtual_host = "/"
    username = "guest"
    password = "guest"

    # 使用 tables_configs 从多个队列中读取
    tables_configs = [
      {
        queue_name = "users_queue"
        schema = {
          table = "users_table" # 定义用于路由的表名
          fields {
            user_id = bigint
            name = string
          }
        }
      },
      {
        queue_name = "orders_queue"
        schema = {
          table = "orders_table" # 定义用于路由的表名
          fields {
            order_id = bigint
            amount = double
          }
        }
      }
    ]
  }
}

sink {
  # 第一个 sink 将仅接收 users_table 的数据
  Console {
    plugin_input = "users_table"
  }

  # 第二个 sink 将仅接收 orders_table 的数据
  Console {
    plugin_input = "orders_table"
  }
}

Protobuf 读取示例​

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

常见问题​

为什么实现精确一次必须将并行度设置为 1？​

RabbitMQ 会在同一个队列的多个活跃消费者之间以轮询方式分发消息。当多个并行 Reader 同时消费同一个队列时，无法保证消息顺序以及分布式 Worker 之间确定性的 offset/ack 协同。因此，必须将并行度设置为 1 才能实现精确一次。

RabbitMQ Source 支持哪些消息格式？​

RabbitMQ Source 默认支持 JSON。当 format 设置为 protobuf 时支持 Protobuf。连接器会按照配置的 schema 将每条 RabbitMQ 消息体反序列化为一行 SeaTunnel 数据。

任务发生故障时未确认的消息如何处理？​

当 SeaTunnel 任务失败或异常退出时，与 RabbitMQ 的连接会断开，RabbitMQ 会自动将所有未确认（unacknowledged）的消息重新入队（requeue）。在任务从检查点恢复后，Reader 可以重新处理这些消息，避免数据丢失。

变更日志​

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 &amp; Connector &amp; 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