【Kafka从成神到升仙系列 五】面试官问我 Kafka 生产者的网络架构，我直接开始从源码背起

网上学习资料一大堆，但如果学到的知识不成体系，遇到问题时只是浅尝辄止，不再深入研究，那么很难做到真正的技术提升。

需要这份系统化资料的朋友，可以戳这里获取

一个人可以走的很快，但一群人才能走的更远！不论你是正从事IT行业的老鸟或是对IT行业感兴趣的新人，都欢迎加入我们的的圈子（技术交流、学习资源、职场吐槽、大厂内推、面试辅导），让我们一起学习成长！

}

void run(long now) {
// 业务代码
}

从代码中不难看出，当我们启动 `Sender` 线程之后，`Sender` 线程会不断的轮询调用 `run(long now)` 该方法，执行其业务。


那 `run(long now)` 方法到底做了些什么呢，我们一起来看一下


##### 2.1 accumulator.ready


* 遍历所有的 `TopicPartition`，获取每一个 `TopicPartition` 的 `Leader` 节点
* 弹出每一个 `TopicPartition` 的第一个 `batch`，校验该 `batch` 有没有符合发送的规定
* 如果该 `batch` 符合了发送的规定后，将节点放至 `readyNodes` 中，标识该节点已经可以发送数据了



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

public ReadyCheckResult ready(Cluster cluster, long nowMs) {
// 准备好的节点
Set readyNodes = new HashSet<>();
// 遍历所有的 TopicPartition
for (Map.Entry<TopicPartition, Deque> entry : this.batches.entrySet()) {
TopicPartition part = entry.getKey();
Deque deque = entry.getValue();
// 获取当前Partition的leader节点
Node leader = cluster.leaderFor(part);
if (leader == null) {
unknownLeadersExist = true;
} else if (!readyNodes.contains(leader) && !muted.contains(part)) {
synchronized (deque) {
// 弹出每一个 TopicPartition 的第一个batch
RecordBatch batch = deque.peekFirst();

            if (batch != null) {
                // bactch 满足 batch.size() 或者 时间达到 linger.ms、
                boolean full = deque.size() > 1 || batch.records.isFull();
                boolean expired = waitedTimeMs >= timeToWaitMs;
                boolean sendable = full || expired || exhausted || closed || flushInProgress();
                if (sendable && !backingOff) {
                    // 将当前的节点添加至准备好的队列中
                    readyNodes.add(leader);
                } else { 
                    nextReadyCheckDelayMs = Math.min(timeLeftMs, nextReadyCheckDelayMs);
                }
            }
        }
    }
}
// 最终返回该节点（这里最重要的还是 Set<String> 也就是准备好的节点集合）
return new ReadyCheckResult(readyNodes, nextReadyCheckDelayMs, unknownLeadersExist);
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

}

public ReadyCheckResult(Set readyNodes, long nextReadyCheckDelayMs, boolean unknownLeadersExist) {
this.readyNodes = readyNodes;
this.nextReadyCheckDelayMs = nextReadyCheckDelayMs;
this.unknownLeadersExist = unknownLeadersExist;
}

##### 2.2 metadata.requestUpdate


* 如果发现有 `TopicPartition` 没有 leader，那么这里就调用 `requestUpdate()` 方法更新 metadata



1
2
3
4
5
6
7
8

// 如果这个地方是 True，说明我们上面有的 TopicPartition 的 leader 节点为 null
if (result.unknownLeadersExist){
// 更新元数据
this.metadata.requestUpdate();
}

// 设置标记位为true,后续进行更新
public synchronized int requestUpdate() {
this.needUpdate = true;
return this.version;
}

##### 2.3 remove any nodes


* 遍历所有准备好的节点，利用 `NetworkClient` 来判断改节点是不是已经准备完毕
* 如果该节点未准备完毕，则从 `readyNodes` 中剔除
* 节点未准备完毕，会初始化链接该节点，便于下一次的消息发送


**PS：这里可能会有同学对上面已经准备好了，下面为什么还有准备好的逻辑筛选有疑问**


* 第一步筛选的是 `TopicPartition` 对应的 `batch` 已经满足了发送的必要
* 第二步筛选的是 `TopicPartition` 对应的 `Broker` 是否建立了链接，如果不是则**初始化链接**



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

