← 返回首页

ESB · Data Flow · Onboarding

ESB 系统数据流程图 · 新人上手指南

esb-broker + data-flow-provider 全链路 · 从"一分钟看懂"到"排错手册" · 所有示例均取自真实生产数据（2026-10-12）
你不需要预先了解任何一个系统，按顺序往下读即可。

我是新同学 → 第 1、2 节理解概念，第 3 节看图 我要接入新下游 → 第 4、5、7 节 我在排查线上问题 → 第 6 节 + 第 8 节排错手册

1 · 一分钟看懂：ESB 在干什么

公司里有很多系统。当 A 系统发生了一件事（下了单、人员变动、审批通过），往往需要通知 B、C、D 系统。 如果让 A 挨个去对接每一家，谁上线了新系统都要改 A 的代码——这就是"面条式集成"。 ESB（企业服务总线）就是给所有系统之间的通知加一个统一中转站：A 只要把"发生了什么事"发给中转站， 中转站负责记台账、找到所有关心这件事的系统、挨个送过去、记录每一单的送达结果。

Publisher · 寄件人

上游系统

发生事件的一方。它只需要调中转站的接口"我要发一个 xxx 事件"，不需要知道谁在听。

mchr-basic.employee
Broker · 中转站

esb-broker

本页主角。收件 → 记台账（MongoDB）→ 按名单投递（RabbitMQ）→ 记录送达结果 → 失败重试、可选回调。

esb-broker-provider
Subscriber · 收件人名单

订阅者

关心这件事的下游系统。每个订阅者配好：送到哪个地址、怎么算"送达成功"、失败要不要重试/发邮件。

kun-realm-server-emp
Message · 运单

消息

中转站给每个收件人单独生成一条记录（独立状态、独立重试）。一次发布 = 一张"总单"（topic）+ N 条"运单"（message）。

broker_message
一句话流程：上游调 /topic/send/... 发布 → broker 按发布者的订阅名单扇出给每个订阅者 → 每条消息经 RabbitMQ 送到投递侧 → 投递侧把报文 POST 给订阅者配置的地址 → 按 expectValue 判定成败 → 成功记 OK，失败按开关重试、按需回调上游。全过程落在 MongoDB 里，可通过接口查询和重推。

2 · 核心术语速查 先混个眼熟，读图时回来查

