将 MQTT 数据传输到 GCP Pub/Sub

Google Cloud Pub/Sub 是一种异步消息传递服务，旨在实现极高的可靠性和可扩缩性。EMQX 支持与 Google Cloud Pub/Sub 的无缝集成，能够实时提取、处理和分析 MQTT 数据，并将数据推送到各类 Google Cloud 服务，如 Cloud Functions、App Engine、Cloud Run、Kubernetes Engine 和 Compute Engine 中，或将 Google Cloud 中的数据通过 MQTT 下发，帮助用户更快的基于 GCP 构建物联网应用。

本页详细介绍了 EMQX 与 GCP Pub/Sub 的数据集成并提供了实用的规则和 Sink/Source 创建指导。

工作原理

GCP Pub/Sub Sink 是 EMQX 的开箱即用功能，旨在帮助用户轻松地将 MQTT 数据流与 Google Cloud 集成，并利用其丰富的服务和功能实现物联网应用开发。

GCP_bridge_architect

EMQX 通过规则引擎与 Sink 将 MQTT 数据转发至 GCP Pub/Sub，以 GCP Pub/Sub 生产者角色为例，其完整流程如下：

物联网设备发布消息：设备通过特定的主题发布遥测和状态数据，消息将触发规则引擎。
规则引擎处理消息：通过内置的规则引擎，可以根据主题匹配处理特定来源的 MQTT 消息。规则引擎会匹配对应的规则，并对消息进行处理，例如转换数据格式、过滤掉特定信息或使用上下文信息丰富消息。
桥接到 GCP Pub/Sub：规则触发将消息转发到 GCP Pub/Sub 的动作，允许轻松配置数据到 GCP Pub/Sub 属性，排序键，以及 MQTT 主题到 GCP Pub/Sub 主题的映射关系，可以为数据集成提供更丰富的上下文信息和顺序保证，实现灵活的物联网数据处理。

MQTT 消息数据写入到 GCP PusSub 后，您可以进行灵活的应用开发，例如：

实时数据处理和分析：利用 Google Cloud 的强大数据处理和分析工具，如 Dataflow、BigQuery 和 Pub/Sub 自身的流处理功能，对消息数据进行实时处理和分析，从而获得有价值的洞察和决策支持。
事件驱动的功能：触发 Google Cloud 的事件处理如 Cloud Functions 和 Cloud Run，以实现动态、灵活的功能触发和处理。
数据存储和共享：将消息数据传输到 Google Cloud 的存储服务中，如 Cloud Storage 和 Firestore，以便安全地存储和管理大量的数据，并与其他 Google Cloud 服务共享和分析这些数据，以满足不同的业务需求。

特性与优势

将 EMQX 与 GCP Pub/Sub 结合使用具有以下特性与优势：

强大的消息传递服务：EMQX 与 GCP Pub/Sub 都具备高可用、可扩展的特性，能够可靠地接收、传递和处理大规模的消息流，支持物联网数据顺序传递、消息质量保证以及持久化等特性，确保消息的可靠传递和处理。
灵活的规则引擎：通过内置的规则引擎，可以根据主题匹配处理特定来源的消息和事件。并对消息和事件进行处理，例如转换数据格式、过滤掉特定信息或使用上下文信息丰富消息，结合 GCP Pub/Sub 可以进行进一步处理和分析。
丰富的上下文信息：通过 GCP Pub/Sub 桥接，你可以在消息中添加更丰富的上下文信息，实现客户端属性与 Pub/Sub 属性、排序键等的映射，可以帮助在后续的应用开发和数据处理中进行更精确的分析和处理。

综上所述，将 EMQX 和 GCP Pub/Sub 结合使用可以实现高可靠性、可扩展性的消息传递，并通过丰富工具和服务进行数据分析与集成，这使得你能够构建强大的物联网应用，并基于事件驱动的功能实现灵活的业务逻辑。

准备工作

本节介绍如何配置 GCP Pub/Sub，并创建主题与获取连接凭证。

前置准备

了解规则。
了解数据集成。

创建服务账号凭证

如果使用服务账号 JSON认证，请在 GCP 中创建服务账号，并生成 JSON 格式的密钥。

