跳到主要内容

MBS ACK (消息确认)

概览​

字段值
Action 名称MBS ACK (消息确认)
Connector TypeMBS
Description确认已消费的消息，防止消息被重复投递

消费者在处理完消息后，必须调用 ACK 确认消费。未确认的消息会在超时后被 MBS 重新投递。

Input Parameters​

NameLabelData TypeControl Type必填说明
brokerNameBroker Namestringtext是Broker 名称，如 builtin
queueNameQueue Namestringtext是队列名称
messageIdMessage IDstringtext是要确认的消息 ID
consumedByConsumed Bystringtext否消费者标识

参数详解​

brokerName / queueName​

与订阅时指定的 Broker 和 Queue 保持一致。

messageId​

要确认的消息唯一 ID。该值来自 MBS 推送消息时请求头中的 X-Message-Id。

当 MBS 向你的 Callback URL 推送消息时，请求头中会包含：

X-Message-Id: 550e8400-e29b-41d4-a716-446655440000

消费者需要从推送请求的 Header 中提取这个 ID，在处理完业务逻辑后传入此参数进行 ACK。

consumedBy​

可选的消费者标识，用于记录是哪个服务实例完成了这条消息的消费。

推荐格式： {服务名}-{实例ID}

示例：order-service-instance-01、notification-handler-pod-abc123

作用：

  • 在 MBS 管理后台可以查看消息被哪个实例消费
  • 方便排查问题时定位消费节点

Output Parameters​

NameLabelData Type说明
timestampTimestampstringMBS 确认的时间戳（ISO 8601 格式）
acknowledgedByAcknowledged Bystring处理确认的 MBS 节点标识

响应示例：

{
  "timestamp": "2024-01-15T10:30:01Z",
  "acknowledgedBy": "mbs-node-1"
}

Action Script​

import java.time.Instant

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

def requestBody = [:]
requestBody.put('timestamp', Instant.now().toString())
if (inputs.consumedBy != null && inputs.consumedBy.trim().length() > 0) {
  requestBody.put('consumedBy', inputs.consumedBy)
}

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

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

return result

脚本逻辑说明：

  1. 拼接 API 路径 - api/subscribe/broker/{brokerName}/queue/{queueName}/ack/{messageId}
  2. 自动填充时间戳 - 使用 Instant.now() 记录确认时间
  3. 处理可选的 consumedBy - 仅在有值且非空时加入请求体
  4. 发送 POST 请求 - 请求体为 JSON 格式
  5. 错误处理 - HTTP 状态码 >= 400 时抛出异常

API 对应关系​

此 Action 对应 MBS 的如下 REST API：

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

请求体：

{
  "consumedBy": "your-service-instance-id",
  "timestamp": "2024-01-15T10:30:00Z"
}

ACK 超时机制详解​

工作流程​

MBS 推送消息到 Callback URL
        │
        ▼
开始计时（messageAckTimeout 秒）
        │
        ├── 在超时时间内收到 ACK ──▶ 标记消息为已消费 ✓
        │
        └── 超时未收到 ACK ──▶ 消息重新投递
                                    │
                                    ├── 推送成功，再次等待 ACK
                                    │
                                    └── 连续失败达到 retry 次 ──▶ 自动退订该订阅者

关键规则​

  1. 消息推送后立即开始计时 - 从 MBS 发出 POST 请求的那一刻起
  2. 超时 = 重新投递 - 超时后消息不会丢失，会再次推送
  3. 连续失败 = 自动退订 - 保护 MBS 不被无效订阅者拖垮
  4. "At-least-once" 语义 - 消息至少投递一次，可能多次（网络抖动、ACK 丢失等场景）

为什么可能收到重复消息​

场景说明
ACK 网络超时消费者处理成功并发送了 ACK，但 ACK 请求在网络中丢失
处理耗时超出超时时间业务逻辑还没处理完，MBS 已经认为超时并重新投递
MBS 节点切换多实例场景下节点切换可能导致重复投递
警告

因为存在重复投递的可能，消费者必须保证消息处理的幂等性。使用 X-Message-Id 做去重判断是最佳实践。

使用示例​

示例 1：基本确认​

参数值
Broker Namebuiltin
Queue Nameorder-events
Message ID550e8400-e29b-41d4-a716-446655440000
Consumed Byorder-service-instance-01

示例 2：在 Webhook 消费者中的完整流程​

// 消费者 Webhook 端点伪代码
app.post('/webhook/messages', async (req, res) => {
  // 1. 提取消息 ID
  const messageId = req.headers['x-message-id'];

  // 2. 处理业务逻辑
  const payload = req.body;
  await processOrder(payload);

  // 3. 返回 200（告诉 MBS 收到了）
  res.status(200).send('OK');

  // 4. 调用 ACK 确认消费
  //    在 Ceta Flow 中，这一步通过 MBS ACK Action 完成
  await mbsAck(brokerName, queueName, messageId, 'my-service-01');
});

错误处理​

HTTP 状态码错误码原因
401AUTHENTICATION_ERRORToken 无效或已过期
403AUTHORIZATION_ERRORToken 没有该 Queue 的 read 权限
404INSTANCE_NOT_FOUNDBroker/Queue 不存在，或 messageId 无效
400INVALID_REQUEST请求参数格式错误

重要规则​

  • broker_name 必须匹配：ACK 请求中的 broker_name 必须与消息投递时注册的 Broker 名称完全一致（区分大小写），否则返回 404
  • 不可重复 ACK：同一条消息对同一个 subscriber 只能 ACK 一次，重复 ACK 返回 404
  • ACK 成功会重置连续失败计数：每次成功 ACK 都会将该订阅者的连续失败计数重置为 0
  • consumedBy 和 timestamp 均为必填：Action Script 已自动填充 timestamp
警告

brokerName 必须与订阅时指定的完全一致（区分大小写）。如果不匹配，ACK 会返回 404，导致消息被重复投递，最终可能触发自动退订。