4. 生产者原理
生产者是消息管道的源头。你想想看，如果源头就出了问题，后面再好的管道设计也是白搭。这一章我带你深入生产者的内部机制，从客户端开发到消息发送模式，再到那些容易踩坑的配置项。
核心要点：生产者的本质就是一个数据泵，它要把业务数据可靠地送到Kafka集群。但“可靠”这两个字，背后藏着不少门道。
4.1 生产者客户端开发基础
先看一个最基础的生产者代码。我习惯用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");
KafkaProducer<String, String> producer = new KafkaProducer<>(props);
ProducerRecord<String, String> record =
new ProducerRecord<>("my-topic", "key1", "Hello Kafka");
producer.send(record);
producer.close();这段代码看起来简单，但有几个关键点我要提醒你：
- bootstrap.servers：不是只写一个broker地址就完事了。我建议至少写两个，防止单点故障。
- 序列化器：Kafka只存字节数组，所以任何消息都要序列化。我个人习惯用Avro或Protobuf，而不是Java原生序列化。
- ProducerRecord：包含topic、partition（可选）、key、value。key的作用很大，后面会讲。
小技巧：生产者的创建是重量级操作，里面会初始化网络连接、元数据拉取等。别在每次发送消息时都new一个，要复用。
4.2 消息发送模式：同步 vs 异步
send()方法返回的是一个Future对象。这就引出了两种发送模式。
4.2.1 同步发送
try {
RecordMetadata metadata = producer.send(record).get();
System.out.println("消息发送成功，offset: " + metadata.offset());
} catch (Exception e) {
e.printStackTrace();
}调用.get()会阻塞当前线程，直到broker返回确认。这样做的好处是能立即知道发送结果，但吞吐量会大打折扣。我在做日志采集时试过同步发送，TPS直接掉了一个数量级。
4.2.2 异步发送
producer.send(record, new Callback() {
@Override
public void onCompletion(RecordMetadata metadata, Exception exception) {
if (exception != null) {
// 处理发送失败
exception.printStackTrace();
} else {
// 发送成功
System.out.println("offset: " + metadata.offset());
}
}
});异步发送才是生产环境的主流。send()立即返回，后台线程负责实际发送。回调函数会在发送完成后被调用。
注意：回调函数是在生产者内部的IO线程中执行的。不要在回调里做耗时操作，否则会阻塞其他消息的发送。我曾经见过有人把数据库写入放在回调里，结果生产者的吞吐量直接崩了。
4.3 acks与retries配置
这两个配置直接决定了消息的可靠性。我见过太多人在这上面栽跟头。
| acks值 | 含义 | 可靠性 | 性能 |
|---|---|---|---|
| 0 | 不等待任何确认 | 最低 | 最高 |
| 1 | 等待leader确认 | 中等 | 中等 |
| all（或-1） | 等待所有ISR副本确认 | 最高 | 最低 |
说说我的经验：
- acks=0：适合对数据丢失不敏感的场景，比如监控指标采集。丢几条数据无所谓。
- acks=1：默认值。大多数场景够用，但leader挂了的话，未同步到follower的数据会丢失。
- acks=all：金融、交易类场景必须用这个。我曾经帮一个支付系统调优，他们一开始用acks=1，结果一次broker宕机丢了十几笔订单数据。
再来看retries。这个参数控制发送失败时的重试次数。
props.put("retries", 3);
props.put("retry.backoff.ms", 100);重试不是越多越好。重试次数太多会导致消息堆积，而且可能造成消息乱序。为什么？因为重试的消息可能比后面发送的消息先到达broker。
避坑指南：我曾经把retries设成10，结果生产者的发送队列被重试消息塞满了，新的消息根本发不出去。后来我改成3次重试，配合幂等性来解决乱序问题。
4.4 幂等性与事务
这两个概念很多人搞混。我简单说清楚。
4.4.1 幂等性
幂等性解决的是“消息重复”问题。开启方式很简单：
props.put("enable.idempotence", true);开启后，生产者会给每条消息分配一个唯一的序列号（Producer ID + 序列号）。broker会检查这个序列号，如果重复就丢弃。
幂等性的前提条件：
- 必须设置acks=all
- retries必须大于0
- max.in.flight.requests.per.connection必须小于等于5
说白了，幂等性保证的是“单分区内、单会话内”的不重复。跨分区或重启后就不管用了。
4.4.2 事务
事务比幂等性更强。它保证“跨分区、跨会话”的原子性写入。
props.put("transactional.id", "my-transactional-id");
KafkaProducer<String, String> producer = new KafkaProducer<>(props);
producer.initTransactions();
try {
producer.beginTransaction();
producer.send(new ProducerRecord<>("topic1", "key1", "value1"));
producer.send(new ProducerRecord<>("topic2", "key2", "value2"));
producer.commitTransaction();
} catch (Exception e) {
producer.abortTransaction();
}事务的典型场景是“精确一次语义”（Exactly-Once Semantics, EOS）。比如从Kafka读数据，处理后写回Kafka，事务能保证整个过程要么全部成功，要么全部回滚。
警告：事务的开销不小。它会引入额外的协调和提交步骤，吞吐量会下降20%-30%。不是所有场景都需要事务，别滥用。
4.5 核心流程总结
下面这张图展示了生产者发送消息的完整流程：
流程其实不复杂：应用代码创建消息 → 序列化成字节 → 分区器决定去哪个分区 → 放入累加器缓冲 → Sender线程批量发送到broker。
我的建议：生产环境至少开启幂等性（enable.idempotence=true），配合acks=all和适当的retries。这样能在性能和可靠性之间取得较好的平衡。事务只在需要跨分区原子写入时才用。
嗯，生产者这块的内容就讲到这里。记住一个原则：先保证数据不丢，再考虑性能优化。很多线上事故都是因为一开始就追求高性能，忽略了可靠性配置。