实时数据源接入：消息队列（Kafka）基础、生产者与消费者API、数据序列化选择

好，咱们进入第二章。这一章很关键，它是整个实时特征工程的「水源」。没有稳定、高效的数据源接入，后面所有的特征计算、模型推理都是空中楼阁。

我个人习惯把实时数据源接入比作「修水管」——水管够粗、接头够牢、水质够纯，下游才能放心用。今天咱们就重点聊三个核心问题：Kafka 到底怎么玩、生产者和消费者 API 怎么用才不踩坑、以及Avro 和 Protobuf 到底选哪个。

2.1 Kafka 基础：不只是个「消息管道」

很多人觉得 Kafka 就是个「先进先出的消息队列」。嗯，这么说也没错，但太浅了。你想想看，Kafka 之所以能成为实时特征工程的事实标准，靠的是它持久化 + 高吞吐 + 可回溯这三个特性。

我在项目中遇到过好几次这样的场景：下游特征服务挂了半小时，重启后数据全丢了。如果用 RabbitMQ 或 Redis Stream，基本就凉了。但 Kafka 不一样——数据落盘，offset 可控，你随时可以从上次中断的地方继续消费。这就是它不可替代的地方。

核心概念速览

概念	一句话解释	在特征工程中的角色
Topic	消息的逻辑分类	一个 Topic 对应一路特征数据流
Partition	Topic 的分片，并行度单位	决定特征计算的并发上限
Offset	消息在 Partition 内的序号	用于精确回溯和断点续传
Consumer Group	消费者的逻辑分组	保证每条消息只被一个特征服务实例处理
Broker	Kafka 服务器节点	集群规模决定了整体吞吐能力

2.2 生产者 API：把数据「塞」进 Kafka

生产者 API 看起来简单，不就是 producer.send() 嘛？但实际用起来，坑比想象的多。

先看一段最基础的生产者代码。我习惯用 Java，但思路是通用的：

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
// 关键参数：acks
props.put("acks", "all");
// 关键参数：重试
props.put("retries", 3);
// 关键参数：幂等
props.put("enable.idempotence", true);

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

// 发送消息
producer.send(new ProducerRecord<>("feature-click", "user_1001", "{\"event\":\"click\",\"ts\":1700000000}"),
    (metadata, exception) -> {
        if (exception != null) {
            // 记录失败日志，别吞异常
            log.error("发送失败", exception);
        }
    });

producer.flush();
producer.close();

这里有几个点我想重点说一下：

• acks=all：保证 Leader 和所有 ISR 副本都写入后才返回。特征数据丢不起，所以别用 acks=0 或 1。
• enable.idempotence=true：开启幂等性，防止网络重试导致消息重复。我曾经因为这个参数没开，线上特征数据翻了一倍，排查了整整一个下午。
• Callback 不能省：异步发送必须带回调，否则失败了你完全不知道。别问我为什么强调这个——线上 silent data loss 是最可怕的。

2.3 消费者 API：把数据「拉」出来

消费者这边，核心是 poll() 循环和 commit 策略。我见过太多人把消费者写成了「死循环 + 不提交 offset」，结果重启后重复消费几十万条数据。

看一个稳健的消费者示例：

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "feature-click-group");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
// 关键：自动提交关掉
props.put("enable.auto.commit", "false");
// 从最早的消息开始消费
props.put("auto.offset.reset", "earliest");

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Arrays.asList("feature-click"));

try {
    while (true) {
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
        for (ConsumerRecord<String, String> record : records) {
            // 处理特征数据
            processFeature(record.value());
        }
        // 手动提交，确保处理完再提交
        consumer.commitSync();
    }
} finally {
    consumer.close();
}

这里有几个设计原则：

• 手动提交 offset：自动提交是「处理前提交」，万一处理到一半挂了，重启后数据就丢了。手动提交是「处理后提交」，虽然可能重复，但绝不丢失。特征工程里，我们宁可重复处理，也不能丢数据。
• poll 间隔要合理：100ms 是个不错的起点。太短了空转浪费 CPU，太长了延迟会变高。
• 消费线程数：一个 Partition 只能被一个 Consumer 实例消费。所以你的 Consumer 实例数不要超过 Partition 数，否则会有实例闲着。

2.4 数据序列化：Avro vs Protobuf

好，终于到了序列化选型。说实话，这是实时特征工程里最容易「翻车」的环节之一。很多人直接用 JSON 或 String 序列化，图省事。但一旦特征数量上来了、Schema 开始演进了，JSON 的解析开销和兼容性问题会让你痛不欲生。

我个人的建议是：生产环境不要用 JSON 做特征数据的序列化。原因有三：

1. 解析慢：JSON 是文本协议，每次都要做字符串解析，CPU 开销大。
2. Schema 无约束：字段名写错了、类型变了，运行时才能发现。
3. 体积大：同样的数据，JSON 比 Avro/Protobuf 大 3~5 倍，浪费带宽和存储。

那 Avro 和 Protobuf 怎么选？我直接给结论：

维度	Avro	Protobuf
Schema 管理	自带 Schema Registry，天然支持	需要额外搭建 Schema Registry
序列化性能	中等，但反序列化快	更快，整体性能略优
数据体积	较小，自带压缩	更小，二进制紧凑
Schema 演进	强，支持向前/向后兼容	强，但需要手动管理字段编号
Kafka 生态集成	原生支持，Confluent 全家桶	需要额外适配
学习曲线	中等	较低，文档丰富

我的选择逻辑是这样的：

• 如果你在用 Confluent Kafka 或需要 Schema Registry：无脑选 Avro。它和 Kafka 的集成是最自然的，Schema 存储在 Registry 里，消息体只带一个 Schema ID，非常优雅。
• 如果你对性能有极致要求，或者团队更熟悉 Protobuf：选 Protobuf。它的序列化/反序列化速度确实比 Avro 快 10%~20%，而且 gRPC 生态里 Protobuf 是标配。
• 如果你只是做原型验证：先用 JSON 也没问题，但上线前一定要换掉。别问我为什么——我见过一个团队用 JSON 跑了半年，最后 Schema 改了 30 多次，每次都要写兼容代码，苦不堪言。

2.5 实战：用 Avro 序列化特征数据

最后，咱们看一个完整的 Avro 序列化示例。假设我们要发送一个「用户点击特征」的消息：

// 1. 定义 Avro Schema (user_click.avsc)
{
  "type": "record",
  "name": "UserClick",
  "namespace": "com.feature.avro",
  "fields": [
    {"name": "user_id", "type": "string"},
    {"name": "item_id", "type": "string"},
    {"name": "click_time", "type": "long"},
    {"name": "page_type", "type": ["null", "string"], "default": null}
  ]
}

// 2. 生产者使用 Avro 序列化器
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("schema.registry.url", "http://localhost:8081");

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

UserClick click = UserClick.newBuilder()
    .setUserId("user_1001")
    .setItemId("item_5002")
    .setClickTime(System.currentTimeMillis())
    .setPageType("search_result")
    .build();

producer.send(new ProducerRecord<>("feature-click", click.getUserId(), click));
producer.flush();

看到没？生产者这边完全不用关心序列化细节，Avro 序列化器会自动把对象转成二进制，并在消息头带上 Schema ID。消费者那边反序列化时，会自动从 Schema Registry 拉取对应的 Schema 来解析。