1
Future<RecordMetadata> future = producer.send(record);

1
KafkaProducer
2
├── ProducerConfig
3
├── Serializer<Key>
4
├── Serializer<Value>
5
├── ProducerInterceptors
6
├── ProducerMetadata
7
├── RecordAccumulator
8
├── BufferPool
9
├── Sender
10
├── NetworkClient / KafkaClient
11
├── Metrics
12
└── I/O Thread

1
// 作为长期复用、线程安全的客户端实例
2
KafkaProducer<String, OrderEvent> producer = new KafkaProducer<>(props);

1
try (KafkaProducer<String, String> producer = new KafkaProducer<>(props)) {
2
    producer.send(record);
3
}

1
Construct
2
   ↓
3
Bootstrap Metadata
4
   ↓
5
Repeated send()
6
   ↓
7
Background Sender Loop
8
   ↓
9
flush() 可选等待
10
   ↓
11
close()

1
KafkaProducer.send(record)
2
        ↓
3
ProducerInterceptors.onSend
4
        ↓
5
doSend
6
        ↓
7
等待 / 获取 Topic Metadata
8
        ↓
9
Serialize Key / Value
10
        ↓
11
选择 Partition
12
        ↓
13
估算 Record Size 并校验
14
        ↓
15
RecordAccumulator.append
16
        ↓
17
必要时唤醒 Sender
18
        ↓
19
返回 Future<RecordMetadata>

1
ProducerRecord<String, OrderEvent> record = new ProducerRecord<>(
2
    "order-events",
3
    null,                     // partition
4
    System.currentTimeMillis(),
5
    "order-1001",            // key
6
    event,                    // value
7
    headers
8
);

1
Topic
2
Optional Partition
3
Optional Timestamp
4
Optional Key
5
Value
6
Headers

1
public ProducerRecord<String, String> onSend(ProducerRecord<String, String> record) {
2
    remoteAuditService.report(record); // 同步网络调用
3
    return record;
4
}

1
Topic 是否存在？
2
Topic 有多少 Partition？
3
每个 Partition 当前 Leader 是谁？
4
Leader 对应哪个 Broker Node？

1
hash(key) % partitionCount

1
keySerializer.serialize(...)
2
valueSerializer.serialize(...)

1
serializedKey: byte[]
2
serializedValue: byte[]

1
Java Object
2
    ↓
3
Stable Binary Representation

1
Record 是否超过 max.request.size？
2
Record 是否超过 Producer 可接受的 Batch / Request 边界？
3
Header 是否异常膨胀？

1
Producer max.request.size
2
Broker message.max.bytes
3
Topic max.message.bytes
4
Replica fetch size
5
Consumer fetch size

1
bootstrap.servers=kafka-1:9092,kafka-2:9092,kafka-3:9092

1
Topic: order-events
2
P0 Leader → Broker 1
3
P1 Leader → Broker 3
4
P2 Leader → Broker 2

1
orders-P1 Batch
2
      ↓
3
直接发送 Broker 3

1
Cluster Metadata
2
├── Brokers
3
├── Topics
4
│   ├── Partition Count
5
│   └── Partition Metadata
6
│       ├── Leader
7
│       ├── Replicas
8
│       └── ISR 摘要
9
└── Cluster / Version State

1
TopicPartition → Leader Node

1
order-events

1
Application Thread
2
      ↓
3
waitOnMetadata(order-events)
4
      ↓
5
Sender 被唤醒
6
      ↓
7
MetadataRequest
8
      ↓
9
Broker Response
10
      ↓
11
Application Thread 获得 Partition 信息

1
哪些 Record 进入同一有序 Log
2
哪些 Record 可以并行写入
3
哪些 Consumer 后续可以并行处理
4
数据是否形成 Hot Partition

1
1. Record 显式指定 Partition
2
   → 使用显式 Partition
3

4
2. 未指定 Partition，但存在 Key
5
   → 根据 Key Hash 选择 Partition
6

7
3. 既没有 Partition，也没有 Key
8
   → 使用 Sticky / Adaptive 选择策略形成更有效 Batch

1
new ProducerRecord<>("order-events", 2, key, value);

1
Key = orderId

