4、深潜KafkaProducer —— RecordAccumulator精析

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写入的时候，都会递增lastOffset
  return 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才能有header
    if (magic < RecordBatch.MAGIC_VALUE_V2 && headers != null && headers.length > 0)
        throw new IllegalArgumentException("...");

    if (this.firstTimestamp == null) // 更新firstTimestamp
        this.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状态
    // 计算offsetDelta
    int offsetDelta = (int) (offset - this.baseOffset);
    // 计算timestampDelta
    long 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的字节数不能超过writeLimit
    return 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()方法写入Record
        Long checksum = this.recordsBuilder.append(timestamp, key, value, headers);
        // 更新maxRecordSize和lastAppendTime
        this.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) // 先忽略nextRecordMetadata
        return 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列表中的ByteBuffer
        if (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中出现了空闲的ByteBuffer
                    if (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要是还存在空闲空间，则唤醒下一个等待线程来尝试分配ByteBuffer
        try {
            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);
    else
        return 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并增加nonPooledAvailableMemory
            this.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>加锁
            // 如果追加操作成功，则返回RecordAppendResult
            RecordAppendResult appendResult = tryAppend(timestamp, key, value, headers, callback, dq, nowMs);
            if (appendResult != null) // 追加成功，则返回的appendResult不为null
                return 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中分配ByteBuffer
        buffer = free.allocate(size, maxTimeToBlock);

        nowMs = time.milliseconds();
        synchronized (dq) { // 再次对ArrayDeque<ProducerBatch>加锁
            if (closed)
                throw new KafkaException("Producer closed while send in progress");
            // 再次尝试tryAppend()方法追加Record
            RecordAppendResult appendResult = tryAppend(timestamp, key, value, headers, callback, dq, nowMs);
            if (appendResult != null) {
                return appendResult;
            }
            // 将ByteBuffer封装成MemoryRecordsBuilder
            MemoryRecordsBuilder 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; // 置空buffer
            return new RecordAppendResult(future, dq.size() > 1 || batch.isFull(), true, false);
        }
    } finally {
        if (buffer != null) // 如果buffer不为空，则表示写入过程中出现异常，这里会释放ByteBuffer
            free.deallocate(buffer);
        // 当前Record已经写入完成，递减appendsInProgress
        appendsInProgress.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的topic
    Set<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;
                    // 检查上述五个条件，找到此次发送涉及到的Node
                    boolean 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中第一个ProducerBatch
                ProducerBatch 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;
}