同步请求

emqx_sync_request 插件允许 HTTP 调用方通过 EMQX REST API 发布一条 MQTT 请求，并同步等待第一条匹配的 MQTT 响应。

当基于 HTTP 的后端服务需要向已连接的 MQTT 客户端发送命令或查询，并在同一个 HTTP 请求中获取结果时，可以使用该插件。插件负责请求投递、响应关联、超时处理和并发请求管理，因此 HTTP 调用方无需运行自己的 MQTT 客户端，也无需自行实现 MQTT 请求/响应跟踪逻辑。

工作原理

该插件通过插件 API 网关暴露运行时 API：

POST /api/v5/plugin_api/emqx_sync_request/request

EMQX 收到 HTTP 请求后，插件会查找请求主题对应的在线 MQTT 订阅者，将 MQTT 请求直接投递给该订阅者，并等待匹配的响应消息。对于 MQTT 5 响应方，响应必须同时匹配响应主题和使用 request_id 作为值的 Correlation Data。对于不支持 Correlation Data 的 MQTT 3 响应方，响应会按响应主题的请求顺序进行匹配。

请求主题必须精确匹配一个在线的非共享订阅者：

• 通配符主题过滤器不会被匹配为请求接收方。
• 共享订阅不会被接受为请求接收方。
• 如果没有在线的精确订阅者，API 返回 404 NO_SUBSCRIBERS。
• 如果请求主题存在共享订阅，或存在多个精确订阅者，API 返回 409 CONFLICT。

投递语义

该插件仅在本地节点内存中保存正在处理的请求。它不会持久化请求，不会订阅响应主题，也不会修改 MQTT Payload。

请求消息通过直接会话投递发送给单个精确订阅者，不经过普通 MQTT 发布流水线。因此，请求消息不会被规则引擎、Schema 验证、消息转换、保留消息或延迟发布处理，也不会使用通用 /publish 路径。

HTTP 等待超时时间是远程调度和本地等待 MQTT 响应共享的同一个截止时间。远程调度耗时会计入同一个超时时间，而不会再叠加一次完整等待。

匹配响应通过投递请求的节点上的 broker message.publish 钩子观察。响应方应从连接到该节点的客户端发布响应，通常就是接收请求的同一个连接。从其他节点发布的响应不会被匹配。

配置

字段	默认值	描述
default_timeout	10s	请求体未指定 timeout 时使用的默认 HTTP 等待超时时间。
max_timeout	60s	单个请求允许的最大 timeout。
max_inflight_requests	10000	单个节点上可同时等待响应的本地 HTTP 请求数上限。
max_payload_size	64KB	MQTT 请求 Payload 和 MQTT 响应 Payload 的最大大小。

配置示例：

default_timeout = "10s"
max_timeout = "60s"
max_inflight_requests = 10000
max_payload_size = "64KB"

通过标准插件配置 API 更新插件配置：

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

HTTP API

该 API 使用与其他 EMQX 管理 API 相同的认证方式。通过 Dashboard 登录获取的 Bearer Token 可以访问该 API。API 密钥必须通过 HTTP Basic 认证发送，并且需要具备 publish 权限范围。

请求体

{
  "timeout": "5s",
  "request": {
    "topic": "devices/1001/request",
    "response_topic": "devices/1001/response",
    "request_id": "request-id-1",
    "qos": 0,
    "payload_encoding": "plain",
    "payload": "{\"cmd\":\"reboot\"}",
    "content_type": "application/json"
  }
}

字段	类型	是否必填	默认值	描述
timeout	duration string	否	default_timeout	等待匹配 MQTT 响应的最长时间。该值必须大于 0，且不超过 max_timeout。示例：100ms、5s、1m。
request	object	是	-	MQTT 请求参数。

request 对象包含以下字段：

字段	类型	是否必填	默认值	描述
topic	string	是	-	MQTT 请求主题。该字段必须是主题名，而不是主题过滤器，因此不允许包含 + 和 #。该主题必须有且仅有一个在线的非共享订阅者。
response_topic	string	是	-	MQTT 响应主题。该字段也必须是不包含 + 或 # 的主题名。
request_id	string	是	-	普通字符串，用作 MQTT 5 Correlation Data，并会在 HTTP 响应中返回。最大长度为 128 字节。
qos	integer	否	0	请求消息的 MQTT QoS。允许值为 0、1 和 2。
payload_encoding	string	否	plain	请求 Payload 编码。允许值为 plain 和 base64。
payload	string	是	-	请求 Payload。使用 plain 时，该字符串的字节会作为 MQTT Payload。使用 base64 时，该值必须是有效的 base64，解码后的字节会作为 MQTT Payload。MQTT Payload 不得超过 max_payload_size。
content_type	string	否	-	请求消息的 MQTT 5 Content Type。MQTT 3 客户端不会收到该属性。