1
OrderCreated(order-1001)
2
OrderPaid(order-1001)
3
OrderShipped(order-1001)

1
R1 → P0
2
R2 → P1
3
R3 → P2
4
R4 → P0

1
一段时间集中写 P1
2
      ↓
3
形成较满 Batch
4
      ↓
5
切换到其他 Partition

1
Partition 扩容怎么办？
2
规则版本如何发布？
3
多个 Producer 是否保持一致？
4
Key 是否稳定？
5
热点如何检测？
6
旧消息和新消息的顺序假设是否仍成立？

1
producer.send(record);

1
Record
2
  ↓
3
Request Header + Protocol Encoding
4
  ↓
5
Socket Write

1
RecordAccumulator

1
Map<TopicPartition, Deque<ProducerBatch>>

1
orders-0 → [Batch A][Batch D]
2
orders-1 → [Batch B]
3
orders-2 → [Batch C][Batch E]

1
batch.size = 16384 bytes

1
每个 ProduceRequest 最大 16 KiB

1
Broker Node 1
2
├── orders-0 Batch
3
├── payments-2 Batch
4
└── audit-7 Batch

1
batch.size
2
≠
3
单条消息绝对最大值

1
estimated serialized size
2
max.request.size
3
Broker / Topic message limits
4
Buffer availability

1
Batch 已满
2
   → 立即 Ready
3

4
Batch 未满但等待时间达到 linger.ms
5
   → Ready
6

7
Broker Backpressure / Connection Not Ready
8
   → 实际等待可能更久

1
Record 1
2
Record 2
3
Record 3
4
      ↓
5
RecordBatch Compression

1
Network Bytes ↓
2
Disk Bytes ↓
3
Broker I/O ↓
4

5
Producer CPU ↑
6
Consumer CPU ↑
7
Latency 可能变化

1
Batch
2
├── Record A → Future A / Callback A
3
├── Record B → Future B / Callback B
4
└── Record C → Future C / Callback C

1
buffer.memory = 33554432 bytes

1
new byte[batch.size]

1
allocate
2
  ↓
3
ProducerBatch
4
  ↓
5
发送完成 / 失败终结
6
  ↓
7
deallocate
8
  ↓
9
返回 Pool

1
Producer Append Rate λ
2
>
3
Sender Completion Rate μ

1
Application Thread
2
      ↓
3
BufferPool.allocate()
4
      ↓
5
等待其他 Batch 完成并释放 Buffer
6
      ↓
7
在剩余 max.block.ms 内成功
8
或
9
超时失败

1
send() latency ↑
2
waiting threads ↑
3
bufferpool-wait-time ↑
4
TimeoutException / BufferExhaustedException

1
短峰值
2
  ↓
3
Buffer 暂时积压
4
  ↓
5
峰值结束后 Sender 追平

1
λ > μ

1
Sender.run()
2
    ↓
3
runOnce()
4
    ↓
5
处理事务 / 幂等状态（存在时）
6
    ↓
7
sendProducerData()
8
    ↓
9
NetworkClient.poll()

1
while (running) {
2
    long pollTimeout = sendProducerData(now);
3
    client.poll(pollTimeout, now);
4
}

1
检查过期 Batch
2
      ↓
3
Accumulator.ready(metadata)
4
      ↓
5
需要时触发 Metadata Refresh
6
      ↓
7
选择 Ready Broker Nodes
8
      ↓
9
Accumulator.drain(...)
10
      ↓
11
构建 ProduceRequest
12
      ↓
13
NetworkClient.send(...)

1
当前 Leader 已知？
2
目标 Broker 连接 Ready？
3
该 Node 的 In-flight Request 是否达到上限？
4
Batch 是否还在 Retry Backoff？

1
orders-0
2
orders-1
3
payments-3

1
orders-0 Leader → Broker 1
2
orders-1 Leader → Broker 2
3
payments-3 Leader → Broker 1

1
Broker 1
2
├── orders-0 Batch
3
└── payments-3 Batch
4

5
Broker 2
6
└── orders-1 Batch

1
ProducerBatch：TopicPartition 级
2
ProduceRequest：Broker Node 级

