ConsumeJMS Processor
描述
使用 BytesMessage、TextMessage、ObjectMessage、MapMessage 或 StreamMessage 类型的 JMS 消息，将其内容转换为 FlowFile 并将其转换为“成功”关系。 JMS 属性（例如标头和属性）将被复制为 FlowFile 属性。 MapMessages 将被转换为 JSON，然后转换为字节数组。 其他类型会将其原始内容作为字节数组传输到 FlowFile 中。
标签
属性
| 名称 | 默认值 | 可选值 | 描述 |
|---|---|---|---|
| Connection Factory Service | | 控制器服务 API：JMSConnectionFactoryProviderDefinitioncn.org.acal.lynxus-lynxus-jms-cf-service-nar-2.11.8 | 用于获取连接工厂的控制器服务。或者，也可以使用「JNDI *」或「JMS *」属性来配置连接工厂。 |
| Destination Name | | JMS 目标的名称。 通常由管理员提供（例如“topic://myTopic”或“myTopic”）。 支持表达式语言：是（使用 FlowFile 属性与环境/系统变量求值） | |
| Destination Type | QUEUE |
| JMS 目标的类型。 可以是“队列”或“主题”之一。 通常由管理员提供。 默认为“队列” |
| Message Selector | | 用于过滤处理器将接收的消息的 JMS 消息选择器 支持表达式语言：是（仅使用环境/系统变量求值） | |
| User Name | | 用于身份验证和授权的用户名。 支持表达式语言：是（仅使用环境/系统变量求值） | |
| Password | | 密码用于认证和授权。 敏感属性：是 | |
| Connection Client ID | | 如果设置，要在连接上设置的客户端 ID。对于持久非共享消费者，这是强制性的，对于所有其他消费者，这是可选的，通常对于共享消费者，不建议设置。请 支持表达式语言：是（仅使用环境/系统变量求值） | |
| Character Set | UTF-8 | 用于构造和解释文本消息的字符集名称 支持表达式语言：是（使用 FlowFile 属性与环境/系统变量求值） | |
| Acknowledgement Mode | 2 |
| JMS 确认模式。使用自动确认可能导致 Lynxus 重新启动时消息丢失，但可能提供比客户端确认更好的性能。 |
| Durable Subscription | false |
| 如果目标是 Topic，则使其成为消费者持久订阅。@see https://jakarta.ee/specifications/platform/9/apidocs/jakarta/jms/session#createDurableConsumer-jakarta.jms.Topic-java.lang.St 支持表达式语言：是（仅使用环境/系统变量求值） |
| Shared Subscription | false |
| 如果目标是主题，则使消费者共享。@see https://jakarta.ee/specifications/platform/9/apidocs/jakarta/jms/session#createSharedConsumer-jakarta.jms.Topic-java.lang.String- 支持表达式语言：是（仅使用环境/系统变量求值） |
| Subscription Name | | 如果目标是主题并且是共享的或持久的，则使用的订阅的名称。 支持表达式语言：是（仅使用环境/系统变量求值） | |
| Timeout | 1 sec | 在放弃之前要等待多长时间才能使用来自远程代理的消息。 支持表达式语言：是（仅使用环境/系统变量求值） | |
| Maximum Batch Size | 1 | 每次调用处理器时要发布或使用的最大消息数。 | |
| Error Queue Name | | JMS 队列的名称，如果设置，未处理的消息将被路由到此队列。通常由管理员提供（例如 'queue://myErrorQueue' 或 'myErrorQueue'）。仅当 'Destination Type' 设置为 'QUEUE' 时适用 支持表达式语言：是（使用 FlowFile 属性与环境/系统变量求值） | |
| Record Reader | | 控制器服务 API：RecordReaderFactorycn.org.acal.lynxus-lynxus-standard-services-api-nar-2.11.8 | Record Reader用于将接收到的 JMS 消息解析为记录。 |
| Record Writer | | 控制器服务 API：RecordSetWriterFactorycn.org.acal.lynxus-lynxus-standard-services-api-nar-2.11.8 | Record Writer用于在将记录写入FlowFile之前序列化记录。 |
| Output Strategy | USE_VALUE |
| 用于将 JMS 消息输出到 FlowFile 记录中的格式。 |
| JNDI Initial Context Factory Class | | JNDI 初始上下文工厂类的完全限定类名 (java.naming.factory.initial)。 支持表达式语言：是（仅使用环境/系统变量求值） | |
| JNDI Provider URL | | 用作 java.naming.provider.url 值的 JNDI 提供程序 URL。有关允许的 URL 方案，请参阅附加详细信息文档。 支持表达式语言：是（仅使用环境/系统变量求值） | |
| JNDI Name of the Connection Factory | | 要查找连接工厂的 JNDI 对象的名称。 支持表达式语言：是（仅使用环境/系统变量求值） | |
| JNDI / JMS Client Libraries | | 指定要添加到 ClassPath 的 jar 文件和/或目录，以便加载 JNDI / JMS 客户端库。这应该是文件、目录和/或 URL 的逗号分隔列表。如果目录 支持表达式语言：是（仅使用环境/系统变量求值） | |
| JNDI Principal | | 使用 JNDI 进行身份验证时使用的主体 (java.naming.security.principal)。 支持表达式语言：是（仅使用环境/系统变量求值） | |
| JNDI Credentials | | 使用 JNDI 进行身份验证时使用的凭据 (java.naming.security.credentials)。 敏感属性：是 | |
| JMS Connection Factory Implementation Class | | JMS ConnectionFactory 实现类的完全限定名称（例如 org.apache.activemq.ActiveMQConnectionFactory）。 支持表达式语言：是（仅使用环境/系统变量求值） | |
| JMS Client Libraries | | 包含要添加到类路径的附加资源（例如 JAR、配置文件等）的目录路径（定义为值的逗号分隔列表）。这些资源通常表示目标 支持表达式语言：是（仅使用环境/系统变量求值） | |
| JMS Broker URI | | 指向 JMS 消息代理网络位置的 URI。ActiveMQ 示例：'tcp://myhost:61616'。IBM MQ 示例：'myhost(1414)' 和 'myhost01(1414),myhost02(1414)'。 支持表达式语言：是（仅使用环境/系统变量求值） | |
| JMS SSL Context Service | | 控制器服务 API：SSLContextServicecn.org.acal.lynxus-lynxus-standard-services-api-nar-2.11.8 | SSL 上下文服务用于为 TLS/SSL 连接提供客户端证书信息。 |
动态属性
| 名称 | 值 | 描述 |
|---|---|---|
| The name of a Connection Factory configuration property. | The value of a given Connection Factory configuration property. | Connection Factory 的附加配置属性。当通过处理器的 'JNDI *' 或 'JMS *' 属性配置 Connection Factory 时可以使用。有关更多信息，请参阅附加详细信息页面。 |
关系 (Relationships)
| 名称 | 描述 | 自动终止 |
|---|---|---|
parse.failure | 如果消息无法使用配置的 Record Reader 进行解析，消息的内容将作为独立的 FlowFile 路由到此关系。 | 是 |
success | 从 JMS 目标接收的所有 FlowFile 都会路由到此关系 | 否 |
写入的 FlowFile 属性
| 名称 | 描述 |
|---|---|
jms_deliveryMode | 消息头中的 JMSDeliveryMode。 |
jms_expiration | 消息头中的 JMSExpiration。 |
jms_priority | 消息头中的 JMSPriority。 |
jms_redelivered | JMS从消息头中重新投递。 |
jms_timestamp | 消息头中的 JMSTimestamp。 |
jms_correlationId | 消息头中的 JMSCorrelationID。 |
jms_messageId | 消息头中的 JMSMessageID。 |
jms_type | 消息头中的 JMSType。 |
jms_replyTo | 消息头中的 JMSReplyTo。 |
jms_destination | 消息头中的 JMSDestination。 |
jms.messagetype | JMS 消息类型可以是 TextMessage、BytesMessage、ObjectMessage、MapMessage 或 StreamMessage。 |
other attributes | 每个消息属性都写入一个属性。 |
状态管理
该组件不存储状态。
受限 (Restricted)
该组件不受限。
输入要求
该组件不允许有传入连接。
详细说明
ConsumeJMS
概述
该 Processor 从符合 JMS 的消息系统消费消息，并把它们转换为 FlowFile，路由到流程中的下一个组件。
该 Processor 做两件事。它通过从消费的 JMS 消息中提取信息 （包括消息体、标准 JMS Header 和 Property）来构建 FlowFile。消息体写入 FlowFile，标准 JMS Header 和 Property 设置为 FlowFile Attribute。
配置详情
编写本文档时，只定义了适用于大多数场景的核心配置属性。 其他属性会随着该组件的发展逐步定义。配置 ConsumeJMS：
- User Name - [可选] 当该 Processor 从预先配置的 javax.jms.ConnectionFactory（见下文）获取 javax.jms.Connection 时，用于认证和授权的用户名。
- Password - [可选] 与 User Name 配合使用的密码。
- Destination Name - [必需] javax.jms.Destination 的名称。通常由管理员提供（例如 'topic://myTopic'）。
- Destination Type - [必需] javax.jms.Destination 的类型。可以是 'QUEUE' 或 'TOPIC' 之一，通常由管理员 提供。默认为 'QUEUE'。
Connection Factory 配置
为 Processor 配置 Connection Factory 有多种方式：
- Connection Factory Service 属性 - 链接到预先配置的 Controller Service（JndiJmsConnectionFactoryProvider 或 JMSConnectionFactoryProvider）
- JNDI * 属性 - Processor 级别的配置，属性与 JndiJmsConnectionFactoryProvider Controller Service 的属性相同，这种情况下也可以使用动态属性
- JMS * 属性 - Processor 级别的配置，属性与 JMSConnectionFactoryProvider Controller Service 的属性相同，这种情况下也可以使用动态属性
推荐方式是使用 Connection Factory Service 属性和预先配置的 Controller Service。这也是最方便的方式，因为只需配置一次 Controller Service，就可以在多个 Processor 中使用。
不过，由于第三方 JMS 客户端库与 Lynxus 之间的 Java ClassLoader 处理不兼容，某些 JMS 客户端库可能无法与 Controller Service 一起工作。如果使用 Controller Service 时遇到 java.lang.ClassCastException 错误，请尝试通过 'JNDI *' 或 'JMS *' 以及 Processor 的动态属性配置 Connection Factory。关于这些属性的更多细节，参见相应 Controller Service 的文档（'JNDI *' 对应 JndiJmsConnectionFactoryProvider，'JMS *' 对应 JMSConnectionFactoryProvider）。