PublishAMQP

文档中心 › 组件索引 › PublishAMQP

PublishAMQP Processor

类型全限定名：cn.org.acal.lynxus.amqp.processors.PublishAMQP

所属 Bundle：cn.org.acal.lynxus:lynxus-amqp-nar:2.11.8

描述

从 FlowFile 的内容创建 AMQP 消息并将消息发送到 AMQP 交换器。

标签

amqprabbitputmessagesendpublish

属性

下表中，加粗的属性为必填项；其余为可选项。表格同时标注默认值、是否支持表达式语言，以及是否为敏感属性（值会被加密）。

名称默认值可选值描述
Exchange Name消息将发送到的 AMQP 交换机的名称。通常由 AMQP 管理员提供（例如「amq.direct」）。这是一个可选属性。如果留空，消息将发送到默认 AMQP 交换机。
支持表达式语言：是（使用 FlowFile 属性与环境/系统变量求值）
Routing KeyAMQP 用于将消息从交换机路由到目标队列的路由键名称。通常由管理员提供（例如「myKey」）。当消息发送到默认交换机时，此属性对应于目标队列名称，否则必须设置从交换机到队列通过路由键的绑定（通常由 AMQP 管理员设置）
支持表达式语言：是（使用 FlowFile 属性与环境/系统变量求值）
Delivery GuaranteeAT_MOST_ONCE
  • At most once — 即发即弃（Fire-and-forget）：发送消息时不等待 broker 确认。 无法传递的消息（无匹配的队列绑定）会记录为警告， FlowFile 被路由到 'success'。提供最大吞吐量。
  • At least once — 启用发布者确认（Publisher Confirms）：处理器会阻塞，直到 broker 确认 消息（ack 或 nack）。无法传递的消息会被可靠地检测并路由 到 'failure'。以更高的延迟为代价防止静默数据丢失，尤其是 使用远程 broker 时。
控制处理器在路由 FlowFile 之前是否等待发布确认（代理确认/否定确认）。「至少一次」启用 RabbitMQ 发布者确认：处理器阻塞直到代理确认消息，不可传递的消息（无匹配队列绑定）被可靠地路由到「failure」。这以防止静默数据丢失为代价，但会显著增加延迟，特别是对于远程代理。「最多一次」使用原始的即发即忘模式：消息发送后不等待确认。不可传递的消息仅作为警告记录，FlowFile 仍被路由到「success」。此模式提供最大吞吐量但不提供传递保证。
Headers SourceAMQP_HEADERS_ATTRIBUTE
  • FlowFile Attributes — 根据正则表达式模式选择 FlowFile 属性作为事件头。匹配属性的键将用作头键
  • AMQP Headers Attribute — 从 'amqp$headers' 属性字符串准备头
将应用于已发布消息的标头的来源。
Headers Pattern将针对 FlowFile 属性评估的正则表达式，用于选择匹配的属性并作为 AMQP 头。属性名将用作头键。
支持表达式语言：是（仅使用环境/系统变量求值）
Header Separator,用于拆分头的键值的字符。值必须仅为一个字符。否则您将收到错误消息
Brokers已知 AMQP 代理的逗号分隔列表，格式为 <host>:<port>（例如 localhost:5672）。如果设置了此项，主机名和端口将被忽略。仅包含来自同一 AMQP 集群的主机。
支持表达式语言：是（仅使用环境/系统变量求值）
Host NamelocalhostAMQP 代理的网络地址（例如本地主机）。 如果设置了 Brokers，则忽略此属性。
支持表达式语言：是（仅使用环境/系统变量求值）
Port5672标识 AMQP 代理端口的数值（例如 5671）。 如果设置了 Brokers，则忽略此属性。
支持表达式语言：是（仅使用环境/系统变量求值）
Virtual Host隔离 AMQP 系统以增强安全性的虚拟主机名。
支持表达式语言：是（仅使用环境/系统变量求值）
Username用于身份验证和授权的用户名。
支持表达式语言：是（仅使用环境/系统变量求值）
Password密码用于认证和授权。
敏感属性：是
AMQP Version0.9.1
  • 0.9.1
AMQP 版本。 目前仅支持AMQP v0.9.1。
SSL Context Service控制器服务 API：
SSLContextProvider
cn.org.acal.lynxus-lynxus-standard-services-api-nar-2.11.8		SSL 上下文服务用于为 TLS/SSL 连接提供客户端证书信息。
Client Certificate Authentication Enabledfalse
  • true
  • false
使用 SSL 证书而不是用户名/密码进行身份验证。

关系 (Relationships)

名称描述自动终止
failure所有无法路由到 AMQP 目标的 FlowFile 都将路由到此关系否
success发送到 AMQP 目标的所有 FlowFile 都会路由到此关系否

读取的 FlowFile 属性

