MBS Subscribe (消息订阅)

概览

字段	值
Action 名称	MBS Subscribe (消息订阅)
Connector Type	MBS
Description	在 MBS 指定 Broker 的 Queue 上注册 Webhook 订阅

注册订阅后，MBS 会将该 Queue 中的新消息通过 HTTP POST 主动推送到你指定的 Callback URL。

Input Parameters

Name	Label	Data Type	Control Type	必填	说明
`brokerName`	Broker Name	string	text	是	Broker 名称，如 `builtin`
`queueName`	Queue Name	string	text	是	队列名称
`subscriberId`	Subscriber ID	string	text	是	订阅者 ID，由客户端生成，用于标识该订阅
`callbackUrl`	Callback URL	string	text	是	消息推送的回调 URL，支持占位符
`messageAckTimeout`	ACK Timeout (秒)	number	text	否	ACK 超时时间（秒），默认 30
`retry`	Retry	number	text	否	连续失败多少次后自动退订，默认 5
`customHeaders`	Custom Headers (JSON)	string	text	否	推送时附加的自定义 HTTP Header，JSON 格式

参数详解

brokerName

MBS 中注册的 Broker 名称，与 Enqueue 中的含义相同。

queueName

要订阅的目标队列名称。

subscriberId

订阅者的唯一标识，由你自行生成。

规则：

同一个 Queue 下不允许重复的 Subscriber ID
建议使用有意义的命名，如 {服务名}-{实例ID}
示例：order-service-01、notification-handler-prod-1

callbackUrl

消息推送的目标 URL，当队列中有新消息时，MBS 会向此 URL 发送 HTTP POST 请求。

基本格式：

http://your-service:3000/webhook/messages

Callback URL 占位符

callbackUrl 和 headers 的值支持以下占位符，MBS 在推送时会自动替换为实际值：

占位符	说明	替换示例
`{BROKER_NAME}`	当前消息所属的 Broker 名称	`builtin`
`{QUEUE_NAME}`	当前消息所属的 Queue 名称	`order-events`
`{MESSAGE_ID}`	当前消息的唯一 ID	`550e8400-e29b-41d4-a716-446655440000`
`{SUBSCRIBER_ID}`	当前订阅者 ID	`my-subscriber-01`

使用示例：

http://my-service:3000/webhook/{BROKER_NAME}/{QUEUE_NAME}

如果 Broker 是 builtin，Queue 是 order-events，实际推送 URL 会被替换为：

http://my-service:3000/webhook/builtin/order-events

在 headers 中使用占位符：

{
  "callbackUrl": "http://my-service:3000/webhook",
  "headers": {
    "X-Message-Id": "{MESSAGE_ID}",
    "X-Subscriber": "{SUBSCRIBER_ID}",
    "X-Source-Queue": "{BROKER_NAME}/{QUEUE_NAME}"
  }
}

适用场景：

通过 URL 路径区分消息来源（不同 Queue 路由到不同 Handler）
在 Header 中传递消息 ID（消费者无需从固定的 X-Message-Id 获取，可自定义 Header 名）
消费者通过 Header 判断订阅者身份

messageAckTimeout

ACK 超时时间，单位为秒，默认 30 秒。

消息推送后开始计时
在超时时间内未收到 ACK 确认，消息会被重新投递
建议设置为业务处理最大耗时的 3-5 倍

场景	建议超时时间
简单数据写入	10-15 秒
一般业务处理	30 秒（默认）
调用外部接口	60-90 秒
复杂计算/批处理	120 秒以上

retry

连续推送失败次数的阈值，默认 5 次。

当 MBS 向 Callback URL 推送消息连续失败达到此次数后，订阅会被自动移除
失败定义：Callback URL 返回非 2xx 状态码，或连接超时/拒绝
设置为 0 表示不自动退订（谨慎使用，可能导致无效推送持续堆积）

customHeaders

推送消息时附加的自定义 HTTP Header，以 JSON 格式填写。

格式：

{"X-Custom-Header": "value", "X-Source": "ceta-platform"}

MBS 在向 Callback URL 推送消息时，会在请求头中附加这些自定义 Header。

应用场景：

消费者需要通过 Header 区分消息来源
消费者端需要特定的认证 Header
传递额外的元数据信息

Output Parameters

无输出参数。

Action Script

import cn.hutool.json.JSONUtil

def pathStr = 'api/broker/' + inputs.brokerName + '/queue/' + inputs.queueName + '/subscribe'