在您的 GCP 账户中创建一个服务账号。确保该服务账号具有检查、读取和发布消息到目标主题的权限（例如，赋予 Pub/Sub Editor 角色）。
点击您创建的服务账号的电子邮件地址。然后，点击密钥选项卡。在添加密钥列表中，选择创建新密钥，为该账号生成一个服务账号密钥，并以 JSON 格式下载。
TIP
请妥善存储服务账号密钥，以便后续使用。

GCP 服务账号凭证

配置工作负载身份联合

工作负载身份联合（WIF）允许 EMQX 无需持有服务账号密钥文件即可访问 GCP 资源。EMQX 将从外部身份提供商（如 Microsoft Azure）获取的 token 通过 GCP Security Token Service 换取临时 GCP token，再凭此模拟指定的 GCP 服务账号。Token 续期由 EMQX 自动处理。

要使用 WIF，请在创建连接器之前在 GCP 项目中完成以下配置。

在 Google Cloud 控制台中，进入 IAM 和管理 -> 工作负载身份联合，创建一个工作负载身份池，并记录池 ID 和项目编号。
向该池添加提供商并记录提供商 ID。如使用基于 OIDC 的认证，请从外部身份提供商处获取 OAuth 2.0 客户端凭证（客户端 ID、客户端密钥和令牌端点 URI）。
授予工作负载身份池权限，使其能够模拟具有 Pub/Sub 主题访问权限的 GCP 服务账号。配置连接器时需要填写服务账号的电子邮件地址。
TIP
详细配置步骤请参阅配置工作负载身份联合。

示例：Microsoft Azure（Entra ID）

在 Microsoft Entra ID 中注册一个公开 API 的应用程序，并为其创建客户端密钥。配置连接器时使用以下值：

连接器字段	值
OAuth Token 端点 URI	`https://login.microsoftonline.com/<租户 ID>/oauth2/v2.0/token`
OAuth 客户端 ID	应用程序（客户端）ID，格式为 `api://<应用程序 ID>`
OAuth 客户端密钥	为该应用程序生成的客户端密钥
OAuth 请求范围	`api://<应用程序 ID>/.default`

注意

OAuth 请求范围必须与应用程序的受众（aud）完全匹配，否则与 GCP STS 的令牌交换将会失败。详情请参阅 Microsoft 文档中的 OAuth 2.0 客户端凭证流。

向 WIF 池授予服务账号访问权限时，请使用对象 ID（而非应用程序 ID）作为主体标识符（Subject）。对象 ID 显示在 Azure 门户企业应用程序下对应应用的概述页面中。

附加服务账号前提条件

要使用附加服务账号认证，EMQX 必须运行在已附加服务账号的 GCP Compute Engine 实例上。确保该实例的 OAuth 访问范围允许访问 Pub/Sub。Google 建议使用 cloud-platform 访问范围（https://www.googleapis.com/auth/cloud-platform），并通过 IAM 角色限制服务账号的权限。该服务账号必须具有访问所需 Pub/Sub 主题和订阅的权限。详情参见 Google Cloud 文档中的服务账号。

目标 Pub/Sub 主题和订阅必须位于该 Compute Engine 实例所属的 GCP 项目中。在 EMQX 集群中，每个节点都必须满足上述要求，并运行在该项目的 Compute Engine 实例上。

连接器启动时，EMQX 会自动从实例元数据端点获取 GCP 项目 ID 和访问令牌，无需上传服务账号密钥文件。

在 GCP Pub/Sub 中创建主题

打开 Pub/Sub 控制台，点击 CREATE TOPIC，输入自定义的 Topic ID，点击 CREATE 即可完成创建。

GCP PubSub 创建主题

点击列表页对应的 Topic ID 即可进入 Topic 详情页面，您需要创建一个 subscription 来保存消息，有关 subscription 详细介绍请参考 CCP Pub/Sub subscription，此处选择 Pull 类型，保留 7 天历史消息。

GCP PubSub 创建订阅

点击 Subscription ID -> MESSAGES -> PULL 可以在线查看发送到主题中的消息。

创建 GCP Pub/Sub 生产者连接器