// 遍历所有准备好的节点
Iterator iter = result.readyNodes.iterator();
long notReadyTimeout = Long.MAX_VALUE;
while (iter.hasNext()) {
Node node = iter.next();
// 利用 NetworkClient 来判断改节点是不是已经准备完毕
// 如果还未准备好，从准备好的队列中剔除掉
if (!this.client.ready(node, now)) {
iter.remove();
notReadyTimeout = Math.min(notReadyTimeout, this.client.connectionDelay(node, now));
}
}

// 判断节是否准备好发送
// 如果没有准备好发送，则会与该节点初始化链接，便于下一次的消息发送
public boolean ready(Node node, long now) {
// 已经准备好
if (isReady(node, now)){
return true;
}
// 与该节点的初始化
if (connectionStates.canConnect(node.idString(), now)){
initiateConnect(node, now);
}
return false;
}

##### 2.4 accumulator.drain


* 遍历所有准备好的 `readyNodes`，得到该 `Broker` 上所有的 `PartitionInfo` 信息，判断该 `Partition` 是否被处理中，如果没有在处理中则获取其对应的 `Deque<RecordBatch>`
* 弹出队列中的 `First`，判断其是否在 `backoff (没有重试过，或者重试了但是间隔已经达到了retryBackoffMs)` 且 `加上该 batch 的大小 < maxRequestSize`，该 `batch` 符合规定
* 将该 `batch`放进 `readyRecordBatchList`中，最终放进 `Map<node.id(), readyRecordBatchList>` ，这样我们一个 `Broker` 可以发送的 `batch` 就已经整理完毕。
* 最终我们得到 `Map<Integer, List<RecordBatch>>`，`key` 代表当前已经连接好的 `Broker`，`value` 代表当前需要发送的 `batch`



1
2
3
4
5
6
7
8
9
10
11

// 生成节点对应的batch消息
Map<Integer, List> batches = this.accumulator.drain(cluster,result.readyNodes,this.maxRequestSize, now);

public Map<Integer, List> drain(Cluster cluster, Set nodes,int maxSize,long now) {
Map<Integer, List> batches = new HashMap<>();
// 遍历所有准备好的node节点
for (Node node : nodes) {
int size = 0;
// 通过node节点获取其所有的Partition
List parts = cluster.partitionsForNode(node.id());
// 存储该节点需要发送的Batch
List ready = new ArrayList<>();
int start = drainIndex = drainIndex % parts.size();
do {
// 取Partition
PartitionInfo part = parts.get(drainIndex);
TopicPartition tp = new TopicPartition(part.topic(), part.partition());
// 当分区没有正在进行的批处理时
if (!muted.contains(tp)) {
// 获取该分区的所有的RecordBatch
Deque deque = getDeque(new TopicPartition(part.topic(), part.partition()));
if (deque != null) {
synchronized (deque) {
// 查看队列第一个
RecordBatch first = deque.peekFirst();
if (first != null) {
// 判断其重试与时间
boolean backoff = first.attempts > 0 && first.lastAttemptMs + retryBackoffMs > now;
if (!backoff) {
// 判断是否超越最大发送限制
if (size + first.records.sizeInBytes() > maxSize && !ready.isEmpty()) {
break;
} else {
// 取出队列第一个
RecordBatch batch = deque.pollFirst();
batch.records.close();
// 当前发送的大小累积
size += batch.records.sizeInBytes();
// 放入准备好的列表中
ready.add(batch);
batch.drainedMs = now;
}
}
}
}
}
}
this.drainIndex = (this.drainIndex + 1) % parts.size();
} while (start != drainIndex);
// 将节点与准备好的batch列表对应
batches.put(node.id(), ready);
}
// 最终返回：所有准备好的节点与对应的batch列表
return batches;
}

##### 2.5 createProduceRequests


* 遍历刚刚我们得到的 `Map<node.id(), readyRecordBatchList`，组装成客户端请求



1
2
3
4
5
6
7
8

List requests = createProduceRequests(batches, now);