术语一句话解释存在哪
发布者 Publisher "谁可以发布什么事件"，以及它的订阅者名单。ID 格式固定为 {service}-{method}（全小写）。 broker_publisher
订阅者 Subscriber 一个下游怎么收消息：推送地址（targetHost+targetPath）、成功判定（expectCode/expectValue）、重试与邮件开关。 broker_subscriber
会话 Topic 一次"发布动作"的汇总记录，包含本条消息扇出给了哪些订阅者。发布接口返回给你的 ID 就是它。 broker_topic
消息 Message 发给某一个订阅者的投递记录：要发的报文（postBody）、当前状态、重试次数、下游响应（respBody）、判定结果。 broker_message
投递日志 Logging 每次真实 HTTP 调用的快照：实际推送的 postBody、下游回包、耗时。"下游到底收到了什么"看它。 broker_logging
Binding / Queue / Exchange RabbitMQ 的路由三件套：消息从交换机按 routingKey 进队列，队列由 broker 的监听器消费。通常不用手改。 broker_binding 等
模板 Template data-flow-provider 的"翻译+转运站"：把 broker 发来的通用报文用 Freemarker 转成最终系统需要的格式，再转发。/message/template/receive/** 每个 URI 对应一张模板。 data_flow_api / data_flow_template
成功判定 expectValue 怎么算"送达成功"：下游响应 JSON 里 expectCode 字段的值必须等于 expectValue。本系统主流约定是 code=201。 订阅者/模板配置

3 · 图① 全景数据流 按 ①→⑨ 顺序读一遍

左侧是"谁在发"，右侧是"发给谁"，中间是 broker 的两工作段：接收侧（接活、记台账）和投递侧（送货、判定、记录结果）。三个浅色大框是区域划分：触发方 / 存储与消息队列 / 下游。

ESB 系统端到端数据流图 上游系统经 HTTP 或 MQ 触发 esb-broker，消息落库后经 RabbitMQ 投递，由投递侧推送至 data-flow-provider 模板链路或直连业务系统，并回写日志与状态。 触发方 存储 / MQ 下游 ① HTTP 推送 ② MQ 投递 ③ 落库 ④ 发布 ⑤ 消费 ⑥ 投递 ⑦ 直达 ⑧ 转发 ⑨ 日志回写 SRC 上游业务系统 obc-order · mchr-basic · … SRC 外部系统 · MQ 投递 thirdhub_shelve · pipeline APP esb-broker · 接收侧 /topic/send · /message/* Forward 监听器（MQ 入口） 查缓存 → 落库 PENDING DB MongoDB · broker_* publisher · subscriber topic · message · logging MQ RabbitMQ · broker_topic routingKey → 订阅者队列 重试 · 回调 · 死信队列 APP esb-broker · 投递侧 DeliveryMessageListener POST postBody → targetUrl 判定 200 且 code == 201 SYS data-flow-provider /message/template/receive/** 模板类型见 图③ SYS 最终业务系统 泛微 OA · EHR · NC · HR SYS 直连业务系统 order-server 等 ISB 响应 code:"201" NOTES · 说明 成功判定：HTTP 200 且响应 JSON 的 code 字段 == expectValue（data-flow 与 ISB 家族成功码均为 201）。 运维操作：/message/sendById 重推 · /message/updatePostBody 修改 postBody 后重新入 broker_topic。 LEGEND 主数据流 · HTTP / MQ 异步 · 回写 焦点：消息总线 · 投递判定

4 · 图② 一条消息的真实旅程 真实数据 · 一次发布如何走完全程

下面这张序列图用生产库里一条真实成功的记录，把"上游发布 → 中转 → 模板转换 → 最终系统 → 回到成功"的每一步画出来。 例子：mchr-basic.employee（人员信息变更）发布一次，扇出给 8 个订阅者，我们跟踪其中 kun-realm-server-emp 这一条的旅程。 时间从上往下流动，竖直虚线是各方的"生命线"。

一条消息的真实旅程序列图 展示 mchr-basic.employee 发布后，其中一条发往 kun-realm-server-emp 的消息从上游调用、broker 落库、RabbitMQ 投递、data-flow 模板转换到最终系统并返回 201 成功的完整时序。 ① HTTP 发布 ② 落库 topic + message ×8 (PENDING) ③ 发布 MQ ④ 投递队列 ⑤ 投递下游（postBody） ⑥ 模板转发 ⑦ 返回 code:0 ⑧ 响应 201 成功 ⑨ 判定 OK → ENDING + logging 上游系统 mchr-basic esb-broker 接收侧 + 投递侧 RabbitMQ broker_topic data-flow-provider 模板引擎 最终系统 kun-realm-server 注：示例取自真实生产记录（2026-10-12）——mchr-basic.employee 一次发布扇出 8 个订阅者，本图跟踪其中 kun-realm-server-emp 一条， 该条 resultStatus=OK。MongoDB 的交互集中在 ② 落库与 ⑨ 回写（broker_topic / broker_message / broker_logging）。 LEGEND 同步调用 返回响应 异步 / 入队 成功（判定通过）

把这张图每一步的真实值拆开对照（可在库里逐项验证）：

步骤发生了什么真实数据
① 发布 上游调 broker 的通发接口，body 是业务报文。 POST /topic/send/mchr-basic?method=employee
② 扇出 + 落库 broker 按 {service}-{method} 找到发布者缓存，拿到全部订阅者，给每个订阅者建一条 message（PENDING）。 broker_topic + broker_message × 8；其中一条 _id=da8f3818…（subscriberId=kun-realm-server-emp，状态 PENDING），并记录所属的 topicId
③ 入队 按订阅者绑定的路由键发到 broker_topic 交换机，消息进入该订阅者的队列。 exchange=broker_topic · queue=broker.topic.hr · routingKey=basic.hr
④ 消费 + 投递 投递侧监听器从队列取到消息，把 postBody POST 给订阅者配置的地址。 POST http://data-flow-provider/message/template/receive/hr/kun/emp
⑤ 模板转换 data-flow 以 URI 查模板：data_flow_api → data_flow_template，用 Freemarker 把通用报文转成最终系统要的格式后转发。 模板 kun-realm-server-emp → https://kcrealm-api.kunmax.cn/system/message/add-kc-contact
⑥ 判定（双层） 第一层：broker 看 data-flow 的回包 code==201；第二层：data-flow 看最终系统的回包 code==0。 data-flow 回 {"code":201,"message":"Success","subCode":200,...} → 判定 OK
⑦ 收尾 消息置 ENDING、写投递日志；如配了 callbackUrl 则通知上游；失败则按开关重试。 resultStatus=OK · broker_logging 新增一条调用快照
注意"双层校验"这个易混点：broker 侧只看 data-flow 的 code（201 成功）；最终系统的成功码（这里是 code=0）由 data-flow 的模板配置校验。 两层配置各管各的——排查"消息 ERROR"时，先确认是哪一层没过。

5 · 图③ 模板路由与模板类型 12 种实现 · 生产共 160 条模板

把所有发往 data-flow-provider 的消息放大看：它本质上是个"按 URI 路由 + 按类型选引擎 + 按模板转换"的转运站。 URI 本身就是数据库主键，查两次表找到模板，再按模板的 serviceId 选择对应的执行引擎（共 12 种实现类）。

data-flow-provider 模板路由与模板类型图 模板请求经 data_flow_api 与 data_flow_template 路由，按 serviceId 匹配 12 种 ITemplateFlow 实现类，分为 OA 审批族、泛微数据接口、三方系统适配、通用模板四组，转发到对应业务系统。 ① 分发 ② 未命中 ③ 匹配 API 模板入口 /message/template/receive/** CommController 兜底 POST APP 路由分发 · RequestSupport 先查 REQUEST_MAPPING 硬编码 命中 3 条 → 直执行 IFlow DB 模板路由 · Mongo 双子表 data_flow_api（URI 为 _id） → data_flow_template（Freemarker） TPL serviceId 匹配实现类 12 个 ITemplateFlow 实现 未匹配 → DefaultTemplateFlow OA OA 审批族 · 41 条 oa.approval · 31 oa.approval.submit · 5 oa.approval.withDraw · 3 · delete · 1 oa.approval2 · 1 → fanwei /api/workflow/paService/* WEB 泛微数据接口 · 37 条 oa.base.restful · 37 saveOrUpdateModeData/{model} 入参透传 + 出参校验 → fanwei /api/cube/restful/interface/* 3RD 三方系统适配 · 14 条 hangao · 11 → 汉高 dmeapi gokuai · 3 → 够快云库 ofs · oa.get → 未配置使用 → HTTPS 适配（hangao 跳过 SSL 校验） GEN 通用模板 · 68 条 未配置 serviceId · 57 DEFAULT · 1 · hr-basic · 6 forEach · 4（循环拆分） → Freemarker 转换 / 透传 ① 硬编码旁路：REQUEST_MAPPING 注册 3 条直连 Flow（tyc/oa · hr/kzzt/emp · wdt/hangao/product/mapping），命中即直执行 IFlow，不进模板。 ② 未登记 data_flow_api 的 URI 抛「未找到接口的实现类」；serviceId 无匹配实现时回落 DefaultTemplateFlow（hr-basic 6 条走此路径）。 ③ 模板出参校验：expectField/expectValue（OA 族 code=SUCCESS、hangao code=0、gokuai error_code=0），不匹配则返回 code=1（broker 判 ERROR）。

6 · 图④ 消息状态、重试与回调 失败之后会发生什么

每条消息都有一个状态机。投递失败不会被丢弃：按订阅者配置的开关（业务失败重试 / 系统失败重试）进入延时重试队列，最多 3 次； 如果订阅者配了回调地址，非链路异常的结果还会回调通知上游。这就是"为什么消息最终一定会有一个明确结果"。

消息状态流转、重试与回调图 broker_message 从 PENDING 经投递判定流转为 OK 或 ERROR/FAILURE，失败可经延时重试队列重投，结果经回调队列通知上游。 消费 判定 成功 失败 条件重试 延时后重投 回调 失败回调 MSG 消息落库 · PENDING broker_message APP 投递中 · SENDING POST postBody → targetUrl RULE 响应判定 200 且 code == 201 → OK OK 业务成功 · ENDING resultStatus = OK ERR 失败 · ERROR / FAILURE code ≠ 201 / 超时异常 MQ 延时重试队列 broker.dead(TTL) → broker.retry APP 回调上游 POST callbackUrl 重试条件：retryWhenError / retryWhenFailure 为真且 retryCount < 3，经 broker.dead 的 TTL 延时后重投。 回调条件：callbackUrl 非空且结果非 FAILURE，由 CallbackMessageListener 推送 messageId 触发的回调。

7 · 数据存在哪 遇到问题先想"该查哪张表"

集合存什么排查时怎么用
broker_publisher 发布者：{service}-{method} + 订阅者名单（subscribers 数组）。 确认"这个事件谁能发、会扇出给谁"。83 条。
broker_subscriber 订阅者：推送地址、成功判定、重试/邮件开关、绑定关系。 确认"发给谁、怎么算成功"。306 条。
broker_topic 一次发布的会话：发布者、报文、扇出了哪些订阅者。 通发接口返回的 ID 在这里；可再按 topicId 找消息。
broker_message 每条消息的完整生命周期：postBody、状态、重试次数、下游回包、判定结果。 排查主战场。按 topicId / sourceService+sourceMethod / 时间查。2600 万+。
broker_logging 每次真实 HTTP 调用的快照（含实际推送的 postBody、回包、耗时）。 查"下游究竟收到了什么"；按 messageId 倒序取最新一条。5000 万+。
broker_binding/queue/exchange/listener RabbitMQ 路由三件套与监听器配置。 确认消息从交换机到队列的路由链路；一般不用动。
data_flow_api URI → 模板 ID 的映射（_id 就是 URI 本身）。 确认"这个地址有没有登记模板"。139 条。
data_flow_template 模板配置：Freemarker 模板、目标地址、出参校验、serviceId。 确认"转发去哪、转换逻辑、怎么算成功"。160 条。
data_flow_service 目标主机别名（如 fanwei → https://fanwei.ikunchi.com）。 解析模板里 targetHost 是主机名时的真实地址。

8 · 新人排错手册 常见问题 · 按实际情况对号入座

调 /topic/send 报「找不到发布者」？
{service} 要填发布者（上游）的 service，不是订阅者。查找键是 {service}-{method} 全小写。例：想测 kun-realm-server-emp 的投递，要发 /topic/send/mchr-basic?method=employee，而不是 kun-realm-server.emp。
/topic/send 返回的 data 是消息 ID 吗？
不是，那是 topicId（会话 ID）。查消息用 broker_message.find({topicId})，或按 sourceService + sourceMethod + 时间 查。
topic 建了，但 broker_message 里没有消息？
多半是 broker_publisher.subscribers 里的 ID 与 broker_subscriber._id 大小写不一致（如 withdrawRequest vs withdrawrequest），缓存查不到订阅者就不会生成消息。核对两侧 ID 并保持一致。
resultStatus=ERROR 和 FAILURE 有什么区别？
ERROR：下游 HTTP 通了，但响应里的 code ≠ expectValue（业务失败）；FAILURE：调用本身异常/超时（链路失败）。是否重试由订阅者的 retryWhenError / retryWhenFailure 开关决定（retryCount<3）。
成功码 201 到底是什么？203 呢？
201 是历史约定的成功码，现在由两层保证：data-flow 侧 ControllerAop 把成功码统一改写成 201；ISB 家族直连系统（order-server、NC 系）原生返回 code:"201"，拒绝则 "203"（REFUSED）。broker 侧按订阅者 expectValue 判定。
只想推给某一个下游，应该用哪个接口？
用 /message/singleSend（传 targetService + targetMethod 直接定位订阅者）。/topic/send 是按发布者扇出给全部订阅者的广播接口。
下游说收到的报文不对 / 没收到，从哪查？
查 broker_logging（按 messageId 倒序，最新一条）：postBody 是实际推送的报文快照，respBody 是下游回包，spend 是耗时。若快照与预期不符，先确认是不是有人改过 postBody。
有一条失败消息想重推，或者想改报文再推？
重推：POST /message/sendById?id=xxx（要求消息状态 ENDING）；改报文：先 /message/updatePostBody。⚠️ 编辑时保持大数字为字符串——19 位 ID 用会解析 JSON 的工具（JS/格式化插件）会丢精度，broker 全程字符串透传不背这个锅。

附 · 三张速查卡 贴到工位上的那种

判定规则

成功码 201 从哪来

  • broker_subscriber.expectCode/expectValue 决定判定，主流配置为 code == "201"。
  • data-flow 侧由 ControllerAop 把 kbs 成功码 0 统一改写为 201。
  • ISB 家族（order-server、NC 系）原生返回 code:"201"（失败 "203" REFUSED）。

模板类型 · 12 实现类

四大模板族（生产 160 条）

  • OA 审批族 · 41 条：oa.approval / approval2 / approval.submit / approval.withDraw / approval.delete → 泛微审批接口。
  • 泛微数据 · 37 条：oa.base.restful → 泛微 cube RESTful。
  • 三方适配 · 14 条：hangao / gokuai / ofs / oa.get。
  • 通用模板 · 68 条：DEFAULT（含未配置 57）/ hr-basic / forEach。

可靠性与运维

重试 · 回调 · 重推

  • 消息先落库再入队，消费确认后才算投递完成；状态与耗时写入 broker_logging。
  • 失败按订阅者开关经 broker.dead → broker.retry 延时重试。
  • 运维入口：/message/sendById 重推、/message/updatePostBody 改报文。