设备消息广播（Bcast）

emqx_bcast 插件支持后端应用通过 HTTP 向 IoT 设备发送命令、通知和其他消息。它提供产品级广播、定向批量投递和 QoS 1 离线重放能力，使应用无需维护 MQTT 连接，也无需自行构建临时离线投递队列和 MQTT PUBACK 跟踪系统。应用还可以预先注册一次 Payload，并在后续批量请求中复用其消息 ID。

该插件从 EMQX 企业版 6.3.3 开始可用。

典型场景包括向同一产品下的设备广播服务通知或配置刷新信号、向指定设备发送命令或配置更新，以及在暂时离线的设备重新连接时投递 QoS 1 命令。

快速开始

本示例将 Hello 广播给产品 P1 下的一个在线设备。开始前，请确保 EMQX 企业版 6.3.3 或更高版本已运行，并已安装用于建立设备连接的 MQTTX CLI。

安装并启动插件

本快速开始使用集群管理插件方式。从下载一节获取安装包，将其复制到 EMQX 插件目录，并将 <version> 替换为下载的插件版本：

bash

cp emqx_bcast-<version>.tar.gz "$EMQX_HOME/plugins/"
emqx ctl plugins install emqx_bcast-<version> --cluster
emqx ctl plugins start emqx_bcast-<version>

单节点部署无需添加 --cluster。也可以通过 Dashboard 安装并启动插件。两种方式的详细步骤以及通过 Dashboard 安装所需的授权，请参阅安装和管理插件。

配置设备标识映射

插件将每个 MQTT 客户端映射为以下标识：

ProductKey：客户端的 client_attrs.tns 命名空间属性。
DeviceName：MQTT 客户端 ID。

在 Dashboard 中进入管理 -> MQTT 配置 -> 客户端属性，添加属性 tns，并将其表达式设置为 nth(1, tokens(username, '-'))。对应的配置如下，它从采用 ProductKey-DeviceName 格式的用户名中提取第一段作为命名空间：

hocon

mqtt.client_attrs_init = [
  { expression = "nth(1, tokens(username, '-'))", set_as_attr = tns }
]

使用该配置时，以用户名 P1-device001、客户端 ID device001 连接的客户端，其 ProductKey 为 P1，DeviceName 为 device001。

有关设置客户端属性的更多信息，请参阅设置客户端属性。

应用该配置后，请重新连接客户端，以便 EMQX 在客户端连接时初始化该属性。

创建 API 密钥

在 Dashboard 中进入系统设置 -> API 密钥，创建一个角色为管理员的 API 密钥。保持命名空间开关关闭以创建全局密钥。选择自定义受限权限，然后选择 plugin_api 权限范围。请立即保存 API Key 和 Secret Key。详细步骤请参阅创建 API 密钥。