def requestBody = [:]
requestBody.put('subscriberId', inputs.subscriberId)
requestBody.put('callbackUrl', inputs.callbackUrl)

if (inputs.messageAckTimeout != null) {
  requestBody.put('messageAckTimeout', inputs.messageAckTimeout)
}
if (inputs.retry != null) {
  requestBody.put('retry', inputs.retry)
}
if (inputs.customHeaders != null && inputs.customHeaders.trim().length() > 0) {
  def hdrs = JSONUtil.toBean(inputs.customHeaders, Map.class)
  requestBody.put('headers', hdrs)
}

def result = context.http {
  path pathStr
  method "POST"
  body (requestBody)
}

if (result?.statusCode != null && result.statusCode >= 400) {
  throw new RuntimeException("MBS failed [" + result.statusCode + "]: " + result.body)
}

return result

脚本逻辑说明：

拼接 API 路径 - api/broker/{brokerName}/queue/{queueName}/subscribe
构建请求体 - 必填字段 subscriberId 和 callbackUrl 始终加入
处理可选参数 - messageAckTimeout 和 retry 仅在有值时加入请求体
解析自定义 Header - 将 customHeaders 从 JSON 字符串解析为 Map，以 headers 字段放入请求体
发送请求 - 以 JSON 格式 POST 到 MBS 订阅接口
错误处理 - HTTP 状态码 >= 400 时抛出异常

API 对应关系

此 Action 对应 MBS 的如下 REST API：

POST /api/broker/{broker_name}/queue/{queue_name}/subscribe
Content-Type: application/json

完整请求体参数（MBS API 层面）：

{
  "subscriberId": "order-service-01",
  "callbackUrl": "http://your-service.com/webhook/{BROKER_NAME}/{QUEUE_NAME}",
  "type": "application/json",
  "encoding": "utf-8",
  "messageAckTimeout": 30,
  "retry": 5,
  "headers": {
    "X-Custom-Header": "value"
  },
  "warp": {
    "field": "data",
    "encoding": "base64"
  }
}

字段	必填	默认值	说明
`subscriberId`	是	-	订阅者唯一 ID
`callbackUrl`	是	-	推送目标 URL，支持 `{BROKER_NAME}` 和 `{QUEUE_NAME}` 占位符
`type`	否	-	推送时的 Content-Type Header
`encoding`	否	-	消息体解码方式
`messageAckTimeout`	否	30	ACK 超时（秒）
`retry`	否	5	连续失败自动退订阈值
`headers`	否	-	推送时附加的自定义 HTTP Header
`warp`	否	-	消息包装配置

info

当前 Action 封装了最常用的参数。如果需要使用 type、encoding、warp 等高级参数，可以参考下方"消息包装"章节，或直接修改 Action Script 扩展支持。

消息推送格式

订阅成功后，当队列中有新消息时，MBS 会向 Callback URL 发送 HTTP POST 请求：

推送请求的结构：

部分	说明
Method	POST
URL	你配置的 callbackUrl（占位符已替换）
Header: `X-Message-Id`	消息的唯一 ID（用于后续 ACK 确认）
Header: 自定义 headers	如果配置了 customHeaders，会附加在请求中（占位符已替换）
Header: `Content-Type`	订阅时配置的 `type` 字段值，默认 `application/octet-stream`
Body	消息内容（原始格式，或经 warp 包装后的 JSON）

消费者回调响应格式（重要）

danger

消费者的 Webhook 端点必须返回符合以下条件的响应，否则 MBS 会判定为投递失败并递增连续失败计数：

HTTP 状态码 2xx（200、201、202 等）
响应体为有效 JSON 对象
JSON 中必须包含 consumer 字段（值为非空字符串）

正确的响应示例：

{"consumer": "my-service-instance-1"}

以下情况会被判定为投递失败：

情况	后果
HTTP 状态码 >= 300（如 404、500）	判定失败，递增连续失败计数
连接超时或网络错误	判定失败，递增连续失败计数
HTTP 2xx 但响应体为空	判定失败，递增连续失败计数
HTTP 2xx 但响应体不是有效 JSON	判定失败，递增连续失败计数
HTTP 2xx + 有效 JSON 但缺少 `consumer` 字段	判定失败，递增连续失败计数
HTTP 2xx + 有效 JSON + 含 `consumer` 字段	成功

错误示例（会导致失败）：

