CETA Flow 集成指南

本章节详细说明如何在 Ceta 的 Flow 流程引擎中集成 MBS，实现消息的生产和消费。

集成架构

┌─────────────────────────────────────────────────────────────┐
│                    CETA Central 平台                          │
│                                                             │
│  ┌─────────────┐    ┌──────────────┐    ┌───────────────┐  │
│  │ MBS Connector│    │  Flow 引擎   │    │  REST Trigger │  │
│  │ (连接器实例) │    │ (流程执行)   │    │ (消息回调端点)│  │
│  └──────┬──────┘    └──────┬───────┘    └───────┬───────┘  │
│         │                  │                    │           │
└─────────┼──────────────────┼────────────────────┼───────────┘
          │                  │                    │
          ▼                  ▼                    ▲
┌─────────────────────────────────────────────────────────────┐
│                    MBS (Message Broker Service)               │
│                                                             │
│  Enqueue API ◄──── Connector Action (推送消息)               │
│  Subscribe API ◄── Connector Action (注册订阅)               │
│  ACK API ◄──────── Connector Action (确认消费)               │
│  消息推送 ──────────────────────────────► Flow REST Trigger  │
└─────────────────────────────────────────────────────────────┘

核心思路：

生产消息 - 在 Flow 中通过 APP 节点调用 MBS Enqueue Action
消费消息 - MBS 将消息推送到 Flow 的 REST Trigger 端点，触发消费流程

消息回调流程设计

当 MBS 推送消息时，会 POST 到 Ceta 的 Flow REST Trigger 端点。你需要设计一个回调流程来处理推送。

节点详解

1. TRIGGER 节点（REST 触发器）

配置项	值	说明
触发方式	REST	HTTP 请求触发
inMemory	true	必须设为 true（见下方说明）
URL 格式	`/flow/api/flow-rest-selfauth/{pbc}/{token}`	selfauth 模式无需额外认证
Request Headers	定义 `X-Message-Id`（string）	接收消息 ID
Request Body	根据消息格式定义（见下方说明）	接收消息内容

Request Body 的定义方式

TRIGGER 节点的 Request Body 定义取决于 MBS 推送的消息体格式：

情况一：消息体是 JSON 格式

如果生产者发送的消息是 JSON（如 {"event": "order_created", "orderId": "123"}），可以在 TRIGGER 的 Body 中直接定义对应的 key：

字段名	类型	说明
`event`	string	对应 JSON 中的 `event` 字段
`orderId`	string	对应 JSON 中的 `orderId` 字段

Flow 会自动将 JSON 请求体解析，按 key 映射到对应的变量中。

情况二：消息体是纯文本或非 JSON 格式

如果消息内容不是 JSON（如纯文本 hello world、XML 等），需要使用 rawBody 来接收整个请求体的原始内容：

字段名	类型	说明
`rawBody`	string	接收完整的原始请求体内容

提示

简单判断规则：

消息体是 JSON → 在 Body 中定义 JSON 的 key 名作为字段即可直接取值
消息体不是 JSON（纯文本、XML、二进制等）→ 定义 rawBody 字段接收原始内容

2. 处理消息节点

Groovy 脚本或业务节点，解析和处理消息数据：

输入：从 Trigger 获取 messageId 和 rawBody
输出：处理后的业务数据

3. ACK 节点（MBS ACK Action）

参数	值	说明
brokerName	与订阅时一致	必须完全匹配，区分大小写
queueName	队列名称
messageId	`$1.restTrigger.requestHeaders.X-Message-Id`	从 Trigger 的 Header 中获取
consumedBy	消费者标识	如 `ceta-flow-consumer-01`

4. 设置输出值节点

危险

必须设置 consumer 字段！ 这是 MBS 判定推送成功的关键条件。

使用"设置流程输出值"SDK 节点，设置返回值：

{"consumer": "sub01", "status": "processed", "messageId": "$2.messageId"}

consumer 字段的值为字符串，内容为消费者标识（任意非空字符串）。

5. STOP_JOB 节点

流程结束。

