快速开始

本章节以一个完整的示例，演示如何在 Ceta 平台中使用 MBS Connector 完成消息的发送、订阅和确认。

前置条件

• MBS 服务已部署并运行（如 http://10.0.1.50:8080）
• 已在 MBS 管理后台创建了 Broker（如 builtin）和 Queue（如 order-events）
• 已获取 API Token（具有 builtin/order-events 的 read + write 权限）
• 消费者 Callback URL 对应的地址已加入 MBS 白名单

第一步：创建 MBS Connector 实例

1. 进入 Ceta 管理后台，找到 Connector 类型的 PBC
2. 点击新建 Connector，选择 ConnectorType 为 MBS
3. 填写参数：

参数	示例值
Base URL	http://10.0.1.50:8080
Token	your-mbs-api-token

4. 保存

第二步：在 Flow 中发送消息

在 Flow 设计器中：

1. 添加一个 APP 节点
2. 选择 Connector 为刚创建的 MBS Connector 实例
3. 选择 Action 为 MBS Enqueue (消息推送)
4. 填写 Input Parameters：

参数	值
Broker Name	builtin
Queue Name	order-events
Message Body	{"event": "order_created", "orderId": "ORD-2024-001", "amount": 299.9}

5. 如需获取返回的 messageId，可在后续节点中通过 Output 变量引用

第三步：注册消息订阅

在需要注册订阅的 Flow（通常是系统初始化 Flow）中：

1. 添加一个 APP 节点
2. 选择 Action 为 MBS Subscribe (消息订阅)
3. 填写 Input Parameters：

参数	值
Broker Name	builtin
Queue Name	order-events
Subscriber ID	ceta-order-handler-01
Callback URL	http://my-service:3000/webhook/{BROKER_NAME}/{QUEUE_NAME}
ACK Timeout (秒)	30
Retry	5

info

Callback URL 中的 {BROKER_NAME} 和 {QUEUE_NAME} 是占位符，MBS 推送时会自动替换为实际的 Broker 名称和 Queue 名称。上例中实际推送 URL 为 http://my-service:3000/webhook/builtin/order-events。

第四步：消费者处理消息

你的 Webhook 消费者服务收到 MBS 推送后：

MBS 推送的请求格式：

POST http://my-service:3000/webhook/builtin/order-events
Headers:
  X-Message-Id: 550e8400-e29b-41d4-a716-446655440000
  Content-Type: application/octet-stream
Body:
  {"event": "order_created", "orderId": "ORD-2024-001", "amount": 299.9}

消费者处理流程：

app.post('/webhook/:broker/:queue', async (req, res) => {
  const messageId = req.headers['x-message-id'];
  const broker = req.params.broker;     // "builtin"
  const queue = req.params.queue;       // "order-events"
  const payload = req.body.toString('utf-8');

  console.log(`收到消息 [${broker}/${queue}]: ${messageId}`);

  // 1. 先返回 200（告诉 MBS 消息已收到）
  res.status(200).send('OK');

  // 2. 异步处理业务逻辑
  try {
    await processOrder(JSON.parse(payload));

    // 3. 处理完成后调用 ACK
    await ackMessage(broker, queue, messageId);
  } catch (error) {
    console.error('处理失败，等待 MBS 重新投递:', error);
    // 不 ACK，MBS 会在超时后重新投递
  }
});

第五步：确认消息消费

在消费者的 ACK 逻辑中，如果消费者也是 Ceta Flow，可以在 Flow 中添加 ACK 节点：

1. 添加一个 APP 节点
2. 选择 Action 为 MBS ACK (消息确认)
3. 填写 Input Parameters：

参数	值
Broker Name	builtin
Queue Name	order-events
Message ID	${messageId}（从推送请求的 X-Message-Id Header 获取）
Consumed By	ceta-order-handler-01

如果消费者是外部服务，可以直接调用 MBS ACK API：

curl -X POST http://10.0.1.50:8080/api/subscribe/broker/builtin/queue/order-events/ack/550e8400-e29b-41d4-a716-446655440000 \
  -H "Authorization: Bearer your-mbs-api-token" \
  -H "Content-Type: application/json" \
  -d '{"consumedBy": "my-service-instance-01"}'

完整流程图

┌────────────────────────────────────────────────────────────────────┐
│ Ceta Flow (生产者)                                                  │
│                                                                    │
│  ┌─────────┐    ┌──────────────────┐    ┌─────────────────┐       │
│  │ Trigger │───▶│ 业务逻辑节点      │───▶│ APP: MBS Enqueue│       │
│  └─────────┘    └──────────────────┘    └────────┬────────┘       │
└──────────────────────────────────────────────────┼────────────────┘
                                                   │
                                                   ▼
                                          ┌─────────────────┐
                                          │   MBS Service    │
                                          │                 │
                                          │  Broker: builtin│
                                          │  Queue: orders  │
                                          └────────┬────────┘
                                                   │ HTTP POST (Push)
                                                   ▼
┌────────────────────────────────────────────────────────────────────┐
│ 消费者服务 (Webhook)                                                 │
│                                                                    │
│  ┌───────────────┐    ┌──────────────┐    ┌───────────────────┐   │
│  │ 接收推送消息   │───▶│ 处理业务逻辑  │───▶│ 调用 MBS ACK 确认 │   │
│  │ X-Message-Id  │    │              │    │                   │   │
│  └───────────────┘    └──────────────┘    └───────────────────┘   │
└────────────────────────────────────────────────────────────────────┘

验证消息流转

通过 curl 快速验证

1. 发送一条测试消息：

curl -X POST http://10.0.1.50:8080/api/broker/builtin/queue/order-events/enqueue \
  -H "Authorization: Bearer your-token" \
  -H "Content-Type: application/octet-stream" \
  -d '{"event": "test", "message": "hello from ceta"}'

预期响应：

{"messageId": "550e8400-e29b-41d4-a716-446655440000"}

2. 确认消费者收到了推送：

检查消费者服务的日志，应该能看到收到的消息和 X-Message-Id。

3. 确认 ACK：

curl -X POST http://10.0.1.50:8080/api/subscribe/broker/builtin/queue/order-events/ack/550e8400-e29b-41d4-a716-446655440000 \
  -H "Authorization: Bearer your-token" \
  -H "Content-Type: application/json" \
  -d '{"consumedBy": "manual-test"}'

预期响应：

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