OK → "OK"                              // 不是 JSON 对象
OK → {"status": "success"}             // 缺少 consumer 字段
OK →                                   // 空响应体

重复订阅（Upsert）

使用相同的 broker_name + queue_name + subscriberId 重新调用 subscribe 接口时：

更新订阅配置（callbackUrl、headers、timeout 等）
连续失败计数重置为 0
状态恢复为 ACTIVE（如果之前是 EXPIRED）

投递模式

MBS 支持两种投递模式，在创建 Queue 时配置：

FANOUT（广播模式，默认）

每条消息推送给所有活跃订阅者
每个订阅者独立 ACK，互不影响
某个订阅者 ACK 超时只影响该订阅者的重推
适用：事件通知、数据同步

QUEUE（轮播模式）

每条消息只投递给一个订阅者（Round-Robin 轮播选择）
投递失败时自动故障转移到下一个订阅者
适用：任务分发、负载均衡

订阅状态说明

状态	说明
`ACTIVE`	正常活跃，接收消息推送
`EXPIRED`	连续失败达阈值，已自动退订
`CANCELED`	管理员手动退订

恢复方式：使用相同 subscriberId 重新订阅即可将 EXPIRED 恢复为 ACTIVE。

消息包装（Warp）

MBS 支持对原始消息进行包装后再推送。启用 warp 后，原始消息体会被编码并放入一个 JSON 结构中：

配置示例：

{
  "warp": {
    "field": "data",
    "encoding": "base64"
  }
}

推送效果：

原始消息 {"orderId": "12345"} 会被包装为：

{
  "data": "eyJvcmRlcklkIjogIjEyMzQ1In0="
}

适用场景：

消费者需要统一的 JSON 格式接收消息
原始消息可能包含特殊字符，需要 base64 编码传输
消费者希望将消息体放在特定字段中

取消订阅

取消订阅可以通过 MBS 管理后台操作，或调用管理 API：

POST /mbs/api/admin/brokers/{name}/queues/{queue}/unsubscribe/{subscriberId}

使用示例

示例 1：基本订阅

参数	值
Broker Name	`builtin`
Queue Name	`order-events`
Subscriber ID	`ceta-order-service-01`
Callback URL	`http://order-service:3000/webhook/messages`
ACK Timeout	30
Retry	5

示例 2：使用占位符的订阅

参数	值
Broker Name	`builtin`
Queue Name	`order-events`
Subscriber ID	`unified-gateway-01`
Callback URL	`http://gateway:8080/mbs/{BROKER_NAME}/{QUEUE_NAME}/receive`

实际推送 URL：http://gateway:8080/mbs/builtin/order-events/receive

参数	值
Broker Name	`kafka-prod`
Queue Name	`notifications`
Subscriber ID	`notification-handler-01`
Callback URL	`http://notify-service:3000/webhook`
ACK Timeout	60
Retry	10
Custom Headers (JSON)	`{"X-Source": "ceta-mbs", "X-Env": "production", "Authorization": "Bearer consumer-token"}`

白名单要求

caution

MBS 会对 Callback URL 进行白名单校验。创建订阅前，必须确保 Callback URL 中的主机名或 IP 已在 MBS 白名单中。否则订阅请求会被拒绝。

详见白名单配置章节。

错误处理

HTTP 状态码	错误码	原因
401	`AUTHENTICATION_ERROR`	Token 无效或已过期
403	`AUTHORIZATION_ERROR`	Token 没有该 Queue 的 read 权限
400	`INVALID_REQUEST`	请求参数格式错误（如 callbackUrl 为空）
404	`INSTANCE_NOT_FOUND`	Broker 或 Queue 不存在
403	白名单校验失败	Callback URL 的主机名/IP 不在白名单中

概览​

Input Parameters​

参数详解​

brokerName​

queueName​

subscriberId​

callbackUrl​

Callback URL 占位符​

messageAckTimeout​

retry​

customHeaders​

Output Parameters​

Action Script​

API 对应关系​

消息推送格式​

消费者回调响应格式（重要）​

重复订阅（Upsert）​

投递模式​

FANOUT（广播模式，默认）​

QUEUE（轮播模式）​

订阅状态说明​

消息包装（Warp）​

取消订阅​

使用示例​

示例 1：基本订阅​

示例 2：使用占位符的订阅​

示例 3：带自定义 Header 的订阅​

白名单要求​

错误处理​

概览