1
ProduceRequest Broker-1
2
├── Topic orders
3
│   ├── Partition 0 → Batch A
4
│   └── Partition 4 → Batch B
5
└── Topic audit
6
    └── Partition 2 → Batch C

1
max.request.size
2
Broker Node
3
Partition Ordering
4
Available Batches

1
In-flight Request

1
Broker Connection State
2
DNS / Address Resolution
3
Request Header / Correlation ID
4
In-flight Requests
5
Socket Read / Write
6
Authentication / TLS
7
Response Matching
8
Connection Failure
9
Timeout Detection
10
Metadata Request

1
socket.write(bytes);

1
Sender Thread
2
    ↓
3
NetworkClient.poll()
4
    ↓
5
Selector
6
    ├── Broker 1 Channel
7
    ├── Broker 2 Channel
8
    └── Broker 3 Channel

1
ProduceRequest
2
    ↓
3
In-flight
4
    ↓
5
ProduceResponse / Timeout / Disconnect

1
max.in.flight.requests.per.connection = 5

1
Request 1 ───────── waiting
2
Request 2 ───────── waiting
3
Request 3 ───────── waiting

1
Batch A 先发 → 失败
2
Batch B 后发 → 成功
3
Batch A 重试 → 成功

1
B → A

1
Application Thread
2
      ↓ wait metadata
3
ProducerMetadata
4
      ↑ update
5
Sender / NetworkClient

1
Error Code
2
Base Offset
3
Log Append Time
4
Record Errors / Error Message（特定版本）
5
Throttle Time

1
producer.send(record, (metadata, exception) -> {
2
    if (exception != null) {
3
        log.error("send failed", exception);
4
        return;
5
    }
6
    log.info("partition={}, offset={}",
7
        metadata.partition(), metadata.offset());
8
});

1
producer.send(record, (metadata, exception) -> {
2
    slowDatabase.saveResult(metadata); // 慢 I/O
3
});

1
Network Poll 变慢
2
Batch Completion 变慢
3
Buffer 释放变慢
4
send() Buffer Wait 增加

1
max.block.ms = 60000

1
等待 Metadata
2
+
3
等待 BufferPool 内存

1
send() 实际耗时
2
可能 > max.block.ms

1
linger.ms = 5

1
request.timeout.ms = 30000

1
delivery.timeout.ms = 120000

1
Record Enter Producer
2
      ↓
3
Accumulator Wait
4
      ↓ linger / scheduling
5
Network Send
6
      ↓ request.timeout
7
Retry Backoff
8
      ↓
9
More Attempts
10
      ↓
11
Success or delivery.timeout.ms

1
delivery.timeout.ms
2
>=
3
request.timeout.ms + linger.ms

1
HTTP Client 3s 后放弃
2
      ↓
3
Producer 仍在后台重试
4
      ↓
5
最终消息可能成功
6
      ↓
7
调用方却认为业务失败并再次提交

1
User Request Timeout
2
Business Transaction Timeout
3
Producer max.block.ms
4
Producer delivery.timeout.ms
5
Broker / Network Timeout
6
Downstream Idempotency

1
ProduceRequest
2
      ↓
3
Retriable Error
4
      ↓
5
Retry Backoff
6
      ↓
7
Re-enqueue Batch
8
      ↓
9
Refresh Metadata if Needed
10
      ↓
11
Send Again

1
try {
2
    producer.send(record).get();
3
} catch (Exception e) {
4
    producer.send(record); // 新的一次业务发送
5
}

1
ProduceRequest 何时被认为完成

1
Record 何时进入 Accumulator

1
RecordAccumulator
2
      ↓
3
Sender
4
      ↓
5
ProduceRequest(acks)
6
      ↓
7
Broker Leader / Replicas
8
      ↓
9
Response Completion
10
      ↓
11
Future / Callback

1
acks = all
2
retries > 0
3
max.in.flight.requests.per.connection <= 5

1
record-queue-time-avg
2
request-latency-avg / max
3
produce-throttle-time
4
batch-size-avg
5
record-retry-rate
6
connection setup
7
Broker request queue

1
JVM Heap
2
Allocation Rate
3
GC Pause
4
Buffer Available Bytes
5
Waiting Threads
6
In-flight Requests

