跳到主要内容
版本：Next

ActiveMQ

ActiveMQ Sink 连接器

支持的引擎​

Spark
Flink
SeaTunnel Zeta

描述​

用于把 SeaTunnel 数据写入 ActiveMQ 队列。每一行数据都会被序列化成一条 JSON 文本消息。 消费队列请参见独立的 ActiveMQ Source。

关键特性​

选项​

名称类型是否必传默认值描述
uristring是-ActiveMQ Broker 地址，例如 tcp://localhost:61616。
queue_namestring是-要写入的队列名称。
usernamestring否-创建 ActiveMQ 连接时使用的用户名。配置该项时必须同时配置 password。
passwordstring否-创建 ActiveMQ 连接时使用的密码。配置该项时必须同时配置 username。
client_idstring否-连接工厂使用的 JMS 客户端 ID。
check_for_duplicateboolean否-是否让 ActiveMQ 客户端检查重复消息。
always_session_asyncboolean否-是否始终为每个 Session 使用独立线程分发消息。
always_sync_sendboolean否-是否始终使用同步方式发送消息。
close_timeoutint否-关闭连接的超时时间，单位毫秒。
dispatch_asyncboolean否-Broker 是否异步分发消息。
nested_map_and_list_enabledboolean否-是否允许结构化消息属性和 MapMessage 条目中包含嵌套的 Map、List 对象。
warn_about_unstarted_connection_timeoutint否-连接没有正确启动时，ActiveMQ 客户端发出警告前等待的毫秒数。设置为小于 0 的值可以关闭这个警告。
consumer_expiry_check_enabledboolean否-是否在每个 MessageConsumer 分发消息前检查消息是否已经过期。
max_thread_pool_sizeint否1000ActiveMQ 连接工厂内部用于消息分发的线程池最大大小。默认值为 1000。
send_timeoutint否0ActiveMQ 连接工厂的发送超时时间，单位毫秒。值为 0 表示不超时。默认值为 0。
use_compressionboolean否false是否在发送消息前对消息体进行压缩。默认值为 false。
connect_response_timeoutint否0连接建立时等待 Broker 响应的超时时间，单位毫秒。值为 0 表示不超时。默认值为 0。
delivery_modeint否2JMS 投递模式：1 为非持久化，2 为持久化。默认值为 2（持久化）。
priorityint否4JMS 消息优先级，0（最低）到 9（最高）。默认值为 4。
producer_window_sizeint否0异步发送时的流控窗口大小，单位字节。值为 0 表示不限流。默认值为 0。
time_to_livelong否0消息存活时间，单位毫秒。值为 0 表示永不过期。默认值为 0。
use_async_sendboolean否false为 true 时发送不等 Broker 确认直接返回，提升吞吐量。默认值为 false。

注意事项​

  • uri 是连接入口，Broker 的主机和端口都写在这里，例如 tcp://activemq-host:61616。
  • username 和 password 是可选项；如果 Broker 需要认证，这两个配置要一起写。
  • 连接器会把每一行 SeaTunnel 数据作为一条 JSON 文本消息写入 queue_name，当前没有单独的 format 配置。
  • Broker 地址请使用 uri 配置，host 和 port 不是 ActiveMQ Sink 的配置项。
  • 该 Sink 前面可以接任意 SeaTunnel Source。ActiveMQ 连接器只负责把最终的数据行发送到队列。
  • 当 use_async_send = true 或 delivery_mode = 1（非持久化）时，ActiveMQ 客户端发送消息后不等 Broker 确认即返回，且本连接器未注册 ExceptionListener，因此发送失败可能不会通知到作业。仅在可接受消息丢失的场景下使用这些选项。
  • 工厂级选项（max_thread_pool_size、send_timeout、use_compression、connect_response_timeout、producer_window_size、use_async_send）仅在显式配置时生效。未配置时，使用 uri 中对应的 jms.* 参数（如果存在）或 ActiveMQ 客户端默认值。

示例​

把 FakeSource 数据写入 ActiveMQ 队列：

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

source {
  FakeSource {
    schema = {
      fields {
        id = int
        name = string
      }
    }
    rows = [
      { kind = INSERT, fields = [1, "Alice"] }
      { kind = INSERT, fields = [2, "Bob"] }
    ]
  }
}

sink {
  ActiveMQ {
    uri = "tcp://localhost:61616"
    username = "admin"
    password = "admin"
    queue_name = "testQueue"
  }
}

在流式模式下，Sink 会保持与 Broker 的连接持续打开，每来一行数据就写入一条。用户名 / 密码也可以直接 写在 uri 中，例如 tcp://admin:admin@localhost:61616：

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

source {
  FakeSource {
    schema = {
      fields {
        id = int
        name = string
      }
    }
    rows = [
      { kind = INSERT, fields = [1, "Alice"] }
    ]
  }
}

sink {
  ActiveMQ {
    uri = "tcp://admin:admin@localhost:61616"
    queue_name = "testQueue"
  }
}

FAQ​

ActiveMQ Sink 支持 topic 吗？​

不支持。当前 Sink 只会写入由 queue_name 指定的 JMS 队列，连接器工厂并不支持 topic 目的地。如果需要发布/订阅语义，请改用通用的 JMS 连接器或者其它专门面向 ActiveMQ topic 的桥接组件；要直接驱动本连接器，请继续用 queue。

username 和 password 是怎么校验的？​

这两项都是可选的。设置的时候必须同时设置（只设其中一个会让任务启动失败）。它们作用于 JMS 连接工厂这一层，因此会覆盖已经嵌在 uri 里的凭证。如果 Broker 需要鉴权，建议显式配置 username/password 而不是把它们写在 URL 里——这样凭证会出现在任务配置日志里，而不是埋在连接串里。

每条数据最终变成什么格式的消息？​

每行 SeaTunnel 数据会被序列化为一条 JSON 文本消息写到配置的 queue_name 里。连接器没有 format 配置项，JSON 结构由 Sink 序列化器固定，所以对端如果想要其它编码，需要自己先解码 JSON body。

支持精确一次写入吗？​

不支持。Sink 是尽力而为模式，重连行为由底层 JMS 客户端决定。如果可以接受来自上游的至少一次重放，请在任务级别启用 checkpoint。

变更日志​

变更日志
变更提交版本
[Feature][Connector-V2] Add checkpoint-aware ActiveMQ queue source-Next