// 组装客户端请求
private List createProduceRequests(Map<Integer, List> collated, long now) {
List requests = new ArrayList(collated.size());
for (Map.Entry<Integer, List> entry : collated.entrySet())
requests.add(produceRequest(now, entry.getKey(), acks, requestTimeout, entry.getValue()));
return requests;
}

##### 2.6 client.send


* 遍历每一个客户端请求并进行发送


PS：这里的发送是通过 `KafkaClient` 提供的接口，具体由 `NetworkClient` 实现，我们后面会讲



1
2
3
4
5
6
7
8
9
10
11

for (ClientRequest request : requests){
client.send(request, now);
}

##### 2.7 client.poll


* 发送消息


PS：这里也同样是通过 `KafkaClient` 提供的接口，具体由 `NetworkClient` 实现，我们后面会讲



1
2
3
4
5
6
7
8
9
10
11

this.client.poll(pollTimeout, now);

#### 3. NetworkClient


我们的 `Sender` 将 `Producer` 发送的消息进行 **校验、筛选、组装**，让我们的 `NetworkClient` 进一步的将消息发送


##### 3.1 send


* 拿到当前客户端请求的 `node`，校验其是否有权限
* 如果有权限的话，我们设置下时间并添加到到 `inFlightRequests`，调用 `selector` 进行发送（这里提前剧透一下，`send` 方法虽然叫发送，实际上并没有发送，只是注册了写事件，后面会讲到）


**inFlightRequests 的作用：**


* 缓存已经发出去但还没有收到响应的请求，保存对象的具体形式为 `Map<NodeId，Deque<Request>>`
* 配置参数 `max.in.flight.requests.per.connection`，默认值为5，即每个连接最多只能缓存5个未收到响应的请求，超过这个数值之后便不能再往这个连接发送更多的请求了



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

public void send(ClientRequest request, long now) {
// 拿到当前客户端请求的node
String nodeId = request.request().destination();
// 是否可以发送请求（我们前面已经校验过，一般情况下都能够发送）
if (!canSendRequest(nodeId))
throw new IllegalStateException(“Attempt to send a request to node " + nodeId + " which is not ready.”);
doSend(request, now);
}

private void doSend(ClientRequest request, long now) {
// 设置时间
request.setSendTimeMs(now);
// 将当前请求添加到 inFlightRequests
this.inFlightRequests.add(request);
selector.send(request.request());
}

##### 3.2 poll


* 判断当前需要更新元数据，如果需要则更新元数据
* 调用 `selector` 的 `poll` 方法进行 `Socket IO` 的操作（这里也在后面会讲到）
* 处理完成之后的操作
	+ 处理已经完成的 send
	+ 处理从 server 端接收到 Receive
	+ 处理连接失败那些连接
	+ 处理新建立的那些连接
	+ 处理超时的连接
* 如果回调的话，处理回调的信息



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

public List poll(long timeout, long now) {
// 判断当前需要更新元数据，如果需要则更新元数据
long metadataTimeout = metadataUpdater.maybeUpdate(now);

// 调用 selector 的 poll 方法进行 Socket IO 的操作
this.selector.poll(Utils.min(timeout, metadataTimeout, requestTimeoutMs));


// 处理完成之后的操作
long updatedNow = this.time.milliseconds();
List<ClientResponse> responses = new ArrayList<>();
// 处理已经完成的 send（不需要 response 的 request,如 send）
 handleCompletedSends(responses, updatedNow);
// 处理从 server 端接收到 Receive（如 Metadata 请求）
handleCompletedReceives(responses, updatedNow);
// 处理连接失败那些连接,重新请求 meta
handleDisconnections(responses, updatedNow);
// 处理新建立的那些连接（还不能发送请求,比如:还未认证）
handleConnections();
// 处理超时的连接
handleTimedOutRequests(responses, updatedNow);

// 处理回调的信息
for (ClientResponse response : responses) {
    if (response.request().hasCallback()) {
        try {
            response.request().callback().onComplete(response);
        } catch (Exception e) {
            log.error("Uncaught error in request completion:", e);
        }
    }
}

// 返回响应结果
return responses;
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

}

#### 4. Selector


终于来到了我们的最后一步，`Kafka` 自己封装的 `Selector`，这个哥们就是真正发送消息的地方


