ConsumeAMQP Processor
描述
使用 AMQP 0.9.1 协议使用来自 AMQP Broker 的 AMQP 消息。 从 AMQP Broker 接收到的每条消息都将作为其自己的 FlowFile 发送到“成功”关系。
标签
属性
| 名称 | 默认值 | 可选值 | 描述 |
|---|---|---|---|
| Queue | | 将从中消费消息的现有 AMQP 队列的名称。通常由 AMQP 管理员预定义。 | |
| Auto-Acknowledge Messages | false |
| 如果为 false（非自动确认），消息将在 FlowFile 转移到 success 并提交 Lynxus 会话后由处理器确认。非自动确认模式提供「至少一次」的传递语义。如果为 true（自动确认），传递到 AMQP 客户端的消息将在发送后立即由 AMQP 代理自动确认。这通常会提供更高的吞吐量，但也会导致在 AMQP 代理、Lynxus 或处理器重启/崩溃时消息丢失。自动确认模式提供「最多一次」的传递语义，仅在可以接受消息丢失时推荐使用。 |
| Batch Size | 10 | 在单个会话中应处理的最大消息数。一旦接收到这么多消息（或不再有更多消息可用），接收到的消息将被转移到「success」关系，并向 AMQP 代理确认消息。将此值设置为更大的数字可能会带来更好的性能，特别是对于非常小的消息，但也可能导致在 Lynxus 突然重启时更多消息被重复。 | |
| Prefetch Count | 0 | 消费者的最大未确认消息数。如果消费者有这么多未确认消息，AMQP 代理将不再发送新消息，直到消费者确认一些已传递的消息。允许值：0 到 65535。0 表示无限制 | |
| Header Output Format | Comma-Separated String |
| 定义如何从接收到的消息中输出标头 |
| Header Key Prefix | consume.amqp | 添加到 FlowFile 属性时要作为标头键前缀的文本。 处理器将附加“.” 到该房产的价值 | |
| Header Separator | , | 用于在字符串中分隔头的键值的字符。值必须仅为一个字符。 | |
| Remove Curly Braces | False |
| 如果 true 删除大括号，标题中的大括号将自动删除。 |
| Max Inbound Message Body Size | 64 MB | 入站（接收）消息的最大正文大小。 支持表达式语言：是（仅使用环境/系统变量求值） | |
| Brokers | | 已知 AMQP 代理的逗号分隔列表，格式为 <host>:<port>（例如 localhost:5672）。如果设置了此项，主机名和端口将被忽略。仅包含来自同一 AMQP 集群的主机。 支持表达式语言：是（仅使用环境/系统变量求值） | |
| Host Name | localhost | AMQP 代理的网络地址（例如本地主机）。 如果设置了 Brokers，则忽略此属性。 支持表达式语言：是（仅使用环境/系统变量求值） | |
| Port | 5672 | 标识 AMQP 代理端口的数值（例如 5671）。 如果设置了 Brokers，则忽略此属性。 支持表达式语言：是（仅使用环境/系统变量求值） | |
| Virtual Host | | 隔离 AMQP 系统以增强安全性的虚拟主机名。 支持表达式语言：是（仅使用环境/系统变量求值） | |
| Username | | 用于身份验证和授权的用户名。 支持表达式语言：是（仅使用环境/系统变量求值） | |
| Password | | 密码用于认证和授权。 敏感属性：是 | |
| AMQP Version | 0.9.1 |
| AMQP 版本。 目前仅支持AMQP v0.9.1。 |
| SSL Context Service | | 控制器服务 API：SSLContextProvidercn.org.acal.lynxus-lynxus-standard-services-api-nar-2.11.8 | SSL 上下文服务用于为 TLS/SSL 连接提供客户端证书信息。 |
| Client Certificate Authentication Enabled | false |
| 使用 SSL 证书而不是用户名/密码进行身份验证。 |
关系 (Relationships)
| 名称 | 描述 | 自动终止 |
|---|---|---|
success | 从 AMQP 队列接收的所有 FlowFile 都会路由到此关系 | 否 |
写入的 FlowFile 属性
| 名称 | 描述 |
|---|---|
amqp$appId | AMQP 消息中的应用程序 ID 字段 |
amqp$contentEncoding | AMQP消息报告的内容编码 |
amqp$contentType | AMQP消息报告的内容类型 |
amqp$headers | AMQP 消息上存在的标头。 仅当处理器配置为输出此属性时才添加。 |
<Header Key Prefix>.<attribute> | 如果处理器配置为将标头输出为属性，则每个消息标头都将插入此属性名称 |
amqp$deliveryMode | 消息传送模式的数字指示器 |
amqp$priority | 消息优先级 |
amqp$correlationId | 消息的关联 ID |
amqp$replyTo | 消息回复字段的值 |
amqp$expiration | 消息过期 |
amqp$messageId | 消息的唯一ID |
amqp$timestamp | 消息的时间戳，即自纪元以来的毫秒数 |
amqp$type | 消息类型 |
amqp$userId | 用户ID |
amqp$clusterId | AMQP集群的ID |
amqp$routingKey | AMQP消息的routingKey |
amqp$exchange | 接收 AMQP 消息的交换器 |
状态管理
该组件不存储状态。
受限 (Restricted)
该组件不受限。
输入要求
该组件不允许有传入连接。
详细说明
ConsumeAMQP
概述
该 Processor 从 AMQP 消息队列消费消息，并把它们转换为 FlowFile，路由到流程中的下一个组件。编写本文档时，支持的 AMQP 协议版本为 v0.9.1。
该组件基于 RabbitMQ Client API， 以下指南和教程 也可以帮你复习一些 AMQP 基础知识。
该 Processor 做两件事。它通过从消费的 AMQP 消息中提取 信息（消息体和 Attribute）来构建 FlowFile。消息被消费后， 构建一个 FlowFile。消息体写入 FlowFile，其 com.rabbitmq.client.AMQP.BasicProperties 作为 Attribute 转移到 FlowFile。AMQP Attribute 名称带 amqp$ 前缀。
AMQP 属性
以下是可能随消息一起出现的标准 AMQP 属性列表： ("amqp$contentType", "amqp$contentEncoding", "amqp$headers", "amqp$deliveryMode", "amqp$priority", "amqp$correlationId", "amqp$replyTo", "amqp$expiration", "amqp$messageId", "amqp$timestamp", "amqp$type", "amqp$userId", "amqp$appId", "amqp$clusterId", "amqp$routingKey")
配置详情
编写本文档时，只定义了适用于大多数场景的核心配置属性。 其他属性会随着该组件的发展逐步定义。配置 ConsumeAMQP：
- Queue - [必需] 从中获取消息的 AMQP 队列名称。通常由 管理员提供（例如 'amq.direct'）
- Host Name - [必需] 运行 AMQP 代理的主机名。通常由 管理员提供（例如 'myhost.com'）。默认为 'localhost'。
- Port - [必需] 运行 AMQP 代理的端口号。通常由 管理员提供（例如 '2453'）。默认为 '5672'。
- User Name - [必需] 连接 AMQP 代理的用户名。通常由 管理员提供（例如 'me'）。默认为 'guest'。
- Password - [必需] 与用户名配合使用的密码。通常由 管理员提供。默认为 'guest'。
- Use Certificate Authentication - [可选] 使用 SSL 证书的通用名称进行认证，而不是用户名/密码。 只能与 SSL 配合使用。默认为 'false'。
- Virtual Host - [可选] 用于隔离 AMQP 系统以增强 安全性的虚拟主机名。关于虚拟主机的更多细节参见 这篇博客。