本文介绍一种基于 kafka streams 的混合处理方案：通过 merge 操作合并多路流，再结合自定义 processor 实现“跨流去重但保留单流内重复”的复杂业务逻辑，解决 outerjoin + aggregate 无法满足的多实例匹配场景。
本文介绍一种基于 kafka streams 的混合处理方案：通过 merge 操作合并多路流，再结合自定义 processor 实现“跨流去重但保留单流内重复”的复杂业务逻辑，解决 outerjoin + aggregate 无法满足的多实例匹配场景。
在 Kafka Streams 应用中，当需要协调多个异构输入流（如原始事件流与增强后带键的流）并执行精细化去重策略时，标准的 outerJoin 或 reduce/aggregate 往往力不从心——尤其当业务规则要求 “同一语义键在第二流中出现多次时，必须全部保留；而第一流中对应键的记录则完全丢弃”，这已超出两两关联或简单状态聚合的能力边界。
此时，推荐采用 merge() + 自定义 Processor 的组合方案，既保持流式处理的实时性，又获得对每条记录上下文的完全控制权。
✅ 核心思路：显式状态管理 + 流合并驱动
- 统一键映射：将两路输入流（raw 和 augmented）分别映射为相同语义键（如 getCommonKeyFromRawInputStream(value)），但不立即 join，而是保留原始元数据（如原始 key、来源标识、时间戳等）；
- 流合并（Merge）：使用 KStream#merge() 将两路流合并为单一逻辑流，确保所有记录按时间戳（或处理顺序）进入后续 Processor；
-
自定义 Processor 状态化处理：
- 使用 ProcessorContext#getStateStore() 绑定一个 KeyValueStore
> 存储每个语义键的历史记录； - 对每条记录判断其来源：
- 若来自 AUGMENTED 流 → 追加到该键对应列表，并标记“该键已被增强流覆盖”；
- 若来自 RAW 流 → 仅当该键尚未被任何 AUGMENTED 记录写入时才暂存（后续可被覆盖）；
- 在 punctuate() 或 close() 阶段（或根据业务选择 commit 时机），对每个键输出：
- 若存在 ≥1 条 AUGMENTED 记录 → 全部输出（满足条件4）；
- 若仅存在 RAW 记录 → 输出该条（满足条件1 & 3）；
- 若 RAW 与 AUGMENTED 并存 → 忽略 RAW，只输出 AUGMENTED（满足条件2）；
- 使用 ProcessorContext#getStateStore() 绑定一个 KeyValueStore
- 保序输出：因 merge 后记录天然按时间戳（或 Kafka offset）有序，且 Processor 内部不改变顺序，最终写入目标 topic 时即可维持原始 stream1 的逻辑顺序（需确保 window 和 grace 设置合理，避免乱序补偿干扰）。
? 示例 Processor 片段（简化版）
public class DedupProcessor implements Processor<string custommessagedetailswithkeyandorigin> {
private ProcessorContext<string custommessagedetailswithkeyandorigin> context;
private KeyValueStore<string list>> store;
@Override
public void init(ProcessorContext<string custommessagedetailswithkeyandorigin> context) {
this.context = context;
this.store = (KeyValueStore<string list>>)
context.getStateStore("dedup-store");
}
@Override
public void process(String key, CustomMessageDetailsWithKeyAndOrigin value) {
String semanticKey = value.getSemanticKey(); // 如 "1", "3", "9"
List<custommessagedetailswithkeyandorigin> list = store.get(semanticKey);
if (list == null) list = new ArrayList();
// 规则4优先：只要来的是 AUGMENTED，无条件追加
if (value.getOrigin() == OriginStream.AUGMENTED) {
list.add(value);
store.put(semanticKey, list);
} else if (value.getOrigin() == OriginStream.RAW) {
// 仅当当前键尚无 AUGMENTED 记录时，才暂存 RAW（后续可能被覆盖）
if (list.stream().noneMatch(v -> v.getOrigin() == OriginStream.AUGMENTED)) {
list.add(value);
store.put(semanticKey, list);
}
}
}
@Override
public void punctuate(long timestamp) {
// 可选：定期 flush 已确认无新 AUGMENTED 到达的键（如基于 watermark）
// 此处省略具体 flush 逻辑，实际中建议结合事件时间窗口做延迟提交
}
}</custommessagedetailswithkeyandorigin></string></string></string></string></string>
⚠️ 关键注意事项
- 状态存储必须启用：在 StreamsBuilder 中注册 Stores.keyValueStoreBuilder(...) 并绑定至 Processor；
- 语义键设计需幂等：getCommonKeyFrom...() 方法必须稳定、可逆、无歧义（如对 "1aug1" 和 "1aug2" 均返回 "1"）；
- 时序敏感性：若 augmented 流严重滞后，需配合 suppressed() 或 withTimestampExtractor() 确保事件时间正确；也可引入 Windowed<...> + suppress() 实现“等待窗口关闭后统一输出”；
- 资源与性能：List 存储虽灵活，但若某键重复极高，建议改用 HashSet 或限长队列（如 ArrayDeque）并配置 TTL 清理；
- Exactly-Once 保障：启用 processing.guarantee=exactly_once_v2，并确保 Processor 的 store 持久化与 checkpoint 机制正常工作。
综上，面对“跨流去重但保留单流内重复”这类非对称、多实例、状态依赖型需求，放弃声明式 join/aggregation，转向命令式 Processor 是更清晰、可控且可维护的选择。它将业务逻辑显式暴露于代码中，便于单元测试、调试与未来演进。