1
对象存储 + Kafka 引用
2
事件拆分
3
字段裁剪
4
压缩格式
5
Schema 优化

1
Metadata Wait
2
BufferPool Wait
3
Serializer / Interceptor CPU
4
GC

1
Batch Scheduling
2
Network
3
Broker
4
ACK
5
Retry
6
Delivery Timeout

1
生产速率是否持续大于交付速率？
2
Broker 是否 Throttle？
3
Request Latency 是否上升？
4
Retry 是否上升？
5
是否有连接故障？
6
Callback 是否阻塞 Sender？

1
kafka-producer-lab/
2
├── docker-compose.yml
3
├── pom.xml
4
└── src/main/java/dev/jupiter/kafka/
5
    └── ProducerLab.java

1
services:
2
  kafka:
3
    image: apache/kafka:4.3.0
4
    container_name: kafka-ch03
5
    hostname: kafka
6
    ports:
7
      - "9092:9092"
8
    environment:
9
      KAFKA_NODE_ID: 1
10
      KAFKA_PROCESS_ROLES: broker,controller
11
      KAFKA_LISTENERS: PLAINTEXT://:9092,CONTROLLER://:9093
12
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
13
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
14
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
15
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka:9093
16
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
17
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
18
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
19
      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
20
    volumes:
21
      - kafka-data:/var/lib/kafka/data
22

23
volumes:
24
  kafka-data:

1
<project xmlns="http://maven.apache.org/POM/4.0.0"
2
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
3
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
4
    <modelVersion>4.0.0</modelVersion>
5

6
    <groupId>dev.jupiter</groupId>
7
    <artifactId>kafka-producer-lab</artifactId>
8
    <version>1.0.0</version>
9

10
    <properties>
11
        <maven.compiler.release>21</maven.compiler.release>
12
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
13
    </properties>
14

15
    <dependencies>
16
        <dependency>
17
            <groupId>org.apache.kafka</groupId>
18
            <artifactId>kafka-clients</artifactId>
19
            <version>4.3.0</version>
20
        </dependency>
21
        <dependency>
22
            <groupId>org.slf4j</groupId>
23
            <artifactId>slf4j-simple</artifactId>
24
            <version>2.0.17</version>
25
        </dependency>
26
    </dependencies>
27

28
    <build>
29
        <plugins>
30
            <plugin>
31
                <groupId>org.codehaus.mojo</groupId>
32
                <artifactId>exec-maven-plugin</artifactId>
33
                <version>3.5.1</version>
34
                <configuration>
35
                    <mainClass>dev.jupiter.kafka.ProducerLab</mainClass>
36
                </configuration>
37
            </plugin>
38
        </plugins>
39
    </build>
40
</project>

1
package dev.jupiter.kafka;
2

3
import org.apache.kafka.clients.producer.KafkaProducer;
4
import org.apache.kafka.clients.producer.ProducerConfig;
5
import org.apache.kafka.clients.producer.ProducerRecord;
6
import org.apache.kafka.clients.producer.RecordMetadata;
7
import org.apache.kafka.common.Metric;
8
import org.apache.kafka.common.MetricName;
9
import org.apache.kafka.common.serialization.StringSerializer;
10

11
import java.time.Duration;
12
import java.util.Comparator;
13
import java.util.Map;
14
import java.util.Properties;
15
import java.util.concurrent.CountDownLatch;
16
import java.util.concurrent.TimeUnit;
17
import java.util.concurrent.atomic.AtomicInteger;
18

19
public final class ProducerLab {
20

21
    private static final String TOPIC = "producer-lab";
22

23
    public static void main(String[] args) throws Exception {
24
        String mode = args.length == 0 ? "async" : args[0];
25

26
        Properties props = new Properties();
27
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
28
        props.put(ProducerConfig.CLIENT_ID_CONFIG, "producer-ch03-lab");
29
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
30
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
31

32
        props.put(ProducerConfig.ACKS_CONFIG, "all");
33
        props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
34
        props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "zstd");
35
        props.put(ProducerConfig.BATCH_SIZE_CONFIG, 64 * 1024);
36
        props.put(ProducerConfig.LINGER_MS_CONFIG, 20);
37
        props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 32L * 1024 * 1024);
