如何在Kafka中实现批量校验并将批次数据路由到对应Topic

核心思路 #

该需求属于有状态流处理场景，核心逻辑是先缓存单个批次从BS（批次开始）到BE（批次结束）的全量数据，待BE标识到达并完成批次有效性校验后，将整个批次统一路由到指定目标Topic。可以通过Kafka Streams原生能力实现，无需引入额外组件。

具体实现步骤 #

1. 前置规则与存储定义 #

• 解析规则：每条消息首先做标识识别，提取批次唯一ID（如B1S/B1E对应批次ID为1），区分BS、普通业务数据（t1/t2/t3）、BE三类消息
• 分区规则：建议上游生产原始数据时，将批次ID作为消息Key，保证同一批次所有消息进入同一分区，避免流处理时跨实例重分区开销
• 状态存储：使用Kafka Streams自带的键值状态存储（KeyValueStore） 缓存批次数据，Key为批次ID，Value为该批次累积的消息列表

2. 流处理核心逻辑 #

以下为Kafka Streams DSL伪代码示例：

KStream<String, String> sourceStream = builder.stream("原始数据Topic");

sourceStream
  // 按批次ID分组，保证同批次数据由同一实例处理
  .groupBy((batchId, record) -> extractBatchId(record))
  .aggregate(
    () -> new ArrayList<String>(), // 初始化空批次消息列表
    (batchId, record, batchList) -> {
      batchList.add(record);
      return batchList;
    },
    Materialized.as("batch-cache-store") // 绑定状态存储
  )
  .toStream()
  // 仅处理已收到BE标识的完整批次
  .filter((batchId, batchList) -> isBatchCompleted(batchList))
  .flatMap((batchId, batchList) -> {
    // 批次校验与目标Topic匹配
    String beRecord = getLastRecord(batchList);
    String targetTopic;
    if (beRecord.equals("B1E") && checkBatchInvalid(batchId)) {
      targetTopic = "T1";
    } else if (beRecord.equals("B2E")) {
      targetTopic = "T2";
    } else {
      targetTopic = "正常业务Topic";
    }
    // 封装批次所有消息待发送
    return batchList.stream()
      .map(msg -> KeyValue.pair(targetTopic, msg))
      .collect(Collectors.toList());
  })
  // 动态路由到目标Topic
  .to((topic, record, ctx) -> topic);

3. 异常场景兼容 #

• 状态过期清理：给状态存储设置TTL（建议24小时，可按业务批次最大周期调整），超时未收到BE的脏批次自动转入死信队列告警，避免占用存储
• 一致性保证：开启Kafka Streams配置processing.guarantee = exactly_once_v2，实现Exactly Once语义，避免批次数据重复/丢失
• 大批次适配：如果单批次数据量超过内存阈值，可将状态存储切换为RocksDB持久化存储，避免OOM问题

轻量备选方案（小业务量场景） #

如果不想引入流处理框架，也可以用原生Kafka Consumer+Producer实现：

• 消费者拉取消息后，用本地缓存/Redis存储未完成的批次数据
• 收到BE标识后完成校验，将整个批次数据批量发送到目标Topic
• 提交消费Offset，清理对应批次的缓存

该方案需要自行实现消费幂等、状态持久化、故障转移逻辑，仅适合业务量小、可用性要求不高的场景

内容的提问来源于stack exchange，提问作者Manish Kumar