在添加 GCP Pub/Sub 生产者 Sink 操作之前，您需要创建 GCP Pub/Sub 生产者连接器，以建立 EMQX 与 GCP Pub/Sub 之间的连接。

转到 EMQX Dashboard，点击集成 -> 连接器。
在页面的右上角点击创建，在连接器选择页面选择 Google PubSub 生产者，然后点击下一步。
输入连接器名称和描述，例如 my-pubsubproducer。名称用于将 GCP Pub/Sub 生产者 Sink 与连接器关联，并且必须在集群内唯一。
在认证列表中选择以下一种认证方式，并配置相应字段：
- 服务账号 JSON：上传您在创建服务账号凭证中导出的 JSON 格式服务账号凭证。
- 工作负载身份联合 (WIF)：填写以下字段。此方式无需服务账号 JSON 文件。前置条件请参见配置工作负载身份联合。
  - GCP 项目 ID：连接器所访问资源的 GCP 项目 ID。
  - GCP 项目编号：连接器所访问资源的 GCP 项目编号。
  - 服务账号邮箱：需要模拟的服务账号电子邮件地址。
  - 工作负载身份池 ID：WIF 令牌交换中使用的工作负载身份池 ID。
  - 工作负载身份提供商 ID：WIF 令牌交换中使用的工作负载身份提供商 ID。
  - 凭证类型：外部身份提供商使用的凭证类型，目前支持 OIDC 客户端凭证，选择后填写以下字段：
    - OAuth 客户端 ID：用于向 OAuth 服务器请求令牌的客户端 ID。
    - OAuth 客户端密钥：用于向 OAuth 服务器请求令牌的客户端密钥。
    - OAuth Token 端点 URI：OIDC 提供商的 OAuth Token 端点 URI。
    - OAuth 请求范围：向 OAuth 服务器请求访问令牌时指定的 scope（如提供商要求则需填写）。
- 附加服务账号：无需填写额外字段。EMQX 会自动从实例元数据端点获取 GCP 项目 ID 和访问令牌。前提条件请参见附加服务账号前提条件。
在点击创建之前，您可以点击 测试连接 以测试连接器是否能连接到 GCP Pub/Sub 服务器。
点击底部的创建按钮完成连接器的创建。在弹出对话框中，您可以点击 返回连接器列表 或点击 创建规则 继续创建带有 GCP Pub/Sub 生产者 Sink 的规则，以指定要转发到 GCP Pub/Sub 的数据。详细步骤请参见创建 GCP Pub/Sub 生产者 Sink 规则。

创建 GCP Pub/Sub 生产者 Sink 规则

本节演示了如何为 Google PubSub Sink 创建一条规则以指定需要转发至 GCP Pub/Sub 的数据。

转到 Dashboard 集成 -> 规则页面。
点击页面右上角的创建。
输入规则 ID，例如： my_rule。
在 SQL 编辑器中输入规则，请确保规则选择出来的字段（SELECT 部分）包含 HTTP 请求消息体模版中用到的变量。例如将 /devices/+/events 主题的 MQTT 消息集成到 GCP Pub/Sub，此处规则 SQL 如下：
sql
```
SELECT
  *
FROM
  "/devices/+/events"
```
TIP
如果您初次使用 SQL，可以点击 SQL 示例和启用调试来学习和测试规则 SQL 的结果。
点击右侧的添加动作按钮，为规则在被触发的情况下指定一个动作。在动作类型下拉框中选择 Google PubSub 生产者，以便 EMQX 将规则处理后的数据发送到 GCP Pub/Sub。
保持动作下拉框为默认的创建动作选项，您也可以选择一个之前已经创建好的 Google PubSub Producer Sink。此处我们创建一个全新的 Sink 并添加到规则中。
在名称字段中，为 Sink 输入一个名称。名称应为大小写字母和数字的组合。
从连接器下拉框中选择刚刚创建的 my_pubsubprodcer。您也可以通过点击下拉框旁边的按钮创建一个新的连接器。关于配置参数，请参见创建连接器。
在 GCP PubSub 主题中，输入以下任一值：
- 主题名称，例如按照在 GCP Pub/Sub 中创建主题中的步骤创建的 my-iot-core。EMQX 在已配置服务账户所属的项目中解析该主题。
- 完整主题路径，格式为 projects/<project-id>/topics/<topic-name>。使用此格式可将消息发布到其他 GCP 项目中的主题。请为已配置的服务账户授予访问该项目中目标主题所需的 Pub/Sub 权限。
在 HTTP 请求消息体模版中定义模板，或留空。

