跳到主要内容

最佳实践与 FAQ

完整使用流程​

前置条件​

  1. MBS 服务已部署并正常运行
  2. 管理员已在 MBS 管理后台创建了 Broker 和 Queue
  3. 已获取 API Token（具有目标 Queue 的相应权限）
  4. Callback URL 对应的主机名/IP 已加入白名单

第一步：创建 MBS Connector 实例​

  1. 在 Connector 类型的 PBC 中新建 Connector
  2. 选择 ConnectorType 为 MBS
  3. 填写 Base URL（MBS 服务地址）和 Token（API 认证令牌）
  4. 保存

第二步：在 Flow 中推送消息​

在 Flow 设计器中添加 Action 节点：

  1. 选择 MBS Enqueue (消息推送) Action
  2. 关联已创建的 MBS Connector 实例
  3. 填写 Broker Name、Queue Name、Message Body
  4. 获取返回的 messageId（可选，用于追踪）

第三步：注册消息订阅​

在 Flow 中（或系统初始化时）调用：

  1. 选择 MBS Subscribe (消息订阅) Action
  2. 填写 Broker Name、Queue Name、Subscriber ID、Callback URL
  3. 按需配置 ACK Timeout 和 Retry

第四步：消费消息并 ACK​

消费者服务收到推送后：

  1. 从请求头提取 X-Message-Id
  2. 处理业务逻辑
  3. 通过 MBS ACK (消息确认) Action 确认消费

最佳实践​

1. 消费者回调响应（最重要）​

确保 Webhook 端点返回 {"consumer": "xxx"} 格式的 JSON 响应。 这是 MBS 判定推送成功的必要条件。

推荐的消费者回调流程：

收到推送 → 处理消息 → 调用 ACK 接口 → 返回 {"consumer": "xxx"}

如果返回纯文本（如 OK）、空响应、或不含 consumer 字段的 JSON，MBS 会判定推送失败并递增连续失败计数。

2. 消息幂等性（重要）​

由于 MBS 使用 "at-least-once" 投递语义，同一消息可能被推送多次。消费者必须保证消息处理的幂等性。

推荐做法：

  • 使用 X-Message-Id 做去重判断（在数据库或 Redis 中记录已处理的 messageId）
  • 业务逻辑设计为可重入（同一消息处理多次，最终结果一致）
  • 使用数据库唯一约束防止重复写入

示例去重逻辑：

app.post('/webhook', async (req, res) => {
  const messageId = req.headers['x-message-id'];

  // 检查是否已处理过
  if (await isMessageProcessed(messageId)) {
    res.status(200).send('Already processed');
    return;
  }

  // 处理业务
  await processMessage(req.body);

  // 标记为已处理
  await markMessageProcessed(messageId);

  // 返回 200
  res.status(200).send('OK');
});

2. ACK 超时设置​

原则说明
超时 > 最大处理耗时避免业务还在处理中就被判定为超时
超时 = 平均耗时 x 3~5留出网络抖动和偶发慢请求的余量
不确定时设大值设 120 秒比设 5 秒风险小

3. broker_name 一致性​

ACK 请求中的 brokerName 路径参数必须与创建订阅时使用的 Broker 名称完全一致（区分大小写）。如果不匹配：

  • ACK 返回 404
  • 消息超时后被重新投递
  • 最终触发连续失败退订

4. ACK 超时设置​

  • 快速响应 - Webhook 端点应在 5 秒内返回 HTTP 200
  • 异步处理 - 如业务逻辑耗时较长，先返回 200（表示收到），异步处理后再 ACK
  • 非 2xx = 失败 - 返回非 2xx 状态码会被 MBS 视为投递失败，触发重试

推荐模式：

收到推送 → 立即返回 200 → 异步处理业务 → 处理完成调用 ACK

4. Token 最小权限原则​

  • 每个业务服务使用独立的 Token
  • 按实际需要分配最小权限：
    • 只需发消息的服务：仅 write 权限
    • 只需消费消息的服务：仅 read 权限
  • 定期轮换 Token

5. Subscriber ID 命名规范​

格式示例适用场景
{服务名}-{实例号}order-service-01多实例部署的服务
{服务名}-{环境}order-service-prod单实例服务
{服务名}-{Pod名}order-service-pod-abc123K8s 环境

6. Callback URL 设计​

推荐使用占位符实现路由分发：

http://gateway:8080/mbs-webhook/{BROKER_NAME}/{QUEUE_NAME}

这样一个消费者服务可以同时订阅多个 Queue，通过 URL 路径区分消息来源，在内部路由到不同的 Handler。

7. 错误处理​

生产者端（Enqueue 失败）：

  1. 记录失败日志，包含消息内容和错误信息
  2. 根据业务需求决定是否重试
  3. 重试时注意幂等性（避免重复发送造成重复消费）

消费者端（处理失败）：

  1. 不要 ACK，让 MBS 在超时后自动重新投递
  2. 如果是明确不可恢复的错误（如消息格式错误），仍然 ACK 并记录到死信日志
  3. 避免让错误消息一直重试导致后续消息堆积

