3. Kafka生产与消费实战：Java/Python Producer API使用、Consumer Rebalance机制、消息投递语义
好，咱们今天聊点实在的。Kafka 的生产与消费，说白了就是数据管道的入口和出口。很多新手上来就写代码，结果生产端丢消息、消费端重复处理，搞得焦头烂额。我当年也踩过这些坑，今天就把这些经验掰开揉碎了讲给你听。
3.1 Java Producer API：从入门到避坑
先说说 Java 的生产者。我个人习惯用 KafkaProducer 这个类，它线程安全，性能也不错。你想想看，一个生产者实例可以给多个 topic 发消息，省心。
核心配置项：
bootstrap.servers：Broker 地址列表，至少写两个，防止单点故障。key.serializer和value.serializer：序列化器，常用StringSerializer。acks：这个参数很关键，后面细说。retries：重试次数，我一般设 3 次。
来个最简单的示例：
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("acks", "all");
Producer<String, String> producer = new KafkaProducer<>(props);
producer.send(new ProducerRecord<>("my-topic", "key1", "Hello Kafka"));
producer.close();
嗯，这里要注意。很多新手写完 send() 就以为消息发出去了。其实 send() 是异步的，它返回一个 Future。如果你不调用 get()，或者不设置回调，消息可能悄无声息地丢了。
我的习惯：生产环境一定要加回调函数，或者用 send().get() 同步等待。虽然性能会降一点，但心里踏实。
producer.send(new ProducerRecord<>("my-topic", "key1", "Hello Kafka"),
(metadata, exception) -> {
if (exception != null) {
System.err.println("发送失败: " + exception.getMessage());
} else {
System.out.println("发送成功，偏移量: " + metadata.offset());
}
});
3.2 Python Producer API：简洁但别大意
Python 这边，我推荐 confluent-kafka 库。它底层是 C 写的，性能比 kafka-python 好不少。我在项目中遇到过，用 kafka-python 在高吞吐场景下 CPU 飙到 90%，换成 confluent-kafka 直接降到 30%。
from confluent_kafka import Producer
conf = {
'bootstrap.servers': 'localhost:9092',
'acks': 'all',
'retries': 3
}
producer = Producer(conf)
def delivery_report(err, msg):
if err is not None:
print(f'发送失败: {err}')
else:
print(f'发送成功: {msg.topic()} [{msg.partition()}]')
producer.produce('my-topic', key='key1', value='Hello Kafka', callback=delivery_report)
producer.flush()
你看，Python 的写法更简洁。但有个坑——produce() 只是把消息放进缓冲区，真正发送要靠 poll() 或 flush()。我曾经见过一个同事，写了个循环一直 produce() 但不 flush()，结果内存爆了。
避坑指南：生产环境一定要定期调用 producer.poll(0) 来触发回调，或者用 flush() 强制刷出。否则消息会积压在缓冲区，直到溢出。
3.3 Consumer Rebalance 机制：别让你的消费者“打架”
消费者组是 Kafka 的精髓。同一个 group.id 下的消费者，共同消费一个 topic 的所有分区。但问题来了——当消费者加入或退出时，会触发 Rebalance。
Rebalance 说白了就是重新分配分区。这个过程里，所有消费者都会暂停消费，等分配完再继续。我刚开始做的时候，以为 Rebalance 是瞬间完成的，结果有一次线上服务抖动，发现是 Rebalance 耗时 30 秒。
为什么会这样？
因为 Rebalance 期间，消费者要停止拉取消息，提交偏移量，然后等待协调器重新分配。如果消费者数量多，或者分区数多，这个过程就会变长。
怎么优化？我总结了三点：
- 控制消费者数量：消费者数不要超过分区数，否则会有消费者闲置。
- 使用静态成员：Kafka 2.3 之后支持
group.instance.id，可以避免不必要的 Rebalance。 - 调大超时时间：
session.timeout.ms默认 10 秒，如果网络不稳定，可以调到 30 秒。
我的经验：在消费端加一个 ConsumerRebalanceListener，在 Rebalance 前后做一些清理工作。比如在 onPartitionsRevoked 里提交偏移量，避免重复消费。
consumer.subscribe(Arrays.asList("my-topic"), new ConsumerRebalanceListener() {
@Override
public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
consumer.commitSync(); // 提交当前偏移量
}
@Override
public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
// 可以在这里做一些初始化
}
});
3.4 消息投递语义：At-least-once 与 Exactly-once
这是面试常考题，也是生产环境最容易出问题的地方。咱们一个一个说。
3.4.1 At-least-once（至少一次）
说白了就是：消息至少被消费一次，但可能重复。这是 Kafka 的默认行为。生产者设置 acks=all，消费者在消费完再提交偏移量，就能保证至少一次。
但问题来了——如果消费者处理完消息，还没来得及提交偏移量就挂了，重启后会从上次提交的位置重新消费，导致重复处理。
避坑指南：如果你的业务不能容忍重复，比如扣钱、发券，那 At-least-once 就不够用。你需要做幂等处理，或者在消费端去重。
3.4.2 Exactly-once（精确一次）
精确一次是终极目标。Kafka 从 0.11 版本开始支持幂等生产者和事务。
幂等生产者：设置 enable.idempotence=true，Kafka 会自动去重。原理是给每条消息加一个序列号，Broker 端检查序列号，重复的就不写入。
props.put("enable.idempotence", true);
props.put("acks", "all");
props.put("retries", Integer.MAX_VALUE);
事务：如果你需要跨分区、跨 topic 的精确一次，那就得用事务。生产者设置 transactional.id，然后调用 initTransactions()、beginTransaction()、commitTransaction()。
我的建议：大多数场景下，幂等生产者就够了。事务的开销比较大，除非你真的需要跨分区原子性，否则别轻易用。
3.5 实战中的选择：到底用哪个？
我整理了一个表格，方便你对比：
| 场景 | 推荐方案 | 说明 |
|---|---|---|
| 日志收集、监控数据 | At-least-once | 丢几条无所谓，性能优先 |
| 订单、支付 | Exactly-once（幂等生产者） | 不能重复，也不能丢 |
| 跨系统数据同步 | Exactly-once（事务） | 需要保证源和目标一致 |
| 实时推荐、风控 | At-least-once + 去重 | 性能与准确性的平衡 |
嗯，最后说一句。Kafka 的生产与消费，看似简单，但真正用好需要理解它的设计哲学。你想想看，一个分布式系统要保证不丢、不重、不乱序，背后有多少权衡。我当年也是踩了无数坑，才慢慢摸透这些门道。
下一章咱们聊聊 Kafka 的存储机制，看看消息到底是怎么落盘的。到时候你会发现，很多性能问题其实都跟存储有关。