4. Kafka Producer基础

好，咱们进入实战手册的第四章。前面几章我们把环境搭好了，概念也理清了。这一章，咱们要亲手往Kafka里写数据了。

说白了，Flink跟Kafka集成，一半的功夫都在Producer上。你数据写得对不对、快不快、稳不稳，全看Producer配得好不好。我见过太多人，代码跑起来数据丢了，或者压力一上来就背压，最后查下来都是Producer参数没调对。

这一章，咱们就围绕三个核心点展开：核心参数、序列化器、分区策略。把这三点吃透了，你写Kafka就跟写本地文件一样踏实。

4.1 Flink Kafka Producer 核心参数

先看一段最基础的代码，感受一下Producer怎么用：

DataStream<String> stream = ...;

stream.addSink(new FlinkKafkaProducer<>(
    "my-topic",                    // 目标topic
    new SimpleStringSchema(),      // 序列化器
    properties,                    // 生产者配置
    FlinkKafkaProducer.Semantic.EXACTLY_ONCE  // 语义保证
));

嗯，看起来很简单对吧？但真正的坑都在那个 properties 里。我列几个我每次必调的参数，你直接拿去用：

参数名 默认值 我的推荐值 说明
bootstrap.servers 无 必填 Kafka集群地址，至少写两个broker
acks 1 all 数据可靠性核心参数，all最安全
batch.size 16384 (16KB) 32768 (32KB) 批次大小，调大能提升吞吐
linger.ms 0 100 等待批次填满的时间，减少小请求
buffer.memory 33554432 (32MB) 67108864 (64MB) Producer缓冲区大小，扛突发流量
retries 2147483647 3 重试次数，配合retry.backoff.ms使用
compression.type none snappy 压缩算法，降低网络开销

我的经验： acks=all 配合 enable.idempotence=true 是生产环境的黄金组合。前者保证数据不丢，后者保证数据不重。但注意，开启幂等性后，retries 默认会变得很大，你最好显式设一个上限，比如3次。

小技巧： 如果你对延迟特别敏感，比如实时风控场景，可以把 linger.ms 设成0，让数据立即发出去。但代价是吞吐会下降，因为每个请求都只带很少的数据。

4.2 序列化器：把数据变成Kafka能懂的样子

Kafka只认字节数组。你的Java对象、JSON字符串、甚至是一坨二进制数据，都得通过序列化器转成byte[]。Flink自带了几个常用的序列化器，但说实话，大部分场景下你都得自己写。

4.2.1 内置序列化器

先看看Flink给我们准备了什么：

  • SimpleStringSchema：最简单的，直接调 toBytes()。适合纯文本数据。
  • JsonSerializationSchema：把对象转成JSON。但注意，它用的是Flink自己的JSON工具，功能有限。
  • TypeInformationSerializationSchema：用Flink的类型系统做序列化。效率高，但数据只能被Flink消费。
  • AvroSerializationSchema：配合Avro使用，适合Schema演进的场景。

我个人用得最多的是 自定义序列化器。为什么？因为内置的要么太简单，要么太死板。比如你有个 UserEvent 对象，想序列化成JSON字符串，但字段名要跟下游约定好，这时候就得自己动手了。

4.2.2 自定义序列化器实战

写一个自定义序列化器，核心就是实现 KafkaSerializationSchema 接口。看代码：

<code>public class UserEventSerializationSchema 
    implements KafkaSerializationSchema<UserEvent> {

    private static final ObjectMapper mapper = new ObjectMapper();
    private String topic;

    public UserEventSerializationSchema(String topic) {
        this.topic = topic;
    }

    @Override
    public ProducerRecord<byte[], byte[]> serialize(
            UserEvent event, @Nullable Long timestamp) {
        try {
            byte[] value = mapper.writeValueAsBytes(event);
            // 这里可以自定义key，比如按用户ID分区
            String key = String.valueOf(event.getUserId());
            return new ProducerRecord<>(topic, key.getBytes(), value);
        } catch (Exception e) {
            throw new RuntimeException("序列化失败", e);
        }
    }
}