38
        props.put(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, 120_000);
39
        props.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, 30_000);
40
        props.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 10_000);
41

42
        try (KafkaProducer<String, String> producer = new KafkaProducer<>(props)) {
43
            switch (mode) {
44
                case "sync" -> runSync(producer, 20_000);
45
                case "keyed" -> runKeyed(producer, 30);
46
                case "metrics" -> runAsync(producer, 100_000, true);
47
                default -> runAsync(producer, 100_000, false);
48
            }
49

50
            producer.flush();
51
            printMetrics(producer.metrics());
52
        }
53
    }
54

55
    private static void runAsync(
56
            KafkaProducer<String, String> producer,
57
            int count,
58
            boolean printProgress) throws InterruptedException {
59

60
        CountDownLatch latch = new CountDownLatch(count);
61
        AtomicInteger failures = new AtomicInteger();
62
        long start = System.nanoTime();
63

64
        for (int i = 0; i < count; i++) {
65
            String key = "order-" + (i % 10_000);
66
            String value = "{\"seq\":" + i + ",\"type\":\"OrderCreated\"}";
67

68
            producer.send(new ProducerRecord<>(TOPIC, key, value),
69
                (metadata, exception) -> {
70
                    if (exception != null) {
71
                        failures.incrementAndGet();
72
                    }
73
                    latch.countDown();
74
                });
75

76
            if (printProgress && i % 10_000 == 0) {
77
                System.out.println("enqueued=" + i);
78
            }
79
        }
80

81
        if (!latch.await(2, TimeUnit.MINUTES)) {
82
            throw new IllegalStateException("send completion timeout");
83
        }
84

85
        long elapsedMs = Duration.ofNanos(System.nanoTime() - start).toMillis();
86
        System.out.printf("async count=%d elapsed=%dms failures=%d%n",
87
            count, elapsedMs, failures.get());
88
    }
89

90
    private static void runSync(
91
            KafkaProducer<String, String> producer,
92
            int count) throws Exception {
93

94
        long start = System.nanoTime();
95
        for (int i = 0; i < count; i++) {
96
            RecordMetadata metadata = producer.send(
97
                new ProducerRecord<>(TOPIC, "sync-key", "value-" + i)
98
            ).get();
99

100
            if (i % 5_000 == 0) {
101
                System.out.printf("i=%d partition=%d offset=%d%n",
102
                    i, metadata.partition(), metadata.offset());
103
            }
104
        }
105
        long elapsedMs = Duration.ofNanos(System.nanoTime() - start).toMillis();
106
        System.out.printf("sync count=%d elapsed=%dms%n", count, elapsedMs);
107
    }
108

109
    private static void runKeyed(
110
            KafkaProducer<String, String> producer,
111
            int count) throws Exception {
112

113
        for (int i = 0; i < count; i++) {
114
            String key = i % 2 == 0 ? "order-A" : "order-B";
115
            RecordMetadata metadata = producer.send(
116
                new ProducerRecord<>(TOPIC, key, "event-" + i)
117
            ).get();
118

119
            System.out.printf("key=%s partition=%d offset=%d%n",
120
                key, metadata.partition(), metadata.offset());
121
        }
122
    }
123

124
    private static void printMetrics(Map<MetricName, ? extends Metric> metrics) {
125
        metrics.entrySet().stream()
126
            .filter(entry -> {
127
                String name = entry.getKey().name();
128
                return name.contains("batch-size")
129
                    || name.contains("records-per-request")
130
                    || name.contains("compression-rate")
131
                    || name.contains("record-queue-time")
132
                    || name.contains("request-latency")
133
                    || name.contains("buffer-available")
134
                    || name.contains("bufferpool-wait")
135
                    || name.contains("record-retry")
136
                    || name.contains("record-error");
137
            })
138
            .sorted(Comparator.comparing(entry -> entry.getKey().name()))
139
            .forEach(entry -> System.out.printf(
140
                "%s tags=%s value=%s%n",
141
                entry.getKey().name(),
142
                entry.getKey().tags(),
143
                entry.getValue().metricValue()
144
            ));
145
    }
146
}