名称描述
amqp$appId要在 AMQP 消息上设置的 App ID 字段
amqp$contentEncoding要在 AMQP 消息上设置的内容编码
amqp$contentType要在 AMQP 消息上设置的内容类型
amqp$headers如果 'Header Source' 设置为使用它，则要在 AMQP 消息上设置的头。请参阅处理器的附加详细信息。
amqp$deliveryMode消息传递模式的数值指示器
amqp$priority消息优先级
amqp$correlationId消息的关联 ID
amqp$replyTo消息的回复字段（Reply-To）的值
amqp$expiration消息过期时间
amqp$messageId消息的唯一 ID
amqp$timestamp消息的时间戳，以自纪元以来的毫秒数表示
amqp$type消息类型
amqp$userId用户的 ID
amqp$clusterIdAMQP 集群的 ID

状态管理

该组件不存储状态。

受限 (Restricted)

该组件不受限。

输入要求

该组件要求有传入连接。

系统资源考量

资源描述
MEMORYAn instance of this component can cause high usage of this system resource. Multiple instances or high concurrency settings may result a degradation of performance.

详细说明

PublishAMQP

概述

该 Processor 把传入 FlowFile 的内容发布到基于 AMQP 的消息系统。编写本文档时，支持的 AMQP 协议版本为 v0.9.1。

该组件基于 RabbitMQ Client API， 以下指南和教程 也可以帮你复习一些 AMQP 基础知识。

该 Processor 做两件事。它通过提取 FlowFile 的内容（消息体和 Attribute）来构建 AMQP Message。消息构建完成后，会被发送到 AMQP Exchange。AMQP Properties 会从 FlowFile 中提取，并转换为 com.rabbitmq.client.AMQP.BasicProperties 与消息一起发送。成功后，传入 FlowFile 会转移到 success Relationship；失败时，FlowFile 会被施加惩罚（penalize）并转移到 failure Relationship。

我的消息去哪了？

在典型的 AMQP exchange 模型中，发送到 AMQP Exchange 的消息会基于 Routing Key 被路由到 队列 中的最终目的地。这称为绑定（Binding）。如果由于某些配置错误， _Exchange、Routing Key 和队列_之间的绑定没有建立，消息将没有最终目的地， 会被退回（即数据不会到达队列）。如果发生这种情况，你会在 app-log 和 bulletin 中看到相应的日志。修复绑定（通常由 AMQP 管理员完成）即可解决问题。

AMQP 属性

从 FlowFile 提取的 Attribute，如果名称带 amqp$ 前缀（例如 amqp$contentType=text/xml），就被视为 AMQP property 的候选。要给消息添加额外的 AMQP property，可以在源 Processor 和 PublishAMQP Processor 之间使用 UpdateAttribute Processor。以下是可用的标准 AMQP property 列表： ("amqp$contentType", "amqp$contentEncoding", "amqp$headers"（当 'Headers Source' 设为 'Attribute "amqp$headers" Value' 时）、"amqp$deliveryMode", "amqp$priority", "amqp$correlationId", "amqp$replyTo", "amqp$expiration", "amqp$messageId", "amqp$timestamp", "amqp$type", "amqp$userId", "amqp$appId", "amqp$clusterId")

AMQP 消息 Header 来源

Processor 附加到 AMQP 消息的 header 取决于 "Headers Source" 属性的值。

  1. Attribute "amqp$headers" Value - Processor 会读取单个 Attribute "amqp$headers"，按 "Header Separator" 拆分，然后按 key=value 格式读取 header。
  2. Attributes Matching Regex - Processor 会通过匹配 "Attributes To Headers Regular Expression" 提供的正则来选取 FlowFile Attribute。Attribute 的名称用作 header 的键。

配置详情

编写本文档时，只定义了适用于大多数场景的核心配置属性。 其他属性会随着该组件的发展逐步定义。配置 PublishAMQP：

  1. Exchange Name - [可选] 消息将发送到的 AMQP exchange 名称。通常由管理员提供（例如 'amq.direct'）。这是一个可选属性。如果留空，消息会发送到默认的 AMQP exchange。关于 AMQP Exchange 的更多信息参见 这里。
  2. Routing Key - [必需] AMQP 用于把消息从 exchange 路由到目标队列的路由键名称。通常由管理员提供（例如 'myKey'）。当消息发送到默认 exchange 时，该属性对应目标队列名称；否则必须设置从 Exchange 经 Routing Key 到队列的绑定（通常由 AMQP 管理员完成）。关于 AMQP Exchange 和 Binding 的更多信息参见 这里。
  3. Host Name - [必需] 运行 AMQP 代理的主机名。通常由管理员提供（例如 'myhost.com'）。默认为 'localhost'。
  4. Port - [必需] 运行 AMQP 代理的端口号。通常由管理员提供（例如 '2453'）。默认为 '5672'。
  5. User Name - [必需] 连接 AMQP 代理的用户名。通常由管理员提供（例如 'me'）。默认为 'guest'。
  6. Password - [必需] 与用户名配合使用的密码。通常由管理员提供。 默认为 'guest'。
  7. Use Certificate Authentication - [可选] 使用 SSL 证书的通用名称进行认证，而不是用户名/密码。只能与 SSL 配合使用。默认为 'false'。
  8. Virtual Host - [可选] 用于隔离 AMQP 系统以增强安全性的虚拟主机名。关于虚拟主机的更多细节参见 这篇博客。
© 2026 ACAL Organization. Licensed under Apache License 2.0.