MBS 消息代理服务

简介

MBS (Message Broker Service) 是一个统一消息代理服务，已作为 Connector 集成到 Ceta 平台中。它将多种消息队列后端（Kafka、RabbitMQ、RocketMQ、ActiveMQ、PostgreSQL）封装为统一的 HTTP API，通过 MBS Connector，你可以在 Flow 中方便地进行消息的生产、订阅和确认，无需直接对接底层消息队列。

核心能力

统一 HTTP 接口 - 无论底层使用哪种 MQ，API 接口完全一致
消息推送 (Enqueue) - 向指定 Broker 的 Queue 发送消息，支持任意格式（JSON、XML、纯文本）
基于 Webhook 的消息订阅 (Subscribe) - 注册回调地址，MBS 主动推送消息到你的服务
消息确认 (ACK) - 确认已消费的消息，防止重复投递
两种投递模式 - FANOUT（广播给所有订阅者）和 QUEUE（轮播，每条消息只投递给一个订阅者）
Callback URL 占位符 - 支持 {BROKER_NAME}、{QUEUE_NAME}、{MESSAGE_ID}、{SUBSCRIBER_ID} 占位符动态替换
消息包装 (Warp) - 支持将原始消息体编码并包装到 JSON 结构中推送
Token 认证与细粒度权限控制 - 支持 Broker/Queue 级别的读写权限配置，通配符匹配
白名单安全控制 - 对 Callback URL 进行白名单校验，支持 IP、CIDR、主机名三种类型
多后端统一管理 - 通过可视化管理后台统一管理所有 Broker、Queue、Token、白名单
高可用 - 支持多实例水平扩展部署，自动 Leader 选举

在 Ceta 中的架构定位

在 Ceta 的分层架构中，MBS 通过 Connector 模块 接入平台：

┌────────────────────────────────────────────────────┐
│             Flow (流程引擎)                          │
│  FlowDefinition → Trigger → Instance → Job → Node │
├────────────────────────────────────────────────────┤
│             Connector (连接器)                       │
│  ConnectorType(MBS) → Connector实例 → Action       │
├────────────────────────────────────────────────────┤
│             外部服务                                 │
│  MBS Service (HTTP API)                            │
└────────────────────────────────────────────────────┘

Ceta 的 Connector SPI 模型为：

ConnectorType(类型模板) → Connector(实例) → Action(操作) → Groovy脚本执行

MBS Connector 遵循这个模型：

ConnectorType = MBS 连接器类型（定义了 baseUrl 和 token 两个参数，以及连接脚本）
Connector = MBS 连接器实例（填写具体的服务地址和 Token）
Action = 三个操作（Enqueue、Subscribe、ACK），每个 Action 通过 Groovy 脚本调用 MBS HTTP API

在 Flow 中使用时，通过 APP 节点 调用 MBS Action 完成消息相关操作。

组成部分

组成部分	名称	说明
ConnectorType	MBS	连接器类型定义，配置 MBS 服务地址和 Token
Action	MBS Enqueue (消息推送)	向队列推送消息
Action	MBS Subscribe (消息订阅)	注册 Webhook 订阅
Action	MBS ACK (消息确认)	确认已消费的消息

整体消息流转

┌─────────────┐     Enqueue      ┌─────────────┐     Push       ┌─────────────┐
│   生产者     │ ──────────────▶  │     MBS     │ ────────────▶  │   消费者     │
│ (Flow/脚本) │                  │   Service   │               │ (Webhook)   │
└─────────────┘                  └─────────────┘               └──────┬──────┘
                                       ▲                              │
                                       │           ACK                │
                                       └──────────────────────────────┘

详细流程：

生产者在 Flow 中通过 APP 节点调用 MBS Enqueue Action，将消息推送到 MBS 指定 Broker/Queue
MBS 将消息存入底层 MQ（Kafka/RabbitMQ/BUILTIN 等）
MBS 根据该 Queue 上的订阅配置，将消息通过 HTTP POST 推送到消费者的 Callback URL
推送时请求头中附带 X-Message-Id（消息唯一标识）
消费者处理完消息后，通过 MBS ACK Action 确认消费
如果 ACK 超时未确认（默认 30 秒），MBS 会自动重新投递消息
如果连续推送失败达到 retry 次数（默认 5 次），订阅会被自动移除

支持的消息队列后端

后端	说明
BUILTIN (PostgreSQL)	内置队列，无需额外部署 MQ，适合中小规模场景
Kafka	高吞吐量分布式流平台，适合大数据场景
RabbitMQ	AMQP 消息中间件，适合复杂路由场景
RocketMQ	阿里开源分布式消息系统，适合金融级可靠性场景
ActiveMQ	Apache 消息中间件，适合传统 JMS 场景

信息

对于 Ceta 平台用户，你无需关心底层使用的是哪种 MQ。MBS 已将差异封装为统一的 HTTP API，只需知道目标 Broker 名称和 Queue 名称即可。具体使用哪种 MQ 后端由 MBS 管理员在管理后台配置。

部署前提

MBS 是一个独立部署的服务，Ceta 通过 HTTP API 与其通信。使用前需确保：

条件	说明
MBS 服务已部署	需要独立部署 MBS 服务（Java 17+, PostgreSQL 12+）
网络可达	Ceta 实例能够访问 MBS 服务地址
Token 已创建	在 MBS 管理后台创建了 API Token
Broker/Queue 已创建	在 MBS 管理后台创建了目标 Broker 和 Queue
白名单已配置	如需使用 Subscribe，消费者地址需加入白名单

MBS 的部署方式支持 Docker、Docker Compose 和直接 JAR 运行，详见 MBS 独立部署文档。

接入方式

MBS ConnectorType 和对应的三个 Action 已预置在平台中。使用时只需：

创建 MBS Connector 实例（填写实际的 MBS 服务地址和 Token）
在 Flow 中通过 APP 节点选择对应的 Action 即可使用

简介​

核心能力​

在 Ceta 中的架构定位​

组成部分​

整体消息流转​

支持的消息队列后端​

部署前提​

接入方式​

简介