激动的心，颤抖的手，跟着我一起看看 `Selector` 到底是怎么发送消息的


##### 4.1send


* 根据当前节点的编号拿到当前客户端的 `channel`
* 向当前的 `KafkaChannel` 注册写事件


**写事件触发的时间：当 Scoket缓冲区 有空闲时，触发该事件**


从这里可以看出来，我们的 `send` 方法其实也没有真正的发送消息，只是向 `KafkaChannel` 注册了 `写事件`，保障后面 `poll` 轮旋事件发送的正确性。



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

public void send(Send send) {
// 根据当前节点的编号拿到当前客户端的channel
KafkaChannel channel = channelOrFail(send.destination());
try {
// 向当前的 KafkaChannel 注册写事件
channel.setSend(send);
} catch (CancelledKeyException e) {
this.failedSends.add(send.destination());
close(channel);
}
}

public void setSend(Send send) {
this.send = send;
this.transportLayer.addInterestOps(SelectionKey.OP_WRITE);
}

##### 4.2 poll


* 清除相关记录
* 获取就绪事件
* 处理 io 操作
* 将处理得到的 `stagedReceives` 添加到 `completedReceives` 中（`NetworkClient`处理响应）
* 关闭老的连接


由于这个方法比较重要，所以我们一个一个的讲，跟着我们的思路来



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

public void poll(long timeout) throws IOException {

// 清除相关缓存记录
clear();
1
2

// 获取就绪事件
long startSelect = time.nanoseconds();
int readyKeys = select(timeout);
long endSelect = time.nanoseconds();
currentTimeNanos = endSelect;
this.sensors.selectTime.record(endSelect - startSelect, time.milliseconds());

// 处理 io 操作
if (readyKeys > 0 || !immediatelyConnectedKeys.isEmpty()) {
    pollSelectionKeys(this.nioSelector.selectedKeys(), false);
    pollSelectionKeys(immediatelyConnectedKeys, true);
}

// 将处理得到的 stagedReceives 添加到 completedReceives 中
addToCompletedReceives();

long endIo = time.nanoseconds();
this.sensors.ioTime.record(endIo - endSelect, time.milliseconds());

// 关闭老的连接
maybeCloseOldestConnection();
1
2
3
4
5
6
7
8
9
10
11
12
13
14

}

###### 4.2.1 clear


`clear()` 方法是在每次 `poll()` 执行的第一步，它作用的就是**清理上一次 poll 过程产生的部分缓存。**


这里的缓存是不是感觉有点熟悉，他就是我们之前在 `NetworkClient` 的 \*\*处理完成之后的操作 \*\*对应的缓存，忘了的小伙伴可以回去看一下



1
2
3
4
5
6
7
8
9
10
11

private void clear() {
this.completedSends.clear();
this.completedReceives.clear();
this.connected.clear();
this.disconnected.clear();
this.disconnected.addAll(this.failedSends);
this.failedSends.clear();
}

###### 4.2.2 select


`select(ms)` 方法主要通过调用 `nioSelector` 的 `select` 方法，返回我们**就绪事件的数量**


这里的 `nioSelector` 是属于 `java.nio.channels.Selector` 的，也就是我们 `Java NIO` 包里面的


* `nioSelector.selectNow`：**非阻塞的**，当前操作没有通道准备好立即返回，返回是0
* `nioSelector.select`：**阻塞的**，当前没有通道准备好会阻塞住，最长时间为 `long ms`



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

private int select(long ms) throws IOException {
if (ms == 0L) {
return this.nioSelector.selectNow();
} else {
return this.nioSelector.select(ms);
}
}

###### 4.2.3 pollSelectionKeys



1
2
3
4
5

pollSelectionKeys(this.nioSelector.selectedKeys(), false);
pollSelectionKeys(immediatelyConnectedKeys, true);

这部分是 `socket IO` 的主要部分，发送 `Send` 及接收 `Receive` 都是在这里完成的，在 `poll()` 方法中，这个方法会调用两次：


* 第一次调用的目的是：处理已经就绪的事件，进行相应的 `IO` 操作；
* 第二次调用的目的是：处理新建立的那些连接，添加缓存及传输层（`Kafka` 又封装了一次，这里后续文章会讲述）的握手与认证。