关键要求：inMemory 模式

警告

回调流程的 TRIGGER 节点必须设置 inMemory: true（内存执行模式）。

原因： MBS 推送消息后会等待 HTTP 响应。如果流程不是内存执行的，Ceta 会在触发后立即返回（不等待流程执行完成），此时响应不包含流程输出值中设置的 consumer 字段，MBS 会判定为推送失败。

设置方式：在流程定义的 TRIGGER 节点中确保 "inMemory": true。

认证方式

MBS 推送消息到 Ceta 的 Flow REST Trigger 时，推荐使用 selfauth 模式：

selfauth 模式（推荐）

URL 格式：/flow/api/flow-rest-selfauth/{pbc}/{token}
无需额外认证 Header
订阅时 callbackUrl 直接使用 selfauth URL

标准认证模式

URL 格式：/flow/api/flow-rest/{pbc}/{token}
需要在订阅 headers 中附加 JWT Token：

{
  "customHeaders": "{\"X-Message-Id\":\"{MESSAGE_ID}\",\"authorization\":\"Bearer <JWT_TOKEN>\"}"
}

警告

JWT Token 有过期时间，使用标准认证模式需要定期更新订阅配置。推荐使用 selfauth 模式避免此问题。

完整集成步骤

步骤 1：创建 MBS Connector 实例

在 Ceta 项目中创建 MBS Connector：

参数	示例
baseUrl	`https://message-broker.app-qa.test.bizops.com.cn`
token	`your-mbs-token`

步骤 2：创建回调流程

创建一个 Flow，设计如下节点：

TRIGGER (REST, inMemory=true, selfauth)
Groovy 脚本 - 处理消息
APP: MBS ACK - 确认消费
设置输出值 - 包含 consumer 字段
STOP_JOB

步骤 3：创建订阅

通过另一个 Flow（或手动）调用 MBS Subscribe Action：

参数	值
brokerName	`builtin`
queueName	`order-events`
subscriberId	`sub01`
callbackUrl	`https://{your-domain}/flow/api/flow-rest-selfauth/{pbc}/{flow-token}`
customHeaders	`{"X-Message-Id":"{MESSAGE_ID}"}`

关键： customHeaders 中的 {MESSAGE_ID} 是 MBS 占位符，每次推送时会被替换为实际的消息 ID。这样回调流程的 TRIGGER 节点就能通过 X-Message-Id Header 获取到当前消息的 ID。

步骤 4：发送消息测试

通过 MBS Enqueue Action 或直接调用 API 发送消息：

参数	值
brokerName	`builtin`
queueName	`order-events`
body	`{"event": "test", "message": "hello"}`

验证回调流程正确执行，并在 MBS 管理后台确认订阅者状态正常（连续失败计数为 0）。

常见集成问题

连续失败计数一直为 1

两个最常见原因：

回调流程未设置 inMemory: true - 非内存模式下流程触发后立即返回，响应中不包含 consumer 字段
流程中未设置 consumer 输出值 - 即使是内存执行，没有设置输出值也不会返回 consumer

ACK 返回 404

确认 brokerName 拼写完全一致（区分大小写）
确认 messageId 从请求 Header 中正确获取
确认消息未被重复 ACK

流程触发了但 ACK 失败

检查 MBS Connector 实例的 Token 是否有对应 Queue 的 read 权限
确认 MBS 服务可达

集成架构​

消息回调流程设计​

推荐的回调流程结构​

节点详解​

1. TRIGGER 节点（REST 触发器）​

Request Body 的定义方式​

2. 处理消息节点​

3. ACK 节点（MBS ACK Action）​

4. 设置输出值节点​

5. STOP_JOB 节点​

关键要求：inMemory 模式​

认证方式​

selfauth 模式（推荐）​

标准认证模式​

完整集成步骤​

步骤 1：创建 MBS Connector 实例​

步骤 2：创建回调流程​

步骤 3：创建订阅​

步骤 4：发送消息测试​

常见集成问题​

连续失败计数一直为 1​

ACK 返回 404​

流程触发了但 ACK 失败​