《Java 消息中间件·事件驱动架构》8.3 流增强

增强不是「加字段」

Stream enrichment 的目的是让一条事件携带足够的信息，使下游消费者不需要再查外部系统就能做出业务判断。在 libraryd 里，一个借阅请求如果只包含 loanId 和 bookId，通知服务还需要去查会员表才能找到手机号；如果流处理阶段就把手机号塞进事件，通知服务就是纯「取即发」。

增强有三种常见方式，各自解决不同的问题：

这三种方式不是排他的。在一个复杂拓扑里，可能先用 join 拼上参照字段，再用 flatMap 拆成通知和审计两条记录，最后用 branch 把「紧急」和「普通」两类消息分别路由到不同的 sink topic。

方式一：Join 增强

Join 增强的本质是「用参照表 lookup 来补字段」。这种方式在第 8.2 节已经详细展开过，结论可以复用：

KStream<String, String> loanStream = builder.stream("loan-events");
KTable<String, String> memberTable = builder.table("member-info");

KStream<String, String> enriched = loanStream.leftJoin(
    memberTable,
    (loan, member) -> {
        if (member == null) return loan + "|MEMBER_UNKNOWN";
        return loan + "|" + member; // 拼上会员信息
    }
);

Join 增强的局限是严格依赖 key 对齐。如果 loan 事件的 key 是 loanId，而参照表的 key 是 memberId，就必须先做 selectKey 把 key 改成 memberId，join 后再改回来。两次 repartition 的代价在吞吐高时不可忽视。

另一个局限是：join 只能把当前快照拼上去。如果事件时间很重要，比如「这笔借阅发生时会员的等级是什么」，而会员表 10 分钟后才更新，join 取得的是更新后的等级，不是事件发生时的等级。这类场景需要事件时间窗口 join 或领域事件设计（把等级也写进借贷事件本身）。

方式二：Branch + Merge 增强

Branch 把一个流拆成多条子流，merge 再把多条子流拼回一条。在增强场景里，branch 用来识别「事件的类型或条件」，每种类型走不同的增强逻辑，最后用 merge 把结果统一送到下游。

libraryd 场景：借阅事件有两个字段标识紧急程度——预借图书（HOLD 类型）和普通借阅（LOAN 类型）。预借需要在 5 分钟内通知，走短信通道；普通借阅走邮件通道。同一个流先拆成两条，分别增强后再合并。

KStream<String, String> source = builder.stream("loan-events");
KStream<String, String>[] branches = source.branch(
    (k, v) -> v.contains("HOLD"),   // 预借，走短信
    (k, v) -> v.contains("LOAN")    // 普通，走邮件
);

KStream<String, String> urgent = branches[0]
    .mapValues(v -> v + "|SMS|HIGH_PRI");
KStream<String, String> normal = branches[1]
    .mapValues(v -> v + "|EMAIL|NORMAL");

// 合并回一条流，下游统一处理
KStream<String, String> merged = urgent.merge(normal);
merged.to("loan-notifications");

Branch + Merge 的代价：

• 每个 branch 在拓扑里被编译成 KStream-BRANCH 处理器，本质上是一个带有多个输出子话题（后面的 topic 后缀为 -branch-0、-branch-1）的内部 repartition。这些内部 topic 默认保留 30 天。
• 如果分支条件有重叠（即一条记录匹配多个分支），Streams 会按分支顺序把它送进第一个匹配的分支。这既是约定也是陷阱——如果条件设计不当，高优先级的分支会「吞掉」本应进低优先级分支的记录。

方式三：FlatMap 增强

FlatMap 是「一条输入产生多条输出」。在增强场景里，最常见的用法是：一条业务事件同时触发「通知」和「审计」两类下游消费，从同一条消息派生出两条不同语义的消息。

实验：1 条输入产生 2 条输出

实验环境：Temurin 25 / Kafka 4.3.0 / kafka-streams 4.3.0。

场景：libraryd 的一条借阅请求 L-101|B-001|M-1|BORROWED，通过 flatMap 拆成「通知消息 + 审计消息」两条。

