📚 课程目录第2章实时数据源接入

2. 实时数据源接入：Kafka 核心概念、生产者/消费者 API、消息序列化与反序列化（Avro/Protobuf）

好，我们正式开始第二章。实时特征计算，数据从哪来？绝大多数场景下，第一站就是 Kafka。你可以把它想象成实时数据的高速公路——所有业务日志、埋点、数据库变更，都先往这条路上扔，下游的 Flink、Spark Streaming 再从路上取。

这一章，我不打算给你念 Kafka 官方文档。我会把重点放在：你在做实时特征工程时，必须搞懂的 Kafka 核心概念，以及生产者和消费者 API 怎么用才不出坑。最后，我会重点聊聊序列化——这块选错了，后面性能调优会非常痛苦。

本章知识体系速览

下面这张图，是我自己画的知识结构。你可以先扫一眼，心里有个全局感。

2.1 Kafka 核心概念：不只是消息队列

很多人把 Kafka 当成普通消息队列用，其实有点浪费。Kafka 的设计哲学是日志即存储——消息写进去就不会丢，还能反复读。这一点对实时特征计算特别重要，因为特征数据经常需要回溯重算。

2.1.1 Topic 与 Partition

Topic 是逻辑分类，Partition 是物理分片。我习惯把 Partition 理解为「写日志的笔记本」——一个 Topic 有多个笔记本，每个笔记本里的消息是有序的，但笔记本之间不保证顺序。

这里有个关键点：Partition 数量决定了并行度。你的 Flink 算子并行度、消费者线程数，通常都和 Partition 数挂钩。我见过一个项目，Topic 只配了 3 个 Partition，下游 Flink 开了 10 个并行度——结果 7 个 Task 一直空闲，白白浪费资源。

概念	说明	特征工程中的影响
Topic	消息的逻辑分类	一个特征源对应一个 Topic
Partition	物理分片，内部有序	决定消费并行度，影响特征延迟
Offset	消息在 Partition 内的序号	用于断点续传、特征回溯
ISR	同步副本集合	影响数据可靠性，特征不能丢

2.1.2 消费位移与 Consumer Group

Consumer Group 是 Kafka 实现「一条消息只被组内一个消费者处理」的机制。说白了，就是负载均衡。但这里有个坑：如果 Group 内的消费者数量超过 Partition 数，多余的消费者会闲置。我曾经排查过一个线上问题，20 个消费者进程，Topic 只有 8 个 Partition，结果 12 个进程一直在空转，CPU 还居高不下——浪费啊。

我的习惯：Consumer Group 的消费者数量，永远等于 Partition 数。除非你有特殊的多线程消费需求，否则别打破这个规则。

2.2 生产者 API：怎么写才稳？

生产者 API 看起来简单——就一个 send() 方法。但实际用起来，细节不少。我直接给出一段我常用的生产代码模板，然后讲几个关键点。

// 生产者配置
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "io.confluent.kafka.serializers.KafkaAvroSerializer");
props.put("acks", "all");                 // 等待所有副本确认
props.put("retries", 3);                  // 重试次数
props.put("enable.idempotence", true);    // 开启幂等
props.put("compression.type", "snappy");  // 压缩

KafkaProducer<String, GenericRecord> producer = new KafkaProducer<>(props);

// 异步发送 + Callback
producer.send(new ProducerRecord<>("feature-user-click", key, avroRecord),
    (metadata, exception) -> {
        if (exception != null) {
            log.error("发送失败，key={}", key, exception);
            // 这里可以写入死信队列或本地缓存重试
        } else {
            log.debug("发送成功，offset={}", metadata.offset());
        }
    });

2.2.1 异步发送必须带 Callback

很多人图省事，直接 producer.send(record) 不写 Callback。结果消息丢了都不知道。我建议：生产环境永远不要用 fire-and-forget 模式。Callback 里至少打一条失败日志，方便排查。

2.2.2 幂等与事务

实时特征场景下，消息重复是家常便饭。网络抖动、Leader 选举，都可能导致 Producer 重试，从而产生重复消息。enable.idempotence=true 可以保证单个 Partition 内不重复。如果你需要跨 Partition 的 Exactly-Once，那就得上事务 API 了。

注意：事务 API 会带来额外的性能开销。我一般只在「特征数据用于金融风控」这种场景下才开启。普通推荐场景，幂等 + 下游去重就够了。

2.3 消费者 API：消费位移是命根子

消费者 API 的核心就两件事：拉取消息和提交位移。位移提交错了，轻则重复消费，重则丢数据。

2.3.1 自动提交 vs 手动提交