成功响应

请求成功时返回 HTTP 200。MQTT 响应 Payload 始终以 base64 形式返回。

{
  "code": "OK",
  "message": "OK",
  "response": {
    "topic": "devices/1001/response",
    "request_id": "request-id-1",
    "payload_encoding": "base64",
    "payload": "eyJyZXN1bHQiOiJvayJ9",
    "content_type": "application/json"
  }
}

字段	描述
code	固定为 OK。
message	固定为 OK。
response.topic	MQTT 响应主题。
response.request_id	HTTP 请求中的 request_id。
response.payload_encoding	固定为 base64。
response.payload	base64 编码后的 MQTT 响应 Payload。
response.content_type	可选字段。响应 PUBLISH 中的 MQTT 5 Content Type。如果响应方未发送该属性，包括 MQTT 3 响应方，该字段会被省略。

错误响应

错误响应使用与其他 EMQX 管理 API 相同的 code 和 message 结构。

HTTP 状态码	Code	含义
400	BAD_REQUEST	JSON 请求体无效、字段值无效、请求 Payload 过大，或 MQTT 响应 Payload 过大。
401	BAD_API_KEY_OR_SECRET	API 密钥认证失败。由 EMQX 管理 API 认证逻辑返回。
403	UNAUTHORIZED_ROLE	API 密钥无权调用该 API。由 EMQX 管理 API 授权逻辑返回。
404	NO_SUBSCRIBERS	请求主题没有在线的精确非共享订阅者。通配符订阅者会被忽略。
409	CONFLICT	请求主题存在共享订阅，或存在多个精确订阅者。
429	TOO_MANY_REQUESTS	本地节点已有 max_inflight_requests 个 HTTP 请求正在等待响应。
503	SERVICE_UNAVAILABLE	未能将请求调度到订阅者所在节点。
504	TIMEOUT	等待匹配 MQTT 响应超时。
500	INTERNAL_ERROR	非预期的服务端错误。

运维诊断

该插件提供节点本地诊断 CLI 命令：

emqx ctl sync_request status

输出示例：

Counters since plugin start:
sync_request.requests.total: 42
sync_request.requests.succeeded: 39
sync_request.requests.failed: 3
sync_request.requests.bad_request: 1
sync_request.requests.no_subscribers: 1
sync_request.requests.conflict: 0
sync_request.requests.too_many_requests: 0
sync_request.requests.dispatch_failed: 0
sync_request.requests.timeout: 1
sync_request.requests.internal_error: 0

Current gauges:
sync_request.inflight_requests: 0
sync_request.pending_responses: 0

这些值不是集群范围聚合结果。该命令只读取其运行节点上的数据。在集群中，应在可能接收 HTTP 请求或投递 MQTT 响应的每个节点上运行该命令。

只有到达插件处理器的请求才会被计数。管理 API 的认证和授权失败会先由 EMQX 处理，不会进入插件。

指标	类型	范围	描述
sync_request.requests.total	counter	节点本地	本节点处理的 HTTP 同步请求次数。
sync_request.requests.succeeded	counter	节点本地	返回 HTTP 200 的请求数。
sync_request.requests.failed	counter	节点本地	返回非 200 HTTP 状态码的请求数。
sync_request.requests.bad_request	counter	节点本地	因 400 BAD_REQUEST 被拒绝的请求数。
sync_request.requests.no_subscribers	counter	节点本地	因没有在线的精确非共享订阅者而被拒绝的请求数。
sync_request.requests.conflict	counter	节点本地	因请求主题匹配多个订阅者或共享订阅者而被拒绝的请求数。
sync_request.requests.too_many_requests	counter	节点本地	因本节点达到 max_inflight_requests 而被拒绝的请求数。
sync_request.requests.dispatch_failed	counter	节点本地	未能调度到订阅者所在节点的请求数。
sync_request.requests.timeout	counter	节点本地	等待匹配 MQTT 响应超时的请求数。
sync_request.requests.internal_error	counter	节点本地	因非预期内部错误失败的请求数。
sync_request.inflight_requests	gauge	节点本地	本节点当前正在等待 MQTT 响应的 HTTP 请求数。
sync_request.pending_responses	gauge	节点本地	请求投递后创建的本地待响应注册数量。

下载

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

EMQX 版本	插件版本	安装包
6.1.4	0.1.0	emqx_sync_request-0.1.0.tar.gz (sha256)