// snippet: StreamEnrichmentDemo.java
public class StreamEnrichmentDemo {
    public static void main(String[] args) {
        StreamsBuilder builder = new StreamsBuilder();
        KStream<String, String> source = builder.stream("arch08-3.loan-requests");

        KStream<String, String> enriched = source.flatMap((key, value) -> {
            List<KeyValue<String, String>> result = new ArrayList<>();
            result.add(KeyValue.pair(key, value + "|NOTIFY"));
            result.add(KeyValue.pair(key, value + "|AUDIT"));
            return result;
        });
        enriched.to("arch08-3.enriched");

        try (TopologyTestDriver driver = new TopologyTestDriver(
                builder.build(), props())) {
            // ...
        }
    }
}

输入：

  IN  M-1 -> L-101|B-001|M-1|BORROWED

输出：

  OUT M-1 -> L-101|B-001|M-1|BORROWED|NOTIFY
  OUT M-1 -> L-101|B-001|M-1|BORROWED|AUDIT
  Total output records: 2

一条输入被拆成两条输出，两条 output 的 key 都是 M-1，value 分别带上不同的后缀标签。下游消费者可以根据 |NOTIFY 或 |AUDIT 的标记做路由，也可以用更结构化的方式（比如把后缀换成 JSON envelope 里的 type 字段）。

FlatMap 与 FlatMapValues 的区别

// flatMap：可以改 key
source.flatMap((key, value) -> {
    return List.of(
        KeyValue.pair(key + "-notify", value + "|NOTIFY"),
        KeyValue.pair(key + "-audit", value + "|AUDIT")
    );
});

// flatMapValues：只改 value，不改 key，不触发 repartition
source.flatMapValues(value -> List.of(
    value + "|NOTIFY",
    value + "|AUDIT"
));

flatMapValues 因为不改 key，拓扑不需要 repartition，性能更高。但如果业务上两条增强记录需要不同的 key（比如通知按手机号分片、审计按部门分片），就必须用 flatMap。

三种方式的决策路径

在 libraryd 的实际拓扑设计中，可以用下面的逻辑选择增强方式：

一个具体的设计示例：libraryd 的借阅事件经 join 补充会员信息后，走 flatMapValues 拆成通知和审计两条，最后用 branch 把「高优先级预借」从普通借阅里分出来走短信通道。

// 伪代码展示三层增强的典型组合
KStream<String, String> loanStream = builder.stream("loan-events");

// Layer 1: Join 补充会员信息
KStream<String, String> withMember = loanStream.leftJoin(memberTable, ...);

// Layer 2: FlatMapValues 拆成通知和审计
KStream<String, String> split = withMember.flatMapValues(v -> List.of(
    v + "|NOTIFY", v + "|AUDIT"
));

// Layer 3: Branch 把紧急通知分出来
KStream<String, String>[] br = split.branch(
    (k, v) -> v.contains("HOLD") && v.contains("NOTIFY"),
    (k, v) -> true
);
br[0].to("urgent-sms");
br[1].to("normal-queue");

FlatMap 与 Branch 的拓扑差异

FlatMap 和 Branch 看起来都能从「一条输入」变出「多条输出」，但它们的拓扑节点类型完全不同。FlatMap 对应的是 KSTREAM-FLATMAP node，Branch 对应的是 KSTREAM-BRANCH node 加多个 repartition topic。

FlatMap 的优势是不产生内部 topic。每条输入直接在内存里展开成 List，然后一次性送给下游 sink。代价是：如果 flatMap 的展开逻辑复杂（比如需要查数据库、调外部 API），展开时间会直接阻塞当前 processing thread。Streams 的 flatMap 是同步的，没有异步选项。

Branch 的优势是分出来的每条子流都可以独立应用不同的算子（map/filter/to），最后再用 merge 拼起来。代价是每条子流如果后续还有聚合算子（groupByKey/reduce），会产生额外的 repartition。

// Branch 的典型拓扑
topic: loan-events
  -> KSTREAM-BRANCH (produces nothing directly)
    -> loan-events-branch-0  (HOLD records)
      -> mapValues
      -> to: urgent-sms
    -> loan-events-branch-1  (LOAN records)
      -> mapValues
      -> to: normal-queue

FlatMap 的典型拓扑：

// FlatMap 的典型拓扑
topic: loan-requests
  -> KSTREAM-FLATMAP
    -> KSTREAM-SINK (topic: enriched)
      // 每条输入直接变成 2 条输出，没有中间 topic

设计上的选择原则是：如果需要「一对多展开且每条子记录语义不同」，用 flatMap；如果需要「按条件走不同处理链再合并」，用 branch+merge。

FlatMap 的消息顺序保证

Kafka Streams 保证：对于同一条输入记录，flatMap 返回的 List 中的 KeyValue 对会保持 List 顺序发送到下游。也就是说，如果 flatMap 返回 [NOTIFY, AUDIT]，那么 output topic 里一定先收到 NOTIFY，再收到 AUDIT。

但这个保证仅在同一输入记录内有效。如果有两条输入记录 A 和 B，A 的 flatMap 产生 [A1, A2]，B 的 flatMap 产生 [B1, B2]，那么 output 的顺序可能是 A1, A2, B1, B2 或者 A1, B1, A2, B2——取决于 records 从 input topic 的消费顺序。如果两个 partition 被不同 task 同时消费，顺序完全由 partition 分配和 produce buffer 决定。

在 libraryd 的通知 + 审计场景中，顺序通常不重要：NOTIFY 和 AUDIT 是独立下游服务消费。但如果 flatMap 的结果是「命令链」——比如 A 必须先发「锁定库存」再发「扣减库存」——顺序就是业务正确性的前提。这种情况下，要么把两条命令写到同一个 partition（通过固定 key），要么在业务层自己维护状态机。

增强的可靠性

增强阶段如果出错，事件会怎么处理？Streams 的默认行为是抛异常、停止任务、进入死循环。生产环境需要显式配置一套容错策略：

对 join 增强来说，最常见的问题是「参照表里没有这个 key」。left join 会返回 null，你的 value joiner 需要处理这种情况，而不是直接 NPE。一个稳健的做法是在 joiner 里返回带 UNKNOWN 标记的字符串，让下游消费者自己决定怎么处理缺失信息，而不是让异常向上穿透导致整个 task 卡住。

(loan, member) -> {
    if (member == null) return loan + "|MEMBER_UNKNOWN";
    return loan + "|" + member;
}

对 flatMap 增强来说，常见的问题是「展开逻辑抛异常」。如果 flatMap 里的逻辑调用了外部服务（比如查 translate API 给 value 加多语言标签），网络超时会导致整个 processing thread 挂掉。 Streams 对此没有内置重试，要么在 flatMap 里自己加 try-catch 和降级逻辑，要么把外部调用拆成独立的异步 processor（用 Kafka Streams 的 Processor API 或 Transformer）。

小结

• Stream enrichment 的三种方式各有定位：join 补字段、branch+merge 按路径增强、flatMap 做一对多展开。
• Join 增强依赖 key 对齐，外键场景需要 repartition，且只能取参照表的当前快照而非事件发生时的状态。
• Branch + Merge 能把同一条流按条件分流再聚合，但分支条件不可重叠，且产生内部 repartition topic。
• FlatMap 实现一条输入多条输出，本机实测 1 条借阅请求拆成通知和审计 2 条输出；不改 key 时用 flatMapValues 可省 repartition。
• 增强阶段必须处理 join 失败和异常记录，避免 NPE 穿透到 Streams 引擎层。
• flatMap 的 List 内部顺序保证在同输入记录内有效，但跨记录的顺序完全由消费顺序决定；命令链场景需用固定 key 或业务层状态机保障顺序。
• Branch+merge 与 flatMap 的最大拓扑差异在于：branch 产生内部 repartition topic，flatMap 不产生。
• 实际生产中三种增强方式经常组合使用：先用 join 补参照字段，再用 flatMap 拆多路输出，最后用 branch 按条件路由到不同 sink topic。

阅读导航：上一节 8.2 KTable 的 Join ｜ 下一节 9.1 事件溯源概述