3. Kafka Java API实战

好，咱们直接进入正题。Kafka 的 Java API 是跟 Flink 打交道的基础，你想想看，如果连消息都发不对、收不对，后面实时计算全是白搭。这一章我带你手写几个核心场景，把 Producer 和 Consumer 的底裤都扒干净。

3.1 Producer API：同步发送与异步发送

Producer 是 Kafka 的入口，说白了就是把数据塞进 Broker。我个人习惯把发送方式分成两种：同步发送和异步发送。别小看这个选择，它直接影响你的吞吐量和数据可靠性。

3.1.1 同步发送

同步发送就是调用 send() 后，等 Broker 返回确认再继续。代码写起来很直白：

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");

Producer<String, String> producer = new KafkaProducer<>(props);
ProducerRecord<String, String> record = new ProducerRecord<>("my-topic", "key1", "value1");

try {
    // 同步发送：get() 会阻塞直到收到 ack
    RecordMetadata metadata = producer.send(record).get();
    System.out.println("发送成功，offset: " + metadata.offset());
} catch (Exception e) {
    e.printStackTrace();
} finally {
    producer.close();
}

这里有个坑——.get() 会抛出 InterruptedException 和 ExecutionException。我曾经在线上环境忘了捕获 ExecutionException，结果消息发送失败时线程直接挂掉，排查了半天才发现是序列化异常。嗯，这里要注意，一定要处理异常。

3.1.2 异步发送

异步发送通过回调函数处理结果，不阻塞主线程。代码长这样：

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");
// 异步发送建议开启压缩，减少网络开销
props.put("compression.type", "snappy");

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

for (int i = 0; i < 1000; i++) {
    ProducerRecord<String, String> record = new ProducerRecord<>("my-topic", "key" + i, "value" + i);
    producer.send(record, (metadata, exception) -> {
        if (exception != null) {
            System.err.println("发送失败: " + exception.getMessage());
            // 这里可以加重试逻辑，比如写入本地文件
        } else {
            System.out.println("发送成功，partition: " + metadata.partition() + ", offset: " + metadata.offset());
        }
    });
}
producer.close();

你想想看，异步发送的核心优势是什么？不阻塞。Producer 内部有一个缓冲区，消息先攒着，后台线程批量发送。这样吞吐量能轻松到每秒几万条。

同步和异步怎么选？我列个对比表，你一看就明白：

特性	同步发送	异步发送
吞吐量	低（每秒几千条）	高（每秒几万条）
数据可靠性	高（每条都确认）	中（依赖回调处理失败）
代码复杂度	简单	中等（需要处理回调）
适用场景	重要数据、小流量	高吞吐、日志采集

3.2 Consumer API：自动提交 vs 手动提交 offset

Consumer 是消费端，核心问题就是 offset 怎么提交。offset 是 Kafka 用来记录消费进度的，提交错了，要么丢数据，要么重复消费。我见过太多人在这上面栽跟头。

3.2.1 自动提交 offset

自动提交最简单，Kafka 每隔一段时间自动提交当前消费的 offset。配置一下就行：

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "my-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", "true");
// 自动提交间隔，默认 5 秒
props.put("auto.commit.interval.ms", "5000");

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

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
    for (ConsumerRecord<String, String> record : records) {
        System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
    }
    // 不需要手动提交，Kafka 自动处理
}

自动提交有什么问题？可能会丢数据。举个例子：你 poll 了一批消息，处理到一半程序挂了，但 offset 已经自动提交了。重启后，Kafka 认为你已经消费完了，那些没处理完的消息就丢了。

3.2.2 手动提交 offset

手动提交把控制权交给你。处理完一批消息后，你再决定什么时候提交。有两种方式：同步提交和异步提交。

同步提交：

props.put("enable.auto.commit", "false");
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Arrays.asList("my-topic"));

try {
    while (true) {
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
        for (ConsumerRecord<String, String> record : records) {
            // 处理消息
            process(record);
        }
        // 处理完后手动提交
        consumer.commitSync();
    }
} catch (Exception e) {
    e.printStackTrace();
} finally {
    consumer.close();
}

异步提交：

props.put("enable.auto.commit", "false");
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Arrays.asList("my-topic"));

try {
    while (true) {
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
        for (ConsumerRecord<String, String> record : records) {
            process(record);
        }
        // 异步提交，不阻塞
        consumer.commitAsync((offsets, exception) -> {
            if (exception != null) {
                System.err.println("提交失败: " + exception.getMessage());
            }
        });
    }
} catch (Exception e) {
    e.printStackTrace();
} finally {
    consumer.close();
}

自动和手动怎么选？看这张表：

特性	自动提交	手动提交
开发效率	高（一行配置）	低（需要写提交代码）
数据可靠性	低（可能丢数据）	高（精确控制）
重复消费风险	中（重启后可能重复）	低（提交前挂了会重试）
适用场景	日志采集、可容忍少量丢失	金融、交易等精确一次语义

3.3 自定义分区器

Kafka 默认分区器是按 key 的 hash 值取模。但实际业务中，你可能需要按业务逻辑路由。比如，我想把同一个用户的订单都发到同一个分区，方便后续顺序消费。这时候就需要自定义分区器。

实现一个自定义分区器很简单，实现 Partitioner 接口：

import org.apache.kafka.clients.producer.Partitioner;
import org.apache.kafka.common.Cluster;
import java.util.Map;

public class UserIdPartitioner implements Partitioner {

    @Override
    public int partition(String topic, Object key, byte[] keyBytes,
                         Object value, byte[] valueBytes, Cluster cluster) {
        // 获取分区数
        Integer partitions = cluster.partitionCountForTopic(topic);
        if (partitions == null) {
            return 0;
        }
        // 按用户 ID 取模，保证同一用户的数据进入同一分区
        String userId = (String) key;
        int partition = Math.abs(userId.hashCode()) % partitions;
        return partition;
    }

    @Override
    public void close() {
        // 清理资源，比如关闭连接
    }

    @Override
    public void configure(Map<String, ?> configs) {
        // 可以读取配置参数
    }
}

然后在 Producer 里配置：

props.put("partitioner.class", "com.example.UserIdPartitioner");

你想想看，自定义分区器有什么用？解决数据倾斜。比如默认分区器按 key hash，如果某些 key 的数据量特别大，会导致某个分区负载过高。你可以自定义分区器，把大 key 的数据打散到多个分区。

好了，Producer 的同步/异步发送、Consumer 的自动/手动提交 offset、自定义分区器，这三个核心点都讲完了。你在项目里可以根据实际场景灵活组合。记住一句话：没有银弹，只有最适合的方案。