我们来剖析下 `pollSelectionKeys` 整理的步骤：



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

private void pollSelectionKeys(Iterable selectionKeys, boolean isImmediatelyConnected) {
// 拿到当前所有准备好的keys
Iterator iterator = selectionKeys.iterator();
while (iterator.hasNext()) {
// 获取key并删除它，防止重复使用
SelectionKey key = iterator.next();
iterator.remove();

    // 根据 key 拿到对应的附件 KafkaChannel
    KafkaChannel channel = channel(key);
    sensors.maybeRegisterConnectionMetrics(channel.id());
    lruConnections.put(channel.id(), currentTimeNanos);

    try {
		   // 处理所有已经完成握手(Tcp)的连接(正常或立即)
        if (isImmediatelyConnected || key.isConnectable()) {
            if (channel.finishConnect()) {
                this.connected.add(channel.id());
                this.sensors.connectionCreated.record();
            } else
                continue;
        }

        // 如果通道未准备好，请完成准备
        if (channel.isConnected() && !channel.ready())
            channel.prepare();

        // 如果通道已准备好从任何具有可读数据的连接中读取
        if (channel.ready() && key.isReadable() && !hasStagedReceive(channel)) {
            NetworkReceive networkReceive;
            while ((networkReceive = channel.read()) != null)
                addToStagedReceives(channel, networkReceive);
        }

		  // 如果通道准备好了，就向缓冲区中有空间且我们有数据的任何套接字写入
        if (channel.ready() && key.isWritable()) {
            Send send = channel.write();
            if (send != null) {
                this.completedSends.add(send);
                this.sensors.recordBytesSent(channel.id(), send.size());
            }
        }

        // 取消所有失效的套接字
        if (!key.isValid()) {
            close(channel);
            this.disconnected.add(channel.id());
        }

    } 
}
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

}

* 拿到所有准备好的 `keys`，获取 `keys` 并删除它（防止重复使用），根据 `key` 拿到对应的附件 `KafkaChannel`
* 处理以下几种情况：
	+ 所有已经完成握手的连接
	+ 通道未准备好的 `key`
	+ 通道准备好的数据
	+ 可写入的 `key`
	+ 取消所有失效的套接字


其中我们不难看出，最重要的当属 **处理可写入的 `key`**，我们有必要来详细说说 `Send send = channel.write();` 的实现



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

public Send write() throws IOException {
Send result = null;
if (send != null && send(send)) {
result = send;
send = null;
}
return result;
}

// 是否发送成功
private boolean send(Send send) throws IOException {
// 写入消息
send.writeTo(transportLayer);
// 写完之后取消写事件，防止无限触发写事件
if (send.completed())
transportLayer.removeInterestOps(SelectionKey.OP_WRITE);

网上学习资料一大堆，但如果学到的知识不成体系，遇到问题时只是浅尝辄止，不再深入研究，那么很难做到真正的技术提升。

需要这份系统化资料的朋友，可以戳这里获取

一个人可以走的很快，但一群人才能走的更远！不论你是正从事IT行业的老鸟或是对IT行业感兴趣的新人，都欢迎加入我们的的圈子（技术交流、学习资源、职场吐槽、大厂内推、面试辅导），让我们一起学习成长！

}
return result;
}

// 是否发送成功
private boolean send(Send send) throws IOException {
// 写入消息
send.writeTo(transportLayer);
// 写完之后取消写事件，防止无限触发写事件
if (send.completed())
transportLayer.removeInterestOps(SelectionKey.OP_WRITE);

[外链图片转存中…(img-8iNjNvJ4-1715353420104)]
[外链图片转存中…(img-2upyGovV-1715353420105)]

网上学习资料一大堆，但如果学到的知识不成体系，遇到问题时只是浅尝辄止，不再深入研究，那么很难做到真正的技术提升。

需要这份系统化资料的朋友，可以戳这里获取

一个人可以走的很快，但一群人才能走的更远！不论你是正从事IT行业的老鸟或是对IT行业感兴趣的新人，都欢迎加入我们的的圈子（技术交流、学习资源、职场吐槽、大厂内推、面试辅导），让我们一起学习成长！