ConsumeAzureEventHub

文档中心 › 组件索引 › ConsumeAzureEventHub

ConsumeAzureEventHub Processor

类型全限定名：cn.org.acal.lynxus.processors.azure.eventhub.ConsumeAzureEventHub

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

描述

从 Microsoft Azure Event Hubs 接收消息并进行检查点跟踪以确保一致的事件处理。

标签

azuremicrosoftcloudeventhubeventsstreamingstreams

属性

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

名称默认值可选值描述
Event Hub NamespaceAzure 事件中心分配到的Namespace。 这通常等于 <事件中心名称>-ns。
支持表达式语言：是（仅使用环境/系统变量求值）
Event Hub Name要从中提取消息的事件中心的名称。
支持表达式语言：是（仅使用环境/系统变量求值）
Service Bus Endpoint.servicebus.windows.net
  • Azure — 通用 Service Bus 端点
  • Azure 中国 — 中国 Service Bus 端点
  • Azure 德国 — 德国 Service Bus 端点
  • Azure 美国政府 — 美国政府 Service Bus 端点
支持不在默认 windows.net 域中的Namespace。
Transport TypeAmqp
  • AMQP — 通过 TCP 在端口 5671 和 5672 上使用 AMQP
  • AMQP_WEB_SOCKETS — 通过 HTTPS 使用 WebSockets 在端口 443 上使用 AMQP
用于与 Azure 事件中心通信的高级消息队列协议传输类型
Shared Access Policy Name共享访问策略的名称。 该保单必须包含 Listen 声明。
支持表达式语言：是（仅使用环境/系统变量求值）
Shared Access Policy Key共享访问策略的关键。 可以使用主键或辅助键。
敏感属性：是
Authentication StrategyMANAGED_IDENTITY
  • Managed Identity — 使用承载它的 Azure 资源的托管标识（Managed Identity）进行身份验证。
  • Shared Access Signature — 使用共享访问策略（Shared Access Policy）名称和密钥进行身份验证。
  • OAuth2 — 使用由 Entra 注册应用程序支持的 OAuth2 访问令牌提供程序（Access Token Provider）进行身份验证。
  • Identity Federation — 使用与 Azure AD 访问令牌交换的工作负载身份令牌进行身份验证。
指定用于向 Azure 事件中心进行身份验证的策略
Event Hubs Access Token Provider控制器服务 API：
OAuth2AccessTokenProvider
cn.org.acal.lynxus-lynxus-standard-services-api-nar-2.11.8		Controller Service提供 OAuth2 访问令牌以对 Azure 事件中心进行身份验证
Event Hubs Identity Federation Token Provider控制器服务 API：
AzureIdentityFederationTokenProvider
cn.org.acal.lynxus-lynxus-azure-services-api-nar-2.11.8		在向 Azure 事件中心进行身份验证时，Controller Service将工作负载身份令牌交换为 Azure AD 访问令牌。
Consumer Group$Default要使用的消费者组的名称。
支持表达式语言：是（仅使用环境/系统变量求值）
Record Reader控制器服务 API：
RecordReaderFactory
cn.org.acal.lynxus-lynxus-standard-services-api-nar-2.11.8		用于读取接收消息的记录读取器。事件中心名称可以通过表达式语言 '${eventhub.name}' 引用来访问模式。
Record Writer控制器服务 API：
RecordSetWriterFactory
cn.org.acal.lynxus-lynxus-standard-services-api-nar-2.11.8		用于将记录序列化到输出 FlowFile 的记录写入器。事件中心名称可以通过表达式语言 '${eventhub.name}' 引用来访问模式。如果未指定，每条消息将作为单独的 FlowFile 写入。
Initial Offsetend-of-stream
  • Start of stream — 从流中保留的最旧消息开始读取。
  • End of stream — 忽略旧的保留消息（即使存在），从现在开始读取新消息。