如果留空，它将使用 JSON 格式对 MQTT 消息中的所有可见输入进行编码，例如 clientid、topic、payload 等。
如果使用定义的模板，占位符的形式为 ${variable_name}，将用 MQTT 上下文中的相应值填充。例如，如果 MQTT 消息主题是 my/topic，则 ${topic} 将被替换为 my/topic。

在属性模板和排序键模板中定义用于格式化传出消息的属性和/或排序键的模板。
- 对于属性模版，键和值都可以使用形式为 ${variable_name} 的占位符。这些值将从 MQTT 上下文中提取。如果键模板解析为空字符串，则该键不会包含在传出到 GCP Pub/Sub 的消息中。
- 对于排序键模版，可以使用形式为 ${variable_name} 的占位符。如果解析的值为空字符串，则不会为 GCP Pub/Sub 传出消息设置 orderingKey 字段。
备选动作（可选）：如果您希望在消息投递失败时提升系统的可靠性，可以为 Sink 配置一个或多个备选动作。当 Sink 无法成功处理消息时，这些备选动作将被触发。更多信息请参见：备选动作。
展开高级设置，根据需要配置可选设置。详细信息请参见高级设置。
在点击创建之前，您可以点击测试连接性来测试连接器是否能连接到 GCP Pub/Sub 服务器。
点击创建按钮完成 Sink 配置，您将在动作输出标签下看到新的 Sink。
回到创建规则页面，点击创建来创建规则。

您现在已成功创建了规则。您可以在集成 -> 规则页面看到新创建的规则。点击**动作(Sink)**标签，您可以看到新的 Google PubSub 生产者 Sink。

您也可以点击集成 -> 流程设计器查看拓扑，并且可以看到，通过规则 my_rule 解析后，主题 /devices/+/events 下的消息被发送并保存到 GCP Pub/Sub。

至此您已经完成整个创建过程，可以前往集成 -> Flow 设计器页面查看拓扑图，此时应当看到 /devices/+/events 主题的消息经过名为 my_rule 的规则处理，处理结果写入到 GCP Pub/Sub 中。

测试 GCP Pub/Sub 生产者 Sink 规则

使用 MQTTX 向 /devices/+/events 主题发布消息：

bash

mqttx pub -i emqx_c -t /devices/+/events -m '{ "msg": "hello GCP PubSub" }'

查看 Sink 运行统计，命中、发送成功次数均 +1。
前往 GCP Pub/Sub 控制台查看数据是否已经发送成功。

创建 GCP Pub/Sub 消费者连接器

在添加 GCP Pub/Sub 消费者 Source 之前，您需要创建 GCP Pub/Sub 消费者连接器，以建立 EMQX 与 GCP Pub/Sub 之间的连接。

转到 EMQX Dashboard，点击集成 -> 连接器。
在页面的右上角点击创建，在连接器选择页面选择 Google PubSub 消费者，然后点击 下一步。
输入连接器名称和描述，例如 my-pubsubconsumer。名称用于将 GCP Pub/Sub 消费者连接器关联，并且必须在集群内唯一。
在认证列表中选择以下一种认证方式，并配置相应字段：
- 服务账号 JSON：上传您在创建服务账号凭证中导出的 JSON 格式服务账号凭证。
- 工作负载身份联合 (WIF)：填写以下字段。此方式无需服务账号 JSON 文件。前置条件请参见配置工作负载身份联合。
  - GCP 项目 ID：连接器所访问资源的 GCP 项目 ID。
  - GCP 项目编号：连接器所访问资源的 GCP 项目编号。
  - 服务账号邮箱：需要模拟的服务账号电子邮件地址。
  - 工作负载身份池 ID：WIF 令牌交换中使用的工作负载身份池 ID。
  - 工作负载身份提供商 ID：WIF 令牌交换中使用的工作负载身份提供商 ID。
  - 凭证类型：外部身份提供商使用的凭证类型，目前支持 OIDC 客户端凭证，选择后填写以下字段：
    - OAuth 客户端 ID：用于向 OAuth 服务器请求令牌的客户端 ID。
    - OAuth 客户端密钥：用于向 OAuth 服务器请求令牌的客户端密钥。
    - OAuth Token 端点 URI：OIDC 提供商的 OAuth Token 端点 URI。
    - OAuth 请求范围：向 OAuth 服务器请求访问令牌时指定的 scope（如提供商要求则需填写）。