避坑指南： 我曾经在序列化器里用了 Gson，结果发现它默认把null字段也序列化进去了，下游解析直接报错。后来换成 Jackson 并配置 @JsonInclude(Include.NON_NULL) 才解决。所以，序列化器的行为一定要跟下游对齐，别想当然。

4.3 分区策略：数据该去哪个分区？

Kafka的Topic由多个分区组成。数据写到哪个分区，直接决定了后续的消费顺序和并行度。Flink Kafka Producer默认的分区策略是：

  • 如果消息带了key，按key的哈希值分区。
  • 如果没带key，轮询（Round-Robin）分配。

但实际场景中，这个默认策略往往不够用。比如你希望同一个用户的所有事件都进同一个分区，保证顺序消费。这时候就得自定义分区器了。

3.3.1 自定义分区器

实现 FlinkKafkaPartitioner 接口，重写 partition() 方法：

public class UserIdPartitioner extends FlinkKafkaPartitioner<UserEvent> {

    @Override
    public int partition(
            UserEvent record, 
            byte[] key, 
            byte[] value, 
            String targetTopic, 
            int[] partitions) {
        
        // 按用户ID取模，保证同一用户进同一分区
        int partitionId = Math.abs(record.getUserId() % partitions.length);
        return partitionId;
    }
}

然后在构造Producer时传入：

stream.addSink(new FlinkKafkaProducer<>(
    "my-topic",
    new UserEventSerializationSchema("my-topic"),
    new UserIdPartitioner(),   // 传入自定义分区器
    properties,
    FlinkKafkaProducer.Semantic.EXACTLY_ONCE
));

注意： 自定义分区器里，partitions 数组是当前topic的所有分区ID。如果你的topic分区数变了（比如扩容），这个数组也会动态更新。但Flink内部会缓存分区信息，不会每次发消息都去查，所以不用担心性能问题。

3.3.2 分区策略选型建议

场景 推荐策略 原因
数据无顺序要求 默认轮询 负载最均衡，吞吐最高
按业务ID保证顺序 自定义哈希分区 同一ID的数据进同一分区，保证局部有序
需要均匀分布 自定义均匀分区 避免数据倾斜，比如按时间戳取模
与外部系统对齐 自定义业务分区 比如按地域、按机房，方便下游消费

我的习惯： 只要数据有业务含义的key，我一定自定义分区器。因为默认的哈希分区虽然均匀，但一旦topic扩容，哈希结果会变，历史数据的分区归属就乱了。自定义分区器可以让你控制这个逻辑，比如用 Math.abs(key.hashCode()) % partitions.length 这种稳定算法。

4.4 本章知识体系总览

下面这张图，把Producer的核心脉络串起来了。你写代码前看一眼，心里就有数了：

Flink Kafka Producer 核心知识体系 Flink Kafka Producer 核心知识体系 Flink DataStream FlinkKafkaProducer Sink 核心参数配置 acks=all batch.size / linger.ms buffer.memory / retries compression.type 序列化器 SimpleStringSchema JsonSerializationSchema 自定义 KafkaSerializationSchema 注意：序列化器决定数据格式 分区策略 默认：有key哈希，无key轮询 自定义 FlinkKafkaPartitioner 按业务ID / 时间戳 / 地域 保证局部有序或负载均衡 Kafka Topic (Partition 0..N) 三者配合：参数保可靠性 + 序列化器定格式 + 分区策略控分布

嗯，这张图把Producer的三个核心模块串起来了。你写代码的时候，脑子里要有这根线：参数保证可靠性，序列化器决定数据长什么样，分区策略控制数据往哪去。三者缺一不可。

← 上一章 📖 返回目录 下一章 →