MBS 管理后台​

访问地址​

http://mbs-host:8080/mbs/ui/

登录​

使用管理员密码登录（默认 admin，可通过 MBS_ADMIN_PASSWORD 环境变量修改）。

管理功能​

功能说明
Broker 管理查看/新建/编辑 Broker，监控连接状态
Queue 管理查看/新建 Queue，查看队列的订阅者列表
Token 管理创建/编辑/删除 Token，配置权限
白名单配置管理 Callback URL 白名单
健康状态查看各 Broker 后端的连接健康状况

API 错误码参考​

错误码HTTP 状态码说明解决方案
AUTHENTICATION_ERROR401Token 缺失或无效检查 Connector 中配置的 Token
AUTHORIZATION_ERROR403无对应权限联系管理员为 Token 添加权限
INSTANCE_NOT_FOUND404Broker 不存在确认名称拼写，或联系管理员创建
MESSAGE_NOT_FOUND404消息不存在或已确认消息可能已被 ACK 或已超时清理
INSTANCE_UNAVAILABLE503Broker 连接不可用联系管理员检查 Broker 后端状态
INVALID_REQUEST400请求格式错误检查参数是否完整、格式是否正确
PAYLOAD_TOO_LARGE413消息体超过大小限制缩小消息体，或联系管理员调整限制

常见问题​

Q: Enqueue 返回 401 错误？​

A: Token 无效或已过期。解决方案：

  1. 在 MBS 管理后台确认 Token 是否存在且有效
  2. 重新获取 Token 并更新 Connector 配置

Q: Enqueue 返回 403 错误？​

A: Token 没有对应 Queue 的 write 权限。解决方案：

  1. 在 MBS 管理后台编辑 Token 的权限配置
  2. 为目标 {brokerName}/{queueName} 添加 write 权限

Q: Subscribe 报白名单校验失败？​

A: Callback URL 的主机名/IP 不在白名单中。解决方案：

  1. 联系管理员在 MBS 管理后台添加白名单条目
  2. 确认主机名格式正确（不带协议、端口、路径）

Q: 消息推送后消费者没有收到？​

排查步骤：

  1. 确认消费者服务正常运行且 Callback URL 可达
  2. 确认 Callback URL 对应的端口已开放
  3. 检查 MBS 管理后台中该订阅是否仍然有效（是否因连续失败被自动退订）
  4. 检查消费者服务日志是否有异常

Q: 同一条消息被重复推送？​

A: ACK 未在超时时间内完成。解决方案：

  1. 增大 Subscribe 中的 messageAckTimeout 值
  2. 优化消费者处理速度
  3. 改为异步处理模式（先返回 200，再异步 ACK）
  4. 确保消费者逻辑具有幂等性

Q: 如何查看消息是否推送成功？​

A: 通过 MBS 管理后台查看：

  1. 进入目标 Queue 的详情页
  2. 查看订阅者列表中的投递状态和最近活动时间
  3. 如果有投递失败，会显示失败次数和最近失败原因

Q: 订阅被自动退订了怎么办？​

A: 当连续推送失败达到 retry 次数后，订阅会被自动移除。解决方案：

  1. 检查并修复消费者服务的问题
  2. 重新调用 Subscribe Action 创建订阅
  3. 考虑增大 retry 值（但不建议设为 0）

Q: 如何在 Flow 中同时发送消息和等待 ACK？​

A: 推荐分离生产者和消费者的逻辑：

  • 生产者 Flow - 只负责调用 Enqueue 推送消息
  • 消费者服务 - 独立的 Webhook 端点，收到消息后处理并 ACK
  • 两者通过 MBS 解耦，无需在同一个 Flow 中等待

Q: 支持哪些消息格式？​

A: MBS 对消息格式不做限制。消息体以二进制方式存储和投递，支持：

  • JSON
  • XML
  • 纯文本
  • 二进制数据
  • 任意自定义格式

消费者需要自行按约定格式解析消息内容。

Q: 如何选择 FANOUT 还是 QUEUE 投递模式？​

A:

  • FANOUT（广播）：所有订阅者都需要收到每条消息（事件通知、数据同步）
  • QUEUE（轮播）：每条消息只需要一个消费者处理（任务分发、负载均衡）

FANOUT 模式下每个订阅者独立投递、独立 ACK，某个订阅者失败不影响其他订阅者。

Q: CETA 流程接收 MBS 推送后，连续失败计数一直为 1？​

A: 两个最常见的原因：

  1. 回调流程未设置 inMemory: true - 非内存模式下流程触发后立即返回，响应中不包含 consumer 字段。解决：将 TRIGGER 节点的 inMemory 设为 true。
  2. 流程中未设置 consumer 输出值 - 即使是内存执行，没有通过"设置流程输出值"节点设置 consumer 字段也不行。解决：添加输出值节点。

详见 Flow 集成指南。

Q: MBS 的健康检查接口是什么？​

A:

  • 存活探针：GET /health - 返回 200 表示进程存活
  • 就绪探针：GET /readiness - 返回 200 表示数据库和 Broker 连接正常

这两个端点无需认证。