Producer 原理与核心配置

主线程 (你的业务线程)
  send(record)
    |
    v
  [拦截器 Interceptors]        onSend() 加工消息 (打标、埋点)
    |
    v
  [序列化器 Serializer]        key/value 对象 -> byte[]
    |
    v
  [分区器 Partitioner]         决定去哪个分区 (有 key 按哈希)
    |
    v
  [RecordAccumulator 缓冲区]   按 "topic-分区" 维度攒批
    |  buffer.memory=32MB       batch1[msg,msg,..] batch2[..]
    |                                 |
- - | - - - - - - - - - - - - - - - - | - - - - - -
    v                                 v
Sender 线程 (IO 线程)          批次满(batch.size) 或到时(linger.ms)
  把就绪批次组装成请求 ──> 网络发送到分区 leader 所在 broker
  等待 acks 响应 ──> 成功: 执行回调 / 失败: 重试或抛错

linger.ms=0:    [m][m][m][m]         每条几乎单独一个请求, 请求数多
linger.ms=20:   [m m m m m m m][...] 攒成大批次, 请求数骤减, 吞吐up

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;
import java.util.Properties;
import java.util.concurrent.Future;
 
public class OrderProducer {
    public static void main(String[] args) throws Exception {
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        // 攒批与压缩
        props.put(ProducerConfig.BATCH_SIZE_CONFIG, 32768);        // 32KB
        props.put(ProducerConfig.LINGER_MS_CONFIG, 20);            // 等 20ms
        props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "lz4");
        props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 64 * 1024 * 1024L);
 
        Producer<String, String> producer = new KafkaProducer<>(props);
        ProducerRecord<String, String> record =
            new ProducerRecord<>("orders", "user-1", "{\"orderId\":1001}");
 
        // 方式1: 发后即忘 —— 最快, 可能丢消息且无感知
        producer.send(record);
 
        // 方式2: 同步发送 —— 逐条确认, 吞吐最低
        RecordMetadata meta = producer.send(record).get();
        System.out.printf("sync -> partition=%d offset=%d%n",
            meta.partition(), meta.offset());
 
        // 方式3: 异步 + 回调 —— 生产推荐, 高吞吐且能感知失败
        producer.send(record, (metadata, exception) -> {
            if (exception != null) {
                // 记日志/告警/落补偿表, 不要吞掉
                exception.printStackTrace();
            } else {
                System.out.printf("async -> partition=%d offset=%d%n",
                    metadata.partition(), metadata.offset());
            }
        });
 
        producer.flush();   // 强制清空缓冲区
        producer.close();   // 会先 flush 再关闭
    }
}

from confluent_kafka import Producer
 
producer = Producer({
    "bootstrap.servers": "localhost:9092",
    "batch.size": 32768,
    "linger.ms": 20,
    "compression.type": "lz4",
})
 
def on_delivery(err, msg):
    if err is not None:
        print(f"delivery failed: {err}")
    else:
        print(f"delivered to {msg.topic()}[{msg.partition()}]@{msg.offset()}")
 
for i in range(10):
    producer.produce(
        topic="orders",
        key=f"user-{i % 3}",
        value=f'{{"orderId": {1000 + i}}}',
        on_delivery=on_delivery,
    )
    producer.poll(0)   # 触发已完成发送的回调
 
producer.flush()       # 退出前必须 flush