1
2
3
4
5
6
7
8
9
10
11

public MemoryRecordsBuilder(ByteBuffer buffer,...) { // 省略其他参数// 将MemoryRecordsBuilder关联的ByteBuffer封装成ByteBufferOutputStream流this(new ByteBufferOutputStream(buffer), ...);
}public MemoryRecordsBuilder(ByteBufferOutputStream bufferStream,...) { // 省略其他参数// 省略其他字段的初始化this.bufferStream = bufferStream;// 在bufferStream流外层套一层压缩流，然后再套一层DataOutputStream流this.appendStream = new DataOutputStream(compressionType.wrapForOutput(this.bufferStream, magic));
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

public Long append(long timestamp, ByteBuffer key, ByteBuffer value, Header[] headers) {return appendWithOffset(nextSequentialOffset(), timestamp, key, value, headers);
}private long nextSequentialOffset() {// 这里的baseOffset是RecordBatch中的baseOffset，lastOffset用来记录当前写入Record的offset值，后面会用lastOffset-baseOffset计算offset delta。每次有新Record写入的时候，都会递增lastOffsetreturn lastOffset == null ? baseOffset : lastOffset + 1;
}private Long appendWithOffset(long offset, boolean isControlRecord, long timestamp, ByteBuffer key, ByteBuffer value, Header[] headers) {if (isControlRecord != this.isControlBatch) // 检查isControl标志是否一致throw new IllegalArgumentException("...");if (lastOffset != null && offset <= this.lastOffset) // 保证offset递增throw new IllegalArgumentException("...");if (timestamp < 0 && timestamp != RecordBatch.NO_TIMESTAMP) // 检查时间戳throw new IllegalArgumentException("...");// 检查：只有V2版本message才能有headerif (magic < RecordBatch.MAGIC_VALUE_V2 && headers != null && headers.length > 0)throw new IllegalArgumentException("...");if (this.firstTimestamp == null) // 更新firstTimestampthis.firstTimestamp = timestamp;if (magic > RecordBatch.MAGIC_VALUE_V1) { // 针对V2的写入appendDefaultRecord(offset, timestamp, key, value, headers);return null;} else { // 针对V0、V1的写入return appendLegacyRecord(offset, timestamp, key, value, magic);}
}

1
2
3
4
5
6
7
8
9
10
11
12

private void appendDefaultRecord(long offset, long timestamp, ByteBuffer key, ByteBuffer value,Header[] headers) throws IOException {ensureOpenForRecordAppend(); // 检查appendStream状态// 计算offsetDeltaint offsetDelta = (int) (offset - this.baseOffset);// 计算timestampDeltalong timestampDelta = timestamp - this.firstTimestamp; // 这里使用的DefaultRecord是一个工具类，其writeTo()方法会按照V2 Record的格式向appendStream流中int sizeInBytes = DefaultRecord.writeTo(appendStream, offsetDelta, timestampDelta, key, value, headers);// 修改RecordBatch中的元信息，例如，Record数量（numRecords）recordWritten(offset, timestamp, sizeInBytes);
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

public boolean hasRoomFor(long timestamp, ByteBuffer key, ByteBuffer value, Header[] headers) {// 检查两个状态，一个是appendStream流状态，另一个是当前已经写入的预估字节数是否以及超过了writeLimit字段指定的写入上限，这里的writeLimit字段用来记录MemoryRecordsBuilder能够写入的字节数上限值if (isFull()) return false;// 每个RecordBatch至少可以写入一个Record，此时要是一个Record都没有，则可以继续写入if (numRecords == 0)return true;final int recordSize;if (magic < RecordBatch.MAGIC_VALUE_V2) { // V0、V1版本的判断recordSize = Records.LOG_OVERHEAD + LegacyRecord.recordSize(magic, key, value);} else { // 估算此次写入的Record大小int nextOffsetDelta = lastOffset == null ? 0 : (int) (lastOffset - baseOffset + 1);long timestampDelta = firstTimestamp == null ? 0 : timestamp - firstTimestamp;recordSize = DefaultRecord.sizeInBytes(nextOffsetDelta, timestampDelta, key, value, headers);}// 已写入字节数+此次写入Record的字节数不能超过writeLimitreturn this.writeLimit >= estimatedBytesWritten() + recordSize;
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

public FutureRecordMetadata tryAppend(long timestamp, byte[] key, byte[] value, Header[] headers, Callback callback, long now) {// 检查MemoryRecordsBuilder是否还有空间继续写入if (!recordsBuilder.hasRoomFor(timestamp, key, value, headers)) {return null; // 没有空间写入的话，则返回null} else {// 调用append()方法写入RecordLong checksum = this.recordsBuilder.append(timestamp, key, value, headers);// 更新maxRecordSize和lastAppendTimethis.maxRecordSize = Math.max(this.maxRecordSize, AbstractRecords.estimateSizeInBytesUpperBound(magic(),recordsBuilder.compressionType(), key, value, headers));this.lastAppendTime = now;// 创建FutureRecordMetadata对象FutureRecordMetadata future = new FutureRecordMetadata(this.produceFuture,this.recordCount,timestamp, checksum, key == null ? -1 : key.length, value == null ? -1 : value.length, Time.SYSTEM);// 将Callback和FutureRecordMetadata记录到thunks中thunks.add(new Thunk(callback, future));this.recordCount++; // 更新recordCount字段return future; // 返回FutureRecordMetadata}
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14

public RecordMetadata get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {// 计算超时时间long now = time.milliseconds();long timeoutMillis = unit.toMillis(timeout);long deadline = Long.MAX_VALUE - timeoutMillis < now ? Long.MAX_VALUE : now + timeoutMillis;// 依赖ProduceRequestResult中的CountDown来实现阻塞等待boolean occurred = this.result.await(timeout, unit);if (!occurred)throw new TimeoutException("Timeout after waiting for " + timeoutMillis + " ms.");if (nextRecordMetadata != null) // 可以先忽略nextRecordMetadata，后面介绍split的时候，再深入介绍return nextRecordMetadata.get(deadline - time.milliseconds(), TimeUnit.MILLISECONDS);// 调用value()方法返回RecordMetadata对象return valueOrError();
}

1
2
3
4
5
6
7
8

RecordMetadata value() {if (nextRecordMetadata != null) // 先忽略nextRecordMetadatareturn nextRecordMetadata.value();// 将partition信息、baseOffset、relativeOffset、时间戳（LogAppendTime或CreateTime）等信息封装成RecordMetadata对象返回return new RecordMetadata(result.topicPartition(), this.result.baseOffset(), this.relativeOffset, timestamp(), this.checksum, this.serializedKeySize, this.serializedValueSize);
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

public boolean done(long baseOffset, long logAppendTime, RuntimeException exception) {// 根据exception字段决定此次ProducerBatch发送的最终状态final FinalState tryFinalState = (exception == null) ? FinalState.SUCCEEDED : FinalState.FAILED;// CAS操作更新finalState状态，只有第一次更新的时候，才会触发completeFutureAndFireCallbacks()方法if (this.finalState.compareAndSet(null, tryFinalState)) {completeFutureAndFireCallbacks(baseOffset, logAppendTime, exception);return true;}// done()方法可能被调用一次或是两次，如果是SUCCEED状态切换成其他的失败状态，直接会抛出异常if (this.finalState.get() != FinalState.SUCCEEDED) {// 省略日志输出代码} else {throw new IllegalStateException("...");}return false;
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

private void completeFutureAndFireCallbacks(long baseOffset, long logAppendTime, RuntimeException exception) {// 更新ProduceRequestResult中的baseOffset、logAppendTime、error字段produceFuture.set(baseOffset, logAppendTime, exception);// 遍历thunks集合，触发每个Record的Callback回调for (Thunk thunk : thunks) { // 省略try/catch代码块if (exception == null) {RecordMetadata metadata = thunk.future.value();if (thunk.callback != null)thunk.callback.onCompletion(metadata, null);} else {if (thunk.callback != null)thunk.callback.onCompletion(null, exception);}}// 调用底层CountDownLatch.countDown()方法，释放阻塞在其上的线程produceFuture.done();
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87

public ByteBuffer allocate(int size, long maxTimeToBlockMs) throws InterruptedException {if (size > this.totalMemory) // 首先检查目标ByteBuffer的大小是否大于了throw new IllegalArgumentException("...");ByteBuffer buffer = null;this.lock.lock(); // 加锁// 检查当前BufferPool的状态，如果当前BufferPool处于关闭状态，则直接抛出异常(略)try {// 目标大小恰好为poolableSize，且free列表为空的话，直接复用free列表中的ByteBufferif (size == poolableSize && !this.free.isEmpty())return this.free.pollFirst();// 计算free列表中的ByteBuffer总空间int freeListSize = freeSize() * this.poolableSize;// 当前BufferPool能够释放出目标大小的空间，则直接通过freeUp()方法进行分配if (this.nonPooledAvailableMemory + freeListSize >= size) {freeUp(size);this.nonPooledAvailableMemory -= size;} else {int accumulated = 0;// 如果当前BufferPool空间不足以提供目标空间，则需要阻塞当前线程Condition moreMemory = this.lock.newCondition();try {// 计算当前线程最大阻塞时长long remainingTimeToBlockNs = TimeUnit.MILLISECONDS.toNanos(maxTimeToBlockMs);this.waiters.addLast(moreMemory);while (accumulated < size) { // 循环等待分配成功long startWaitNs = time.nanoseconds();long timeNs;boolean waitingTimeElapsed;try {// 当前线程阻塞等待，返回结果为false则表示阻塞超时waitingTimeElapsed = !moreMemory.await(remainingTimeToBlockNs, TimeUnit.NANOSECONDS);} finally {long endWaitNs = time.nanoseconds();timeNs = Math.max(0L, endWaitNs - startWaitNs);}// 检查当前BufferPool的状态，如果当前BufferPool处于关闭状态，则直接抛出异常(略)if (waitingTimeElapsed) { // 指定时长内依旧没有获取到目标大小的空间，则抛出异常throw new BufferExhaustedException("...");}remainingTimeToBlockNs -= timeNs;// 目标大小是poolableSize大小的ByteBuffer，且free中出现了空闲的ByteBufferif (accumulated == 0 && size == this.poolableSize && !this.free.isEmpty()) {buffer = this.free.pollFirst();accumulated = size;} else {// 先分配一部分空间，并继续等待空闲空间freeUp(size - accumulated);int got = (int) Math.min(size - accumulated, this.nonPooledAvailableMemory);this.nonPooledAvailableMemory -= got;accumulated += got;}}accumulated = 0;} finally {// 如果上面的while循环不是正常结束的，则accumulated不为0，这里会归还this.nonPooledAvailableMemory += accumulated;this.waiters.remove(moreMemory);}}} finally {// 当前BufferPool要是还存在空闲空间，则唤醒下一个等待线程来尝试分配ByteBuffertry {if (!(this.nonPooledAvailableMemory == 0 && this.free.isEmpty()) && !this.waiters.isEmpty())this.waiters.peekFirst().signal();} finally {lock.unlock(); // 解锁}}if (buffer == null) // 分配成功但无法复用free列表中的ByteBuffer(可能目标大小不是poolableSize大小，或是free列表本身是空的)return safeAllocateByteBuffer(size);elsereturn buffer; // 直接复用free大小的ByteBuffer
}// 这里简单看一下freeUp()方法，这里会不断从free列表中释放空闲的ByteBuffer来补充nonPooledAvailableMemory
private void freeUp(int size) {while (!this.free.isEmpty() && this.nonPooledAvailableMemory < size)this.nonPooledAvailableMemory += this.free.pollLast().capacity();
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

public void deallocate(ByteBuffer buffer, int size) {lock.lock(); // 加锁try {// 如果待释放的ByteBuffer大小为poolableSize，则直接放入free列表中if (size == this.poolableSize && size == buffer.capacity()) {buffer.clear();this.free.add(buffer);} else {// 如果不是poolableSize大小，则由 JVM GC来回收ByteBuffer并增加nonPooledAvailableMemorythis.nonPooledAvailableMemory += size;}// 唤醒waiters中的第一个阻塞线程Condition moreMem = this.waiters.peekFirst();if (moreMem != null)moreMem.signal();} finally {lock.unlock(); // 释放锁}
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

private RecordAppendResult tryAppend(long timestamp, byte[] key, byte[] value, Header[] headers, Callback callback, Deque<ProducerBatch> deque, long nowMs) {// 获取ArrayDeque<ProducerBatch>集合中最后一个ProducerBatch对象ProducerBatch last = deque.peekLast();if (last != null) {// 尝试将消息写入last这个ProducerBatch对象FutureRecordMetadata future = last.tryAppend(timestamp, key, value, headers, callback, nowMs);if (future == null) // 写入失败，则关闭last指向的ProducerBatch对象，同时返回null表示写入失败last.closeForRecordAppends();else// 写入成功，则返回RecordAppendResult对象return new RecordAppendResult(future, deque.size() > 1 || last.isFull(), false, false);}return null;
}

1
2
3
4

if (result.batchIsFull || result.newBatchCreated) {// 当此次写入填满一个ProducerBatch或是有新ProducerBatch创建的时候，会唤醒Sender线程来进行发送this.sender.wakeup();
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61

public RecordAppendResult append(TopicPartition tp, long timestamp,byte[] key, byte[] value, Header[] headers, Callback callback,long maxTimeToBlock, boolean abortOnNewBatch, long nowMs) throws InterruptedException {// 统计正在向RecordAccumulator中写入的Record数量appendsInProgress.incrementAndGet();ByteBuffer buffer = null;if (headers == null) headers = Record.EMPTY_HEADERS;try {// 在 batches 集合中查找目标 partition 对应的 ArrayDeque<ProducerBatch> 集合，// 如果查找失败，则创建新的ArrayDeque<ProducerBatch>，并添加到 batches 集合中。Deque<ProducerBatch> dq = getOrCreateDeque(tp);synchronized (dq) { // 对ArrayDeque<ProducerBatch>加锁// 如果追加操作成功，则返回RecordAppendResultRecordAppendResult appendResult = tryAppend(timestamp, key, value, headers, callback, dq, nowMs);if (appendResult != null) // 追加成功，则返回的appendResult不为nullreturn appendResult;}// 如果追加Record失败，则可能是因为当前使用的ProducerBatch已经被填满了，这里会根据abortOnNewBatch参数，// 决定是否立即返回RecordAppendResult结果，返回的RecordAppendResult中如果abortForNewBatch为true，// 会再触发一次append()方法if (abortOnNewBatch) {return new RecordAppendResult(null, false, false, true);}byte maxUsableMagic = apiVersions.maxUsableProduceMagic();int size = Math.max(this.batchSize, AbstractRecords.estimateSizeInBytesUpperBound(maxUsableMagic, compression, key, value, headers));// 从BufferPool中分配ByteBufferbuffer = free.allocate(size, maxTimeToBlock);nowMs = time.milliseconds();synchronized (dq) { // 再次对ArrayDeque<ProducerBatch>加锁if (closed)throw new KafkaException("Producer closed while send in progress");// 再次尝试tryAppend()方法追加RecordRecordAppendResult appendResult = tryAppend(timestamp, key, value, headers, callback, dq, nowMs);if (appendResult != null) {return appendResult;}// 将ByteBuffer封装成MemoryRecordsBuilderMemoryRecordsBuilder recordsBuilder = recordsBuilder(buffer, maxUsableMagic);// 创建ProducerBatch对象ProducerBatch batch = new ProducerBatch(tp, recordsBuilder, nowMs);// 通过tryAppend()方法将Record追加到ProducerBatch中FutureRecordMetadata future = Objects.requireNonNull(batch.tryAppend(timestamp, key, value, headers,callback, nowMs));// 将ProducerBatch添加到ArrayDeque<ProducerBatch>中dq.addLast(batch);// 将ProducerBatch添加到IncompleteBatches中incomplete.add(batch);buffer = null; // 置空bufferreturn new RecordAppendResult(future, dq.size() > 1 || batch.isFull(), true, false);}} finally {if (buffer != null) // 如果buffer不为空，则表示写入过程中出现异常，这里会释放ByteBufferfree.deallocate(buffer);// 当前Record已经写入完成，递减appendsInProgressappendsInProgress.decrementAndGet();}
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43

public ReadyCheckResult ready(Cluster cluster, long nowMs) {// 用来记录可以向哪些Node节点发送数据Set<Node> readyNodes = new HashSet<>();  // 记录下次需要调用ready()方法的时间间隔long nextReadyCheckDelayMs = Long.MAX_VALUE;// 记录Cluster元数据中找不到leader replica的topicSet<String> unknownLeaderTopics = new HashSet<>();// 是否有线程在阻塞等待BufferPool释放空间boolean exhausted = this.free.queued() > 0;// 下面遍历batches集合，对其中每个partition的leader replica所在的Node都进行判断for (Map.Entry<TopicPartition, Deque<ProducerBatch>> entry : this.batches.entrySet()) {Deque<ProducerBatch> deque = entry.getValue();synchronized (deque) {ProducerBatch batch = deque.peekFirst();if (batch != null) {TopicPartition part = entry.getKey();// 查找目标partition的leader replica所在的节点Node leader = cluster.leaderFor(part);if (leader == null) {// leader replica找不到，会认为是异常情况，不能发送消息unknownLeaderTopics.add(part.topic());} else if (!readyNodes.contains(leader) && !isMuted(part)) {boolean full = deque.size() > 1 || batch.isFull();long waitedTimeMs = batch.waitedTimeMs(nowMs);boolean backingOff = batch.attempts() > 0 && waitedTimeMs < retryBackoffMs;long timeToWaitMs = backingOff ? retryBackoffMs : lingerMs;boolean expired = waitedTimeMs >= timeToWaitMs;// 检查上述五个条件，找到此次发送涉及到的Nodeboolean sendable = full || expired || exhausted || closed || flushInProgress();if (sendable && !backingOff) {readyNodes.add(leader); } else {long timeLeftMs = Math.max(timeToWaitMs - waitedTimeMs, 0);// 记录下次需要调用ready()方法检查的时间间隔nextReadyCheckDelayMs = Math.min(timeLeftMs, nextReadyCheckDelayMs);}}}}}return new ReadyCheckResult(readyNodes, nextReadyCheckDelayMs, unknownLeaderTopics);
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55

public Map<Integer, List<ProducerBatch>> drain(Cluster cluster, Set<Node> nodes, int maxSize, long now) {if (nodes.isEmpty())return Collections.emptyMap();// 转换后的结果，Key是目标Node的Id，Value是发送到目标Node的ProducerBatch集合Map<Integer, List<ProducerBatch>> batches = new HashMap<>();for (Node node : nodes) {// 获取目标Node的ProducerBatch集合List<ProducerBatch> ready = drainBatchesForOneNode(cluster, node, maxSize, now);batches.put(node.id(), ready);}return batches;
}private List<ProducerBatch> drainBatchesForOneNode(Cluster cluster, Node node, int maxSize, long now) {int size = 0;// 获取当前Node上的partition集合List<PartitionInfo> parts = cluster.partitionsForNode(node.id());// 记录发往目标Node的ProducerBatch集合List<ProducerBatch> ready = new ArrayList<>();// drainIndex是batches的下标，记录上次发送停止时的位置，下次继续从此位置开始发送。如果始终从     // 索引0的队列开始发送，可能会出现一直只发送前几个partition的消息的情况，造成其他partition饥饿。int start = drainIndex = drainIndex % parts.size();do {// 获取partition的元数据PartitionInfo part = parts.get(drainIndex);TopicPartition tp = new TopicPartition(part.topic(), part.partition());this.drainIndex = (this.drainIndex + 1) % parts.size();// 检查目标partition对应的ArrayDeque是否为空(略)synchronized (deque) {// 获取ArrayDeque中第一个ProducerBatch对象ProducerBatch first = deque.peekFirst();if (first == null)continue;// 重试操作的话，需要检查是否已经等待了足够的退避时间(略)if (size + first.estimatedSizeInBytes() > maxSize && !ready.isEmpty()) {// 此次请求要发送的数据量已满，结束循环break;} else {if (shouldStopDrainBatchesForPartition(first, tp))break;// 获取ArrayDeque中第一个ProducerBatchProducerBatch batch = deque.pollFirst();// 事务相关的处理(略)// 关闭底层输出流，将ProducerBatch设置成只读状态batch.close();size += batch.records().sizeInBytes();// 将ProducerBatch记录到ready集合中ready.add(batch);// 修改ProducerBatch的drainedMs标记batch.drained(now);}}} while (start != drainIndex);return ready;
}