ConsumeMQTT Processor
描述
订阅主题并从 MQTT 代理接收消息
标签
属性
| 名称 | 默认值 | 可选值 | 描述 |
|---|---|---|---|
| Broker URI | | 用于连接到 MQTT 代理的 URI（例如 tcp://localhost:1883）。支持 'tcp'、'ssl'、'ws' 和 'wss' 协议。要使用 'ssl'，必须设置 SSL Context Service 属性。当设置逗号分隔的 URI 列表时（例如 tcp://localhost:1883,tcp://localhost:1884），处理器将在连接失败时使用轮询算法连接到代理。 支持表达式语言：是（仅使用环境/系统变量求值） | |
| MQTT Specification Version | 0 |
| 与代理连接时的 MQTT 规范版本。 有关详细信息，请参阅允许值说明。 |
| Username | | 连接到代理时使用的用户名 支持表达式语言：是（仅使用环境/系统变量求值） | |
| Password | | 连接到代理时使用的密码 敏感属性：是 | |
| SSL Context Service | | 控制器服务 API：SSLContextServicecn.org.acal.lynxus-lynxus-standard-services-api-nar-2.11.8 | SSL 上下文服务用于为 TLS/SSL 连接提供客户端证书信息。 |
| Session State | true |
| 是否开始新的流程或恢复之前的流程。 有关详细信息，请参阅允许值说明。 |
| Session Expiry Interval | 24 hrs | 在此间隔之后，代理将使客户端过期并清除会话状态。 | |
| Client ID | | 要使用的 MQTT 客户端 ID。 如果未设置，将生成一个 UUID。 支持表达式语言：是（仅使用环境/系统变量求值） | |
| Group ID | | 要使用的 MQTT 消费者组 ID。 如果未设置组 ID，客户端将作为个人消费者进行连接。 | |
| Topic Filter | | MQTT 主题过滤器，用于指定要订阅的主题。 支持表达式语言：是（仅使用环境/系统变量求值） | |
| Quality of Service | 0 |
| 接收消息的服务质量 (QoS)。 接受值“0”、“1”或“2”； “0”表示“最多一次”，“1”表示“至少一次”，“2”表示“恰好一次”。 |
| Record Reader | | 控制器服务 API：RecordReaderFactorycn.org.acal.lynxus-lynxus-standard-services-api-nar-2.11.8 | Record Reader用于将接收到的 MQTT 消息解析为记录。 |
| Record Writer | | 控制器服务 API：RecordSetWriterFactorycn.org.acal.lynxus-lynxus-standard-services-api-nar-2.11.8 | Record Writer用于在将记录写入FlowFile之前序列化记录。 |
| Add Attributes as Fields | true |
| 如果将此属性设置为 true，将在每条记录中添加默认字段：_topic、_qos、_isDuplicate、_isRetained。 |
| Message Demarcator | | 使用此属性，您可以选择输出包含多条消息的 FlowFile。此属性允许您提供一个字符串（解释为 UTF-8）用于分隔多条消息。这是可选属性；如果未提供且未定义 Record Reader/Writer，则收到的每条消息将生成单个 FlowFile。要输入特殊字符（如'换行符'），请使用 CTRL+Enter 或 Shift+Enter，具体取决于操作系统。 支持表达式语言：是（仅使用环境/系统变量求值） | |
| 连接超时 | 30 | 客户端等待与 MQTT 服务器的网络连接建立的最大时间间隔（以秒为单位）。默认超时为 30 秒。值为 0 将禁用超时处理，表示客户端将一直等待直到网络连接成功建立或失败。 | |
| Keep Alive | 60 | 定义发送或接收消息之间的最大时间间隔（以秒为单位）。它使客户端能够检测服务器是否不再可用，而无需等待 TCP/IP 超时。客户端将确保在每个保活周期内至少有一条消息穿越网络。在该时间段内如果没有数据相关消息，客户端将发送一个非常小的 "ping" 消息，服务器将确认该消息。值为 0 将禁用客户端中的保活处理。 | |
| Last Will Message | | 作为客户的遗嘱发送的消息。 | |
| Last Will Topic | | 发送客户遗嘱的主题。 | |
| Last Will Retain | false |
| 是否保留委托人的遗嘱。 |
| Last Will QoS Level | 0 |
| 发布遗嘱消息时使用的 QoS 级别。 |
| Max Queue Size | | MQTT 消息始终被发送到主题上的订阅者，无论处理器的调度频率如何。如果'运行调度'明显落后于消息到达此处理器的速率，则可能在此处理器的内部队列中出现积压。此属性指定此处理器在内部队列中一次可以保存在内存中的最大消息数。如果 Lynxus 重新启动，这些数据将丢失。 |
关系 (Relationships)
| 名称 | 描述 | 自动终止 |
|---|---|---|
Message | MQTT消息输出 | 否 |
parse.failure | 如果无法使用配置的 Record Reader 解析消息，消息的内容将作为其独立的 FlowFile 路由到此关系。 | 是 |
写入的 FlowFile 属性
| 名称 | 描述 |
|---|---|
record.count | 收到的记录数 |
mqtt.broker | 作为消息源的 MQTT 代理 |
mqtt.topic | 收到消息的 MQTT 主题 |
mqtt.qos | 此消息的服务质量。 |
mqtt.isDuplicate | 此消息是否可能与已收到的消息重复。 |
mqtt.isRetained | 此消息是否来自当前发布者，或者是 "retained" by the server as the last message published on the topic. |
状态管理
该组件不存储状态。
受限 (Restricted)
该组件不受限。
输入要求
该组件不允许有传入连接。
系统资源考量
| 资源 | 描述 |
|---|---|
| MEMORY | 'Max Queue Size' 指定 Lynxus 中此处理器的单个实例可以在内存中保存的最大消息数。此属性的较高值可能表示大量数据存储在内存中。 |
另请参阅
详细说明
ConsumeMQTT
无论 Processor 的调度频率如何，MQTT 消息都会持续发送给主题上的订阅者。 如果 'Run Schedule' 明显落后于消息到达该 Processor 的速率，该 Processor 的内部队列可能会发生积压。每次 Processor 被调度时，内部 队列中的消息会被写入 FlowFile。如果内部队列已满，MQTT 客户端会 在最长 1 秒内尝试把消息加入内部队列。如果这段时间之后内部 队列仍然已满，会抛出 'The subscriber queue is full' 异常，消息 会被丢弃，客户端会断开连接。当 QoS 属性设为 0 时，消息会丢失。 当 QoS 属性设为 1 或 2 时，消息会在客户端重连后被接收。