Java 生产者骨架
Java 生产者是 Kafka 客户端里最容易被低估的一块：send() 是异步的，方法返回时消息通常还在本地缓冲区里，真正的发送结果要通过回调或返回的 Future 才能知道。所以一个可用的最小示例必须包含三件事——配置好序列化器与确认级别、在回调里处理异常、退出前 flush 并关闭。少了任何一项，线上表现都可能是「程序没报错但消息丢了」。
基本信息
| 依赖坐标 | kafka-clients（版本需与 Broker 兼容） |
|---|---|
| 发送方式 | send() 异步，返回 Future 或使用回调 |
| 关键配置 | bootstrap.servers、key/value.serializer、acks |
| 收尾动作 | flush() + close() 保证缓冲消息发出 |
| 可靠性开关 | enable.idempotence 开启幂等，重试不会产生重复 |
| 线程安全 | Producer 是线程安全的，可多线程共享一个实例 |
最小示例
最小示例
Properties props = new Properties();
props.put("bootstrap.servers", "kafka1: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");
props.put("enable.idempotence", "true");
try (Producer<String, String> producer = new KafkaProducer<>(props)) {
ProducerRecord<String, String> record =
new ProducerRecord<>("orders", "o-1001", "{\"amount\":9.9}");
producer.send(record, (metadata, exception) -> {
if (exception != null) {
System.err.println("send failed: " + exception.getMessage());
} else {
System.out.println(metadata.topic() + "-" + metadata.partition()
+ "@" + metadata.offset());
}
});
producer.flush(); // 确保缓冲中的记录全部发出
}典型使用场景
- 业务异步解耦：把订单、日志等事件投递到 Kafka 供下游消费
- 批量导入：循环 send 后一次 flush，靠客户端自动成批提高吞吐
- 需要确认送达：用 acks=all 配合回调记录失败并告警
- 需要顺序语义：指定 key 让同一业务键落到同一分区
- 高可靠场景：开启幂等并结合事务实现精确一次写入
关键配置读法
| 配置 | 作用 | 注意 |
|---|---|---|
| acks | 确认级别，all 表示所有同步副本确认 | 配合 min.insync.replicas 才完整 |
| enable.idempotence | 幂等写入，避免重试重复 | 开启后对并发请求数有约束 |
| retries | 失败重试次数 | 过大可能堆高延迟 |
| max.block.ms | 缓冲区满时 send 的阻塞上限 | 超时抛 TimeoutException |
| serializer | 键值序列化方式 | 与消费端反序列化器必须对应 |
生产环境建议在回调里记录失败并接入告警；只依赖自动重试会在超时后丢消息。
send() 的异常也可能在回调中才出现（如序列化失败或缓冲区超时），只捕获方法本身的异常是不够的。