Kafka 生产者的网络架构

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

2.6 client.send

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

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

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

2.7 client.poll

• 发送消息

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

this.client.poll(pollTimeout, now);

3. NetworkClient

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

3.1 send

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

inFlightRequests 的作用：

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

public void send(ClientRequest request, long now) {// 拿到当前客户端请求的nodeString 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);// 将当前请求添加到 inFlightRequeststhis.inFlightRequests.add(request);selector.send(request.request());
}

3.2 poll

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

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 responses = new ArrayList<>();// 处理已经完成的 send（不需要 response 的 request,如 send）handleCompletedSends(responses, updatedNow);// 处理从 server 端接收到 Receive（如 Metadata 请求）handleCompletedReceives(responses, updatedNow);// 处理连接失败那些连接,重新请求 metahandleDisconnections(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;
}

4. Selector

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

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

4.1send

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

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

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

public void send(Send send) {// 根据当前节点的编号拿到当前客户端的channelKafkaChannel 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处理响应）
• 关闭老的连接

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

public void poll(long timeout) throws IOException {// 清除相关缓存记录clear();// 获取就绪事件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();
}

4.2.1 clear

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

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

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

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

4.2.3 pollSelectionKeys

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

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

• 第一次调用的目的是：处理已经就绪的事件，进行相应的 IO 操作；

• 第二次调用的目的是：处理新建立的那些连接，添加缓存及传输层（Kafka 又封装了一次，这里后续文章会讲述）的握手与认证。

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

private void pollSelectionKeys(Iterable selectionKeys, boolean isImmediatelyConnected) {// 拿到当前所有准备好的keysIterator iterator = selectionKeys.iterator();while (iterator.hasNext()) {// 获取key并删除它，防止重复使用SelectionKey key = iterator.next();iterator.remove();// 根据 key 拿到对应的附件 KafkaChannelKafkaChannel 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();} elsecontinue;}// 如果通道未准备好，请完成准备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());}} }
}

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

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

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);return send.completed();
}// 通过客户端的channel向服务端发送信息
public long writeTo(GatheringByteChannel channel) throws IOException {long written = channel.write(buffers);remaining -= written;if (channel instanceof TransportLayer)pending = ((TransportLayer) channel).hasPendingWrites();return written;
}

最终还是调用了我们 Java NIO 中的 channel.write(buffers) 方法完成发送消息。

4.2.4 addToCompletedReceives

client 的时序性而是通过 InFlightRequests 和 RecordAccumulator 的 mutePartition 来保证的。因此对于 Client 端而言，这里接收到的所有 Receive 都会被放入到 completedReceives 的集合中等待后续处理。

这里面的数据主要我们上面 ``pollSelectionKeys中添加的，然后在这放入到completedReceives，随后被我们NetworkClient` 中被处理

// 处理响应放入到 completedReceives 中
private void addToCompletedReceives() {if (!this.stagedReceives.isEmpty()) {Iterator>> iter = this.stagedReceives.entrySet().iterator();while (iter.hasNext()) {Map.Entry> entry = iter.next();KafkaChannel channel = entry.getKey();if (!channel.isMute()) {Deque deque = entry.getValue();NetworkReceive networkReceive = deque.poll();this.completedReceives.add(networkReceive);this.sensors.recordBytesReceived(channel.id(), networkReceive.payload().limit());if (deque.isEmpty())iter.remove();}}}
}

五、总结

终于写完了，其实最开始学 kafka 的时候是今年 2 月份，那时候还不懂什么是 IO，看源码的通信基本看不懂

后来，花了几个月的时间学了 操作系统 --> 计算机网络 --> Linux 通信 --> Java NIO --> Netty，现在看 Kafka 的通信就变得通透了。

另外，基本现在所有源码的通信都有 Netty 架构的影子

所以，如果你也想学源码的话，最好是先看看 Netty 的相关知识，学完之后，你会发现，通信架构不过如此。

如果你能看到这，想必已经跟完了整个 Producer 的网络架构部分，你有没有感觉到一个事情：网络架构就是整个生产者运行的全部流程

对的，本来我以为讲网络架构就是网络架构，越写越发觉，这不就是整个生产者发送的全部流程嘛

所以，我们生产者全部的文章就结束了，总体如下：

• 生产者如何将消息放入到内存缓冲区
• 你真的了解 Kafka 的元数据嘛
• 你真的了解 Kafka 的缓存池机制嘛

喜欢的可以点个关注吆，后续会继续更新 kafka 源码系列文章，下一部分应该就是 服务端 了

我是爱敲代码的小黄，独角兽企业的Java开发工程师，CSDN博客专家，Java领域新星创作者，喜欢后端架构和中间件源码。

我们下期再见。