- 附加服务账号：无需填写额外字段。EMQX 会自动从实例元数据端点获取 GCP 项目 ID 和访问令牌。前提条件请参见附加服务账号前提条件。
在点击创建之前，您可以点击 测试连接 以测试连接器是否能连接到 GCP Pub/Sub 服务器。
点击底部的创建按钮完成连接器的创建。在弹出对话框中，您可以点击 返回连接器列表 或点击 创建规则 继续创建带有 GCP Pub/Sub 消费者 Source 的规则，以消费来自 GCP Pub/Sub 的数据并转发到 EMQX 本地。详细步骤请参见创建 GCP Pub/Sub 消费者 Source 规则。

创建 GCP Pub/Sub 消费者 Source 规则

本节演示如何在 EMQX 中创建规则，以消费来自 GCP Pub/Sub 的消息并将消息转发给 EMQX。您需要创建并配置一个 Google PubSub 消费者 Source，并将其作为数据输入添加到规则中。您还需要向规则中添加一个重发布动作，以将消息从 GCP Pub/Sub 转发至 EMQX。

转到 EMQX 控制台，点击集成 -> 规则。
在页面右上角点击创建。
输入 my_rule_source 作为规则 ID。
在右侧的数据输入页签下，删除默认输入 消息。点击添加输入。
从输入类型下拉菜单中选择 Google PubSub 消费者。
保持 Source 下拉菜单中的默认值 创建 Source。本演示将创建一个新的 Source 并将其添加到规则中。
为 Source 输入名称和描述（可选）。名称应该是大小写字母和数字的组合，例如 my-gcppubsub-source。
从连接器下拉框中选择刚刚创建的 my_pubsubconsumer。您也可以通过点击下拉框旁边的按钮创建一个新的连接器。关于配置参数，请参见创建连接器。
想要从 GCP Pub/Sub 消费消息到 EMQX，可以为 Source 配置以下信息：
- GCP PubSub 主题：输入主题名称（例如 my-iot-core）或完整主题路径（格式为 projects/<project-id>/topics/<topic-name>）。主题名称在已配置服务账户所属的项目中解析。要消费其他 GCP 项目中的主题，请输入完整路径，并为服务账户授予访问该主题所需的 Pub/Sub 权限。消费者订阅仍创建在服务账户所属的项目中，只有主题引用指向其他项目。
- 拉取的最大消息数：指定在单次拉取请求中从 GCP PubSub 检索的最大消息数量。实际数量可能小于指定的值。
展开高级设置，根据需要配置可选设置。详细信息请参见高级设置。
在点击创建之前，您可以点击测试连接以测试是否能成功连接到 GCP Pub/Sub 服务器。

点击创建完成 Source 的创建。Source 即被添加规则页面右侧的数据输入页签下，您同时可以看到 SQL 编辑器中的规则显示为：

sql

SELECT
  *
FROM
  "$bridges/gcppubsub:my-gcppubsub-source"

注意：如果您是初级用户，点击 SQL 示例和启用测试学习和测试 SQL 规则。

从 my-gcppubsub-source 中，规则 SQL 可以获取在下面的 GCP PubSub 到 MQTT 主题映射表中列出的 GCP PubSub 消息字段。您可以调整 SQL 进行数据处理操作。此处使用默认 SQL 即可。

字段名称	描述
`attributes`	（可选）包含字符串键值对的对象（如果有）
`message_id`	GCP Pub/Sub 分配给此消息的消息 ID
`ordering_key`	（可选）消息排序键（如果有）
`publishing_time`	GCP Pub/Sub 定义的消息时间戳
`topic`	源自 GCP Pub/Sub 的主题
`value`	（可选）消息负载（如果存在）

注意：每个 GCP Pub/Sub 到 MQTT 主题映射必须包含唯一的 GCP Pub/Sub 主题名称。即 GCP Pub/Sub 主题不能在多个映射中存在。

现在您已成功创建了 GCP Pub/Sub 消费者 Source，但消息不会直接发布到 EMQX。接下来，继续按照添加消息重发布动作的步骤来创建消息重发布动作并将其添加到规则中。

添加消息重发布动作

本节演示如何在规则中添加消息重发布动作，以转发从 GCP Pub/Sub Source 消费的消息并发布到 EMQX 主题 t/1。

选择页面右侧的动作输出页签，点击添加动作按钮，并从动作类型下拉列表中选择 消息重发布 动作。
填写消息重发布的配置：
- 主题：要发布到 MQTT 的主题，在这里输入 t/1。
- QoS：选择 0、1、2 或 ${qos}，或输入占位符从其他字段设置 QoS。在这里选择 ${qos} 意味着跟随原始消息的 QoS。
- Retain：选择 true 或 false。确定是否将消息作为保留消息发布，也可以输入占位符从其他字段设置保留消息标志。此示例中选择 false。
- Payload：设置生成转发消息 payload 的模板。留空表示转发规则输出结果。输入 ${.value} 表示仅转发 GCP Pub/Sub 消息的 payload。
  MQTT payload 模板的默认值是 ${.}，包括所有可用数据编码为 JSON 对象。例如，对于包含所有可选字段的 GCP PubSub 消息，选择 ${.} 作为模板将产生以下内容：
  json
```
{
  "attributes": {"attribute_key": "attribute_value"},
  "message_id": "1679665968238",
  "ordering_key": "my-ordering-key",
  "topic": "my-pubsub-topic",
  "publishing_time": "2023-08-18T14:15:18.470Z",
  "value": "my payload"
}
```
  可以使用点表示法获取 GCP Pub/Sub 消息的子字段。例如，${.value} 将解析为 GCP Pub/Sub 消息的值，${.attributes.h1} 将解析为 h1 消息属性键的值（如果这样的子字段存在）。缺失的值将被空字符串替换。
- MQTT 5.0 消息属性：默认禁用。详细设置，请参见添加消息重发布动作。
点击创建完成动作的创建。创建成功后，您将返回到创建规则页面，并且消息重发布动作将被添加到动作输出标签。
在规则创建页面，点击创建按钮完成整个规则的创建。

现在您已成功创建了规则，您可以在规则页面看到新创建的规则。在 Sources 标签下，您可以看到新创建的 GCP Pub/Sub 消费者 Source。

您也可以点击集成 -> Flow 设计器查看拓扑图。通过拓扑图，您可以直观地看到来自 GCP Pub/Sub 消费者 Source 的消息将通过消息重发布发布到 t/1。

测试 GCP Pub/Sub 消费者 Source 规则

按照以下步骤，验证 GCP Pub/Sub 消费者 Source 能否消费 GCP Pub/Sub 消息，并将消息重新发布到 EMQX 的 MQTT 主题 t/1。

使用 MQTTX CLI 订阅 EMQX 的 MQTT 主题 t/1：
bash
```
mqttx sub -t t/1 -v
```
在 Google Cloud 控制台中，进入 Pub/Sub -> 主题，点击主题 my-iot-core，然后发布以下消息：
json
```
{"msg":"hello GCP PubSub"}
```
确认 MQTTX 在主题 t/1 上收到以下消息：
text
```
topic: t/1
payload: {"msg":"hello GCP PubSub"}
```

高级设置

本节介绍 GCP Pub/Sub 连接器、生产者 Sink 和消费者 Source 的高级设置。

连接器高级设置

GCP Pub/Sub 生产者和消费者连接器使用相同的高级设置。

字段名称	描述	默认值
HTTP 流水线	不等待每个响应即可连续发送的 HTTP 请求上限。设置为 `1` 时，发送下一个请求前必须等待当前请求的响应。	`100`
连接池大小	连接池中维护的连接数。	`8`
连接超时	等待 HTTP 连接建立的最长时间。	`15` 秒
最大空闲时间	HTTP 客户端尝试重新连接前允许无活动的最长时间。	`10` 秒
最大重试次数	发送请求出错后的最大重试次数。	`2`
启动超时	创建连接器后，等待连接器达到健康状态的最长时间。	`5` 秒
健康检查间隔	执行连接器健康检查的时间间隔。	`15` 秒
健康检查超时	健康检查返回结果的最长时间。超时后，连接器将被视为已断开。	`60` 秒

生产者 Sink 和消费者 Source 共同高级设置

生产者 Sink 和消费者 Source 使用以下共同高级设置。两者的健康检查间隔默认值不同。

字段名称	描述	生产者 Sink 默认值	消费者 Source 默认值
请求超期	从请求进入缓冲区开始计算的最长有效期。如果请求在该时间内仍停留在缓冲区，或发送后未收到响应或确认，则请求过期。	`45` 秒	`45` 秒
健康检查间隔	执行 Sink 或 Source 健康检查的时间间隔。	`15` 秒	`30` 秒
健康检查间隔抖动	在健康检查间隔上增加均匀分布的随机延迟，使共用连接器的 Action 和 Source 不会同时开始健康检查。	`0` 毫秒	`0` 毫秒
健康检查超时	健康检查返回结果的最长时间。超时后，Sink 或 Source 将被视为已断开。	`60` 秒	`60` 秒

生产者 Sink 特有高级设置

GCP Pub/Sub 生产者 Sink 提供以下特有高级设置。

字段名称	描述	默认值
缓存池大小	将数据发送到 GCP Pub/Sub 之前，用于缓存和处理数据的缓冲工作进程数量。	`16`
分发策略	将未显式设置选择键的请求分配给缓冲工作进程的策略。`按客户端 ID` 将同一客户端的请求分配给同一工作进程；`随机` 将请求分散到不同工作进程。	`按客户端 ID`
缓存队列最大长度	每个缓冲工作进程可缓存的最大数据量。	`256` MB
最大批量请求大小	一个批次中的最大请求数。设置为 `1` 时禁用批处理。	`1`
请求模式	控制同步或异步发送请求。`异步` 模式下，EMQX 无需等待 GCP Pub/Sub 响应即可继续处理消息。	`异步`
请求飞行队列窗口	当请求模式为 `异步` 时，指定未收到响应前可发送的最大请求数。如果来自同一 MQTT 客户端的消息必须严格按顺序处理，请设置为 `1`。	`100`

消费者 Source 特有高级设置

GCP Pub/Sub 消费者 Source 提供以下特有高级设置。

字段名称	描述	默认值
消息确认截止时间	GCP Pub/Sub 等待 Source 确认收到消息的近似时长。超过该时长后，GCP Pub/Sub 可能重新投递消息。取值范围为 `10` 至 `600` 秒。	`60` 秒

Kubernetes

EMQX Operator

管理 EMQX 集群

API Reference

密码认证

监控

访问控制

集成

管理

集成 Prometheus

集成 OpenTelemetry

端到端追踪

插件目录

JT/T 808 协议网关

GB/T 32960 协议网关

火山引擎 RTC

场景样例

GPT-Realtime

将 MQTT 数据传输到 GCP Pub/Sub ​

工作原理 ​

特性与优势 ​

准备工作 ​

前置准备 ​

创建服务账号凭证 ​

配置工作负载身份联合 ​

附加服务账号前提条件 ​

在 GCP Pub/Sub 中创建主题 ​

创建 GCP Pub/Sub 生产者连接器 ​

创建 GCP Pub/Sub 生产者 Sink 规则 ​

测试 GCP Pub/Sub 生产者 Sink 规则 ​

创建 GCP Pub/Sub 消费者连接器 ​

创建 GCP Pub/Sub 消费者 Source 规则 ​

添加消息重发布动作 ​

测试 GCP Pub/Sub 消费者 Source 规则 ​

高级设置 ​

连接器高级设置 ​

生产者 Sink 和消费者 Source 共同高级设置 ​

生产者 Sink 特有高级设置 ​

消费者 Source 特有高级设置 ​