plugin_api 权限范围用于访问 /api/v5/plugin_api/emqx_bcast/* 下的 Bcast 运行时端点，不允许访问 /api/v5/plugins* 下的插件管理端点；这些端点需要 system 权限范围。

连接设备并订阅主题

运行以下命令，以用户名 P1-device001 连接客户端 device001，并订阅产品 P1 的默认广播主题：

bash

mqttx sub -h 127.0.0.1 -p 1883 \
  -i device001 \
  -u P1-device001 \
  -t '/sys/broadcast/P1' \
  -q 1 -v

如果 MQTT 监听器启用了认证，请添加对应的密码和 TLS 选项。若要通过默认主题接收 BatchPub 消息，请改为订阅 /P1/device001/user/get。

目标客户端必须订阅投递主题。客户端仅保持连接并不足以接收消息。

发送并验证广播消息

保持 MQTTX 客户端运行。在另一个终端中，将 <api_key> 和 <api_secret> 替换为上一步创建的凭证，然后运行：

bash

curl -u '<api_key>:<api_secret>' \
  -X POST 'http://127.0.0.1:18083/api/v5/plugin_api/emqx_bcast/pub' \
  -H 'Content-Type: application/json' \
  -d '{"Action":"PubBroadcast","ProductKey":"P1","MessageContent":"SGVsbG8="}'

MessageContent 必须使用 base64 编码。本示例中的 SGVsbG8= 解码后为 Hello。

EMQX 返回 HTTP 200，响应类似于：

json

{
  "Success": true,
  "RequestId": "550e8400-e29b-41d4-a716-446655440000",
  "MessageId": "550e8400-e29b-41d4-a716-446655440000"
}

MQTTX 客户端将在主题 /sys/broadcast/P1 上收到 Payload Hello。

HTTP 200 表示 EMQX 已接受请求，不表示消息已经投递。各 QoS 等级的投递保证请参阅投递语义。

重要提示

Bcast 将消息直接发送到客户端通道。这些消息会绕过发布鉴权和常规 MQTT 发布流水线，包括规则引擎、Schema 验证、消息转换、保留消息和延迟发布。

仅向可信调用方授予 Bcast API 访问权限，并在调用方应用中验证 Payload 和目标标识。

消息操作

所有消息操作都使用以下端点：

http

POST /api/v5/plugin_api/emqx_bcast/pub

在 JSON 请求体中设置 Action 字段以选择具体操作。MessageContent 必须使用 base64 编码。

向在线设备广播

PubBroadcast 向一个产品下所有在线且已订阅投递主题的设备发送消息。该操作不会存储消息，因此离线设备无法收到消息。

json

{
  "Action": "PubBroadcast",
  "ProductKey": "P1",
  "MessageContent": "SGVsbG8="
}

插件会展开已配置的 broadcast_topic 模板，其默认值为 /sys/broadcast/${productKey}。可以通过 TopicFullName 指定一个具体的自定义主题。TopicFullName 不允许包含通配符或模板占位符。

解码后的 Payload 默认上限为 64 KiB。

向设备列表发送消息

BatchPub 向明确指定的设备列表发送消息。单个请求最多可以包含 max_device_count 个不重复的设备名称。默认上限为 10,000。

json

{
  "Action": "BatchPub",
  "ProductKey": "P1",
  "DeviceName": ["device001", "device002"],
  "MessageContent": "SGVsbG8=",
  "Qos": 1
}

必须且只能使用以下一种 Payload 字段：

MessageContent：内联的 base64 编码 Payload。解码后的 Payload 默认上限为 10 KiB。
MessageId：由 RegisterMessage 或之前的 QoS 1 批量请求创建的 Payload ID。

对于 BatchPub，投递主题按以下优先级确定：

TopicTemplateName，并展开 ${productKey} 和 ${deviceName}。
TopicShortName，展开为 /${productKey}/${deviceName}/user/${TopicShortName}。
已配置的 batch_topic，默认值为 /${productKey}/${deviceName}/user/get。

注册并复用消息

RegisterMessage 存储 Payload，以供后续 BatchPub 请求复用：

json

{
  "Action": "RegisterMessage",
  "MessageContent": "SGVsbG8="
}

响应中包含 MessageId。重复注册相同内容会返回同一个 ID，并刷新其 TTL。已存储的 Payload 不可修改。

如需刷新已有消息的 TTL，请提交其 ID，而不是消息内容：

json

{
  "Action": "RegisterMessage",
  "MessageId": "550e8400-e29b-41d4-a716-446655440000"
}

MessageContent 和 MessageId 互斥。

投递语义

BatchPub 的 Qos 字段接受 0 或 1，默认值为 0。

请求 QoS	在线设备	离线设备	存储与确认
`0`	目标在线并已订阅时投递。	跳过。	不存储，也不跟踪确认。
`1`	投递消息；订阅 QoS 为 1 时跟踪消息直至收到 PUBACK。	设备重新连接并订阅后重放。	待投递消息在 `msg_ttl` 后过期，默认 TTL 为 15 天。

实际投递 QoS 取请求 QoS 和目标订阅 QoS 中的较小值。如果 QoS 1 请求发送到 QoS 0 订阅，插件在将消息交给客户端通道后即视为投递完成。

PUBACK 仅确认设备在 MQTT 协议层面收到消息，不表示设备应用已经处理消息或执行命令。

QoS 1 批量投递采用异步的至少一次语义。客户端必须能够容忍重复消息，尤其是在重新连接和会话接管期间。HTTP 200 成功响应仅表示 EMQX 已接受请求并准备异步处理，并不表示消息已经存储或投递。如果接收请求的节点在存储完成前发生故障，该请求可能丢失。

待处理的 QoS 1 状态以内存方式复制到多个核心节点，因此可以承受单个核心节点故障，但无法在整个集群重启后保留。生产端必须能够在集群完全重启后重放消息。

QoS 0 消息在会话接管窗口中也可能重复。

配置插件

下表列出了生效的配置项及其默认值：

配置项	默认值	描述
`broadcast_topic`	`/sys/broadcast/${productKey}`	`PubBroadcast` 的主题模板。
`batch_topic`	`/${productKey}/${deviceName}/user/get`	`BatchPub` 的默认主题模板。
`msg_ttl`	`15d`	已注册消息和待处理 QoS 1 投递的 TTL。修改此配置只影响配置更新后创建的记录。
`cleanup_interval`	`60s`	清理过期记录的间隔。
`max_device_count`	`10000`	单个 `BatchPub` 请求允许包含的最大设备数。
`max_message_size_broadcast`	`65536`	广播消息解码后的最大字节数。
`max_message_size_batch`	`10240`	批量消息解码后的最大字节数。
`max_pending_deliveries`	`10000000`	集群范围内待处理 QoS 1 投递数上限。
`max_pending_deliveries_per_device`	`100`	单个设备待处理 QoS 1 投递数上限。配置值会被限制在 10 到 200 之间。启动或故障切换期间，此限制以尽力而为方式执行。
`delivery_pool_size`	`0`	每个投递池的工作进程数。`0` 表示自动为每个 Erlang 调度器配置一个工作进程。修改该配置会重启投递池。

配置示例：

hocon

broadcast_topic = "/sys/broadcast/${productKey}"
batch_topic = "/${productKey}/${deviceName}/user/get"
msg_ttl = 15d
cleanup_interval = 60s
max_device_count = 10000
max_message_size_broadcast = 65536
max_message_size_batch = 10240
max_pending_deliveries = 10000000
max_pending_deliveries_per_device = 100
delivery_pool_size = 0

使用标准插件配置 API 更新这些配置：

http

PUT /api/v5/plugins/<name-vsn>/config

有关安装、启动和配置步骤，请参阅安装和管理插件。/api/v5/plugins* 下的 REST API 操作需要 system 权限范围。对于节点固定插件，仍可通过 REST API 更新配置，但不能通过 REST API 管理其生命周期。

调用 API

使用 EMQX API 密钥和 Secret 进行 HTTP Basic 认证：

bash

curl -u '<api_key>:<api_secret>' \
  -X POST 'http://127.0.0.1:18083/api/v5/plugin_api/emqx_bcast/pub' \
  -H 'Content-Type: application/json' \
  -d '{"Action":"PubBroadcast","ProductKey":"P1","MessageContent":"SGVsbG8="}'

Dashboard 用户或 API 密钥必须具有 plugin_api 或 system 权限范围，其角色和命名空间权限也必须允许所请求的 HTTP 方法。插件 API 网关接受 system 权限范围，但 Bcast 运行时调用应优先使用受限的 plugin_api 权限范围。plugin_api 权限范围不允许访问 /api/v5/plugins* 下的插件生命周期端点。

plugin_api 权限范围允许访问通过插件 API 网关发布的所有端点。对于 Bcast，这些端点包括消息和投递删除操作以及指标重置操作，因此请在检查所有公开操作后再分配该权限范围。

请求成功时返回 HTTP 200：

json

{
  "Success": true,
  "RequestId": "550e8400-e29b-41d4-a716-446655440000",
  "MessageId": "550e8400-e29b-41d4-a716-446655440000"
}

错误响应采用以下结构：

json

{
  "Success": false,
  "RequestId": "550e8400-e29b-41d4-a716-446655440000",
  "Code": "InvalidBase64",
  "ErrorMessage": "Invalid Base64 encoding"
}

常见错误包括：

HTTP 状态码	Code	原因
`400`	`InvalidProductKey`、`InvalidDeviceName`、`DuplicateDeviceName`	产品或设备标识缺失或无效，或者设备列表包含重复项。
`400`	`DeviceCountExceeded`、`MessageTooLarge`、`InvalidBase64`	请求超过配置限制，或包含无效的 base64 内容。
`400`	`InvalidTopicTemplate`、`InvalidQos`	主题字段无效，或 `Qos` 不是 `0` 或 `1`。
`400`	`MessageNotFound`、`MessageIdContentConflict`、`InvalidMessageId`	引用的消息不存在、Payload 字段冲突，或 ID 无效。
`429`	`QuotaExceeded`	已达到全局或单设备待投递数量上限。
`429`	`Busy`	本地 QoS 1 接入队列已满，请稍后重试。

管理消息和投递

管理端点仅公开元数据，不会返回消息 Payload 内容。

方法	端点	描述
`GET`	`/api/v5/plugin_api/emqx_bcast/messages?limit=100&cursor=<cursor>`	列出已注册的消息。`limit` 的取值范围为 1 到 1000。
`GET`	`/api/v5/plugin_api/emqx_bcast/messages/:messageId`	获取消息元数据及其未完成投递数。
`DELETE`	`/api/v5/plugin_api/emqx_bcast/messages/:messageId`	删除消息及其关联的待处理投递。
`GET`	`/api/v5/plugin_api/emqx_bcast/deliveries?product_key=<key>&device_name=<name>`	列出一个设备的待处理投递。两个查询参数都必须提供。
`GET`	`/api/v5/plugin_api/emqx_bcast/deliveries/:deliveryId`	获取一条待处理投递。
`DELETE`	`/api/v5/plugin_api/emqx_bcast/deliveries/:deliveryId`	删除一条待处理投递。

删除操作无法撤回已经进入投递过程的消息。删除消息需要所有运行中的核心节点参与。如果某个核心节点不可用或版本不兼容，请求会返回 500 InternalError，并且不会删除任何记录。

监控 Bcast

插件通过以下端点公开 Prometheus 文本格式指标：

http

GET /api/v5/plugin_api/emqx_bcast/metrics

这些指标属于节点本地指标。请抓取每个 EMQX 节点，并在集群范围内聚合指标值。QoS 1 计数器最终满足以下关系：

text

wanted = acked + auto_acked + ttl_expired + canceled + queued + inflight

对应的指标名称使用 bcast_batch_pub_qos1_ 前缀。可以通过 queued 和 inflight gauge 观察实时积压，通过 redelivered counter 观察重复投递尝试。

以下端点会重置所有运行节点上的指标：

http

POST /api/v5/plugin_api/emqx_bcast/metrics/reset

仅在维护或测试窗口中使用该操作。只要任一节点仍有排队中或投递中的消息，该端点就会返回 409 PendingDeliveries。如果部分节点完成指标重置后操作失败，该端点将返回 500 PartialReset，并指出受影响的节点。

使用限制

Bcast 不支持监听器挂载点或 namespace_as_mountpoint。这些功能会为存储的订阅过滤器添加前缀，导致其无法与 Bcast 未挂载的投递主题匹配。
滚动升级期间，连接到尚未运行该插件的节点的客户端无法接收待处理的 Bcast 投递。请先启动插件，并确认 GET /api/v5/plugin_api/emqx_bcast/metrics 返回 200，再将 MQTT 或 Bcast API 流量转发到该节点。

下载

各 EMQX 版本对应的插件安装包：

EMQX 版本	插件版本	安装包
6.3.3	0.4.4	emqx_bcast-0.4.4.tar.gz (sha256)

集群负载均衡

性能与调优

网络与 TLS

认证

密码认证

授权

单点登录 (SSO)

监控

访问控制

集成

管理

集成 Prometheus

集成 OpenTelemetry

端到端追踪

插件目录

设备消息广播（Bcast）

快速开始

安装并启动插件

配置设备标识映射

创建 API 密钥

连接设备并订阅主题

发送并验证广播消息

消息操作

向在线设备广播

向设备列表发送消息

注册并复用消息

投递语义

配置插件

调用 API

管理消息和投递

监控 Bcast

使用限制

下载

密码认证

端到端追踪

设备消息广播（Bcast） ​

快速开始 ​

安装并启动插件 ​

配置设备标识映射 ​

创建 API 密钥 ​

连接设备并订阅主题 ​

发送并验证广播消息 ​

消息操作 ​

向在线设备广播 ​

向设备列表发送消息 ​

注册并复用消息 ​

投递语义 ​

配置插件 ​

调用 API ​

管理消息和投递 ​

监控 Bcast ​

使用限制 ​

下载 ​

设备消息广播（Bcast）

快速开始

安装并启动插件

配置设备标识映射

创建 API 密钥

连接设备并订阅主题

发送并验证广播消息

消息操作

向在线设备广播

向设备列表发送消息

注册并复用消息

投递语义

配置插件

调用 API

管理消息和投递

监控 Bcast

使用限制

下载