ConsumeGCPubSub

ConsumeGCPubSub Processor

类型全限定名：cn.org.acal.lynxus.processors.gcp.pubsub.ConsumeGCPubSub

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

描述

从配置的 Google Cloud PubSub 订阅中消费消息。'Batch Size' 属性指定了在单个请求中将从订阅中拉取的最大消息数。'Processing Strategy' 属性指定每条消息是否应作为独立的 FlowFile 或将消息分组到单个 FlowFile 中。使用 Demarcator 策略将在格式允许时提供最佳吞吐量。使用 Record 可以转换数据格式以及进行 Schema 管理。使用 FlowFile 策略将为每条消息生成一个 FlowFile，并包含消息的属性作为 FlowFile 属性。

属性

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

名称	默认值	可选值	描述
GCP Credentials Provider Service		控制器服务 API： `GCPCredentialsService` cn.org.acal.lynxus-lynxus-gcp-services-api-nar-2.11.8	用于获取 Google Cloud Platform 凭据的Controller Service。
Project ID			谷歌云项目 ID 支持表达式语言：是（仅使用环境/系统变量求值）
Subscription			Google Cloud Pub/Sub 订阅的名称支持表达式语言：是（仅使用环境/系统变量求值）
Batch Size Threshold	`15`		指示云服务应在批次中捆绑的消息数量。如果未设置且留空，则批次中仅使用一条消息。
Processing Strategy	`FLOW_FILE`	FLOW_FILE — 为消费的每条 PubSub 消息写入一个 FlowFile DEMARCATOR — 为消费的每批 PubSub 消息写入一个 FlowFile RECORD — 写入一个 FlowFile，其中包含使用 Record Reader 和 Record Writer 消费并处理的多条 PubSub 消息	处理 PubSub 记录并将序列化输出写入 FlowFiles 的策略
Record Reader		控制器服务 API： `RecordReaderFactory` cn.org.acal.lynxus-lynxus-standard-services-api-nar-2.11.8	用于传入消息的Record Reader
Record Writer		控制器服务 API： `RecordSetWriterFactory` cn.org.acal.lynxus-lynxus-standard-services-api-nar-2.11.8	用于序列化传出FlowFile的Record Writer
Output Strategy	`USE_VALUE`	Use Content as Value — 仅将消息负载写入 FlowFile 记录。 Use Wrapper — 将消息的属性写入 FlowFile 记录。	用于将 Kafka 记录输出到 FlowFile 记录中的格式。
Message Demarcator			由于 PubSub 客户端按批次接收消息，该处理器提供了一个选项，可输出包含单个批次中所有消息的 FlowFile。此属性允许你提供一个字符串（按 UTF-8 解释），用于分隔多条消息。要输入诸如 'new line'（换行）之类的特殊字符，请根据操作系统使用 CTRL+Enter 或 Shift+Enter。
API Endpoint	`pubsub.googleapis.com:443`		以 [host:port] 形式覆盖 gRPC 端点支持表达式语言：是（仅使用环境/系统变量求值）
Proxy Configuration Service		控制器服务 API： `ProxyConfigurationService` cn.org.acal.lynxus-lynxus-standard-services-api-nar-2.11.8	指定代理配置 Controller Service 来代理网络请求。

关系 (Relationships)

名称	描述	自动终止
`success`	成功执行 Google Cloud Pub/Sub 操作后，FlowFile 将路由到此关系。	否

写入的 FlowFile 属性

名称	描述
`gcp.pubsub.ackId`	已消费的 Google Cloud PubSub 消息的确认 ID
`gcp.pubsub.messageSize`	已消费的 Google Cloud PubSub 消息的序列化大小
`gcp.pubsub.attributesCount`	已消费的 PubSub 消息具有的属性数量（如果有）
`gcp.pubsub.publishTime`	消息发布时的时间戳
`gcp.pubsub.subscription`	PubSub 订阅的名称
`Dynamic Attributes`	除列出的属性外，如果原始 Google Cloud Publisher 客户端在发送时向消息添加了任何属性，此处理器可能会写入零个或多个属性

状态管理

该组件不存储状态。

受限 (Restricted)

该组件不受限。

输入要求

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

另请参阅

PublishGCPubSub