指定如果偏移量尚未存储在检查点存储中则从何处开始接收消息。
Prefetch Count300
支持表达式语言：是（仅使用环境/系统变量求值）
Batch Size10在 Lynxus 会话中要处理的消息数。此参数影响吞吐量和一致性。Lynxus 在处理完此数量的消息后提交其会话和事件中心检查点
支持表达式语言：是（仅使用环境/系统变量求值）
Message Receive Timeout1 min该消费者在返回之前应等待接收批量大小的时间。
支持表达式语言：是（仅使用环境/系统变量求值）
Checkpoint StrategyAZURE_BLOB_STORAGE
  • Azure Blob Storage — 使用 Azure Blob Storage 存储分区所有权和检查点信息
  • Component State — 使用组件状态来存储分区所有权和检查点信息
指定使用哪种策略来存储和检索每个分区的分区所有权和检查点信息。
Storage Account Name用于存储事件中心使用者组状态的 Azure 存储帐户的名称。
支持表达式语言：是（仅使用环境/系统变量求值）
Storage Container Name用于存储事件中心消费者组状态的 Azure 存储容器名称。如果未指定，则使用事件中心名称。
支持表达式语言：是（仅使用环境/系统变量求值）
Blob Storage Authentication StrategySTORAGE_ACCOUNT_KEY
  • Storage Account Key — 使用账户密钥（account key）对 Azure Blob Storage 进行身份验证。
  • Shared Access Signature — 使用 SAS 令牌对 Azure Blob Storage 进行身份验证。
  • OAuth2 — 使用由 Entra 注册应用程序支持的 OAuth2 访问令牌提供程序（Access Token Provider）对 Azure Blob Storage 进行身份验证。
  • Identity Federation — 使用与 Azure AD 访问令牌交换的工作负载身份令牌对 Azure Blob 存储进行身份验证。
用于在持久检查点时访问 Azure Blob 存储的身份验证策略。
Storage Account Key用于存储事件中心使用者组状态的 Azure 存储帐户密钥。
敏感属性：是
支持表达式语言：是（仅使用环境/系统变量求值）
Storage SAS Token用于存储事件中心使用者组状态的 Azure 存储 SAS 令牌。 总是以 ? 开头 特点。
敏感属性：是
支持表达式语言：是（仅使用环境/系统变量求值）
Storage Access Token Provider控制器服务 API：
OAuth2AccessTokenProvider
cn.org.acal.lynxus-lynxus-standard-services-api-nar-2.11.8		Controller Service提供 OAuth2 访问令牌，以便在保留检查点时向 Azure Blob 存储进行身份验证。
Storage Identity Federation Token Provider控制器服务 API：
AzureIdentityFederationTokenProvider
cn.org.acal.lynxus-lynxus-azure-services-api-nar-2.11.8		将身份联合与 Azure Blob 存储结合使用时，Controller Service将工作负载身份令牌交换为 Azure AD 访问令牌。
Proxy Configuration Service控制器服务 API：
ProxyConfigurationService
cn.org.acal.lynxus-lynxus-standard-services-api-nar-2.11.8		指定代理配置 Controller Service 来代理网络请求。

关系 (Relationships)

名称描述自动终止
success从事件中心接收的FlowFile。否

写入的 FlowFile 属性

名称描述
eventhub.enqueued.timestamp消息在事件中心排队的时间（自纪元以来的毫秒数，UTC）
eventhub.offset存储消息的分区的偏移量
eventhub.sequence与消息关联的序列号
eventhub.name从中提取消息的事件中心的名称
eventhub.partition从中提取消息的分区的名称
eventhub.property.*此消息的应用程序属性。 IE：“应用程序”将是“eventhub.property.application”

状态管理

作用域描述
LOCAL、CLUSTER本地状态用于存储客户端 ID。当组件状态配置为检查点策略时，集群状态用于存储分区所有权和检查点信息。

受限 (Restricted)

该组件不受限。

输入要求

该组件不允许有传入连接。

© 2026 ACAL Organization. Licensed under Apache License 2.0.