自动提交省事，但不可控。默认 5 秒提交一次，如果这 5 秒内进程挂了，重启后就会重复消费最近 5 秒的数据。对于特征计算来说，重复数据可能导致特征值异常。

我个人的习惯是：手动提交，而且是在业务处理完成之后提交。下面这段代码，是我在多个项目中用过的模式：

while (true) {
    ConsumerRecords<String, GenericRecord> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, GenericRecord> record : records) {
        try {
            // 1. 处理消息：更新特征状态
            processFeature(record.value());
            // 2. 处理成功，记录当前 offset
            lastProcessedOffset = record.offset();
        } catch (Exception e) {
            // 3. 处理失败，记录死信，继续下一条
            sendToDlq(record);
            log.warn("消息处理失败，offset={}", record.offset(), e);
        }
    }
    // 4. 批量提交位移
    consumer.commitSync();
}

2.3.2 Rebalance 监听

Consumer Group 发生 Rebalance 时，分区会重新分配。如果你在 Rebalance 之前没有提交位移，消费进度就丢了。我建议实现 ConsumerRebalanceListener，在分区被撤销前提交一次位移：

consumer.subscribe(Collections.singletonList("feature-user-click"), new ConsumerRebalanceListener() {
    @Override
    public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
        // 撤销前提交位移
        consumer.commitSync();
        log.info("分区 {} 被撤销，已提交位移", partitions);
    }

    @Override
    public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
        log.info("分区 {} 已分配", partitions);
    }
});

2.4 序列化与反序列化：选型决定生死

好，终于到了这一章我最想聊的部分。序列化选型，直接决定了你的特征数据体积、解析速度和Schema 演化的成本。

2.4.1 三种主流方案对比

方案	序列化速度	数据体积	Schema 演化	生态支持
JSON	慢	大（含字段名）	弱（需手动兼容）	通用
Avro	中等	小（二进制）	强（Schema Registry）	Kafka 原生支持
Protobuf	快	极小	中等（需管理 proto 文件）	跨语言优秀

我个人在实时特征场景下的选择标准：

如果团队已经用了 Confluent Schema Registry，无脑选 Avro。Schema 演化、兼容性检查都现成的。
如果对性能有极致要求，比如每秒百万级消息，选 Protobuf。它的编解码速度确实比 Avro 快一截。
JSON 呢？ 我只有在快速原型、调试阶段才用。生产环境用 JSON 做特征数据，带宽和解析开销都扛不住。

2.4.2 Avro 实战：Schema Registry 是灵魂

Avro 的精髓在于 Schema 与数据分离。消息里只存二进制数据，Schema 存在 Registry 里。这样消息体积小，而且 Schema 可以向后兼容。

我曾经遇到过一个经典问题：上游特征表新增了一个字段，下游消费端没更新 Schema，结果反序列化直接抛异常。后来用了 Schema Registry 的兼容性检查（BACKWARD 或 FORWARD 策略），新增字段必须设置默认值，才避免了这类问题。

// Avro Schema 定义（示例）
{
  "type": "record",
  "name": "UserClickFeature",
  "fields": [
    {"name": "user_id", "type": "string"},
    {"name": "item_id", "type": "string"},
    {"name": "click_time", "type": "long"},
    {"name": "click_count", "type": "int", "default": 0}  // 新增字段，必须给默认值
  ]
}

避坑指南：Schema Registry 的兼容性策略，我建议从项目一开始就定好。中途改策略，历史 Schema 可能全得重建。我个人偏好 BACKWARD 策略——消费者可以先升级，生产者后升级，对线上影响最小。

2.4.3 Protobuf 实战：管理好 .proto 文件

Protobuf 的优势是强类型和跨语言。在微服务架构中，特征数据可能被 Java、Python、Go 多个语言消费，Protobuf 的代码生成机制就很舒服。

但 Protobuf 也有个麻烦：.proto 文件的管理。我见过一个项目，.proto 文件散落在各个 Git 仓库里，版本完全对不上。后来我们统一放到一个独立的 Schema 仓库，用 CI 流程自动编译和发布，才解决了这个问题。

// Protobuf 定义（示例）
syntax = "proto3";
package feature;

message UserClickFeature {
  string user_id = 1;
  string item_id = 2;
  int64 click_time = 3;
  int32 click_count = 4;  // 新增字段，使用新编号
}

2.5 本章小结

这一章的内容，说白了就是实时特征计算的「数据入口」怎么搭。Kafka 的核心概念要理解透，尤其是 Partition 和 Offset 对并行度和可靠性的影响。生产者和消费者 API 的代码模式要形成肌肉记忆——异步发送带 Callback、手动提交位移、监听 Rebalance。序列化选型上，Avro 和 Protobuf 是主流，JSON 只适合调试。