从源码全面解析 Kafka 元数据的来龙去脉

桃扇骨 2023-10-12 22:21 27阅读 0赞

### 一、引言 ###

初学一个技术，怎么了解该技术的源码至关重要。

对我而言，最佳的阅读源码的方式，那就是：不求甚解，观其大略

你如果进到庐山里头，二话不说，蹲下头来，弯下腰，就对着某棵树某棵小草猛研究而不是说先把庐山的整体脉络研究清楚了，那么你的学习方法肯定效率巨低而且特别痛苦。

最重要的还是慢慢地打击你的积极性，说我的学习怎么那么不 happy 啊，怎么那么没劲那，因为你的学习方法错了，大体读明白，先拿来用，用着用着，很多道理你就明白了。

先从整体上把关源码，再去扣一些细节问题。

举个简单的例子：

如果你刚接触 HashMap，你刚有兴趣去看其源码，在看 HashMap 的时候，有一个知识：当链表长度达到 8 之后，就变为了红黑树，小于 6 就变成了链表，当然，还和当前的长度有关。

这个时候，如果你去深究红黑树、为什么是 8 不是别的，又去查 泊松分布，最终会慢慢的搞死自己。

所以，正确的做法，我们先把这一部分给略过去，知道这个概念即可，等后面我们把整个庐山看完之后，再回过头抠细节。

当然，本章我们讲述 **Kafka中的生产者怎么获取的元数据**

### 二、Kafka为何不用 Netty ###

上篇文章我们讲述到，我们的 producer 会将消息发送至 `RecordAccumulator` 中，然后启动 `Sender` 线程

我们大概率可以猜测到，我们的 `Sender` 线程是和我们的 `Broker` 进行通信的，提到通信，不得不说一下 `Netty`

大家都知道，`Netty` 是一个优秀的 `I/O` 框架，但 `Kafka` 在通信方面并没有采用 `Netty`，让人比较难以理解

当然，博主也查到了关于 `kafka` 开发者的回答：

> 一共两个原因：
> 
> 1、由于性能问题，kafka 的通信过程并不需要 netty 那么庞大的通信体系
> 
> 2、kafka客户端原始时期，需要让用户将整个东西作为依赖项包含其内，如果引入了 netty，那么每个人依赖的版本号不同，将会产生巨大的兼容问题
> 
> 3、kafka的安全层和一些另外的问题，需要 kafka 自己来解决，而这些烦恼的问题，netty 中已经解决了
> 
> 博主感觉，极大概率由于历史原因，现在就算换成 netty，一些代码也不容易重构，更何况现在 kafka 自研的 I/O 通信模型反响还可以，所以 kafka 一直都没使用 netty 的想法。

![format_png][]

### 三、元数据是什么 ###

如果我们的 `Sender` 线程想要连接 `Broker`，最大的一个环节就是获取 `Broker` 的元数据

而元数据的获取，是通过 `Sender` 线程来获取的

> Sender 线程是一个不断轮询的线程，类似我们之前提到的 EventLoop 线程

我们首先解释一下 **元数据（Metadata ）** 是个什么东西：

java

复制代码

`// 这个类被 client 线程和后台 sender 所共享,它只保存了所有 topic 的部分数据,当我们请求一个它上面没有的 topic meta 时,它会通过发送 metadata update 来更新 meta 信息, // 如果 topic meta 过期策略是允许的,那么任何 topic 过期的话都会被从集合中移除, // 但是 consumer 是不允许 topic 过期的因为它明确地知道它需要管理哪些 topic public final class Metadata { private static final Logger log = LoggerFactory.getLogger(Metadata.class); public static final long TOPIC_EXPIRY_MS = 5 * 60 * 1000; private static final long TOPIC_EXPIRY_NEEDS_UPDATE = -1L; private final long refreshBackoffMs; // metadata 更新失败时,为避免频繁更新 meta,最小的间隔时间,默认 100ms private final long metadataExpireMs; // metadata 的过期时间, 默认 60,000ms private int version; // 每更新成功1次，version自增1,主要是用于判断 metadata 是否更新 private long lastRefreshMs; // 最近一次更新时的时间（包含更新失败的情况） private long lastSuccessfulRefreshMs; // 最近一次成功更新的时间（如果每次都成功的话，与前面的值相等, 否则，lastSuccessulRefreshMs < lastRefreshMs) private Cluster cluster; // 集群中一些 topic 的信息 private boolean needUpdate; // 是都需要更新 metadata /* Topics with expiry time */ private final Map<String, Long> topics; // topic 与其过期时间的对应关系 private final List<Listener> listeners; // 事件监控者 private final ClusterResourceListeners clusterResourceListeners; //当接收到 metadata 更新时, ClusterResourceListeners的列表 private boolean needMetadataForAllTopics; // 是否强制更新所有的 metadata private final boolean topicExpiryEnabled; // 默认为 true, Producer 会定时移除过期的 topic,consumer 则不会移除 }`

如果你感觉参数有点多，难以看懂，就记住一个 `cluster` 和 `needUpdate` 就好了，其他的不太重要

我们进一步看下 `cluster` 的参数：**主要是 broker、topic、partition 的一些对应信息**

java

复制代码

`public final class Cluster { // 从命名直接就看出了各个变量的用途 private final boolean isBootstrapConfigured; private final List<Node> nodes; // node 列表 private final Set<String> unauthorizedTopics; // 未认证的 topic 列表 private final Set<String> internalTopics; // 内置的 topic 列表 private final Map<TopicPartition, PartitionInfo> partitionsByTopicPartition; // partition 的详细信息 private final Map<String, List<PartitionInfo>> partitionsByTopic; // topic 与 partition 的对应关系 private final Map<String, List<PartitionInfo>> availablePartitionsByTopic; // 可用（leader 不为 null）的 topic 与 partition 的对应关系 private final Map<Integer, List<PartitionInfo>> partitionsByNode; // node 与 partition 的对应关系 private final Map<Integer, Node> nodesById; // node 与 id 的对应关系 private final ClusterResource clusterResource; } // org.apache.kafka.common.PartitionInfo // topic-partition: 包含 topic、partition、leader、replicas、isr public class PartitionInfo { private final String topic; private final int partition; private final Node leader; private final Node[] replicas; private final Node[] inSyncReplicas; }`

### 四、元数据的请求及获取 ###

#### 1、Producer 元数据获取 ####

我们这块的元数据已经了解的差不多了，我们来看看元数据是怎么请求及获取的

当我们第一次发送 `Kafka` 消息时，会有一个 `waitOnMetadata(record.topic(), this.maxBlockTimeMs)` 方法

java

复制代码

`/** * 第一次发送消息时，这里会判断当前是否拿到了元数据 * 如果没有拿到元数据信息，这里会堵塞循环并唤醒 Sender 线程，让其帮忙更新元数据 */ long waitedOnMetadataMs = waitOnMetadata(record.topic(), this.maxBlockTimeMs); // 等待元数据的更新 private long waitOnMetadata(String topic, long maxWaitMs) throws InterruptedException { // 判断是否有元数据，没有的话，则一直循环堵塞 while (metadata.fetch().partitionsForTopic(topic) == null) { int version = metadata.requestUpdate(); // 唤醒我们的 Sender 线程，让其更新元数据 sender.wakeup(); metadata.awaitUpdate(version, remainingWaitMs); } return time.milliseconds() - begin; }`

这里如同我们的注释所说，会唤醒我们的 `Sender` 线程，让其帮忙更新我们的 `Metadata`

#### 2、Sender 获取元数据 ####

##### 2.1 第一次 poll 调用 #####

当 `Sender` 线程收到唤醒时，第一轮直接调用 `this.client.poll(pollTimeout, now)` 方法，如下：

java

复制代码

`void run(long now) { // 获取元数据中的 Cluster 信息 Cluster cluster = metadata.fetch(); this.client.poll(pollTimeout, now); } public List<ClientResponse> poll(long timeout, long now) { // 判断是否需要更新 metadata，如果需要就更新 long metadataTimeout = metadataUpdater.maybeUpdate(now); try { this.selector.poll(Utils.min(timeout, metadataTimeout, requestTimeoutMs)); } }`

我们看一下 `metadataUpdater.maybeUpdate` 做了什么

java

复制代码

`@Override public long maybeUpdate(long now) { // 根据当前更新的时间判断是否需要更新 long timeToNextMetadataUpdate = metadata.timeToNextUpdate(now); long timeToNextReconnectAttempt = Math.max(this.lastNoNodeAvailableMs + metadata.refreshBackoff() - now, 0); long waitForMetadataFetch = this.metadataFetchInProgress ? Integer.MAX_VALUE : 0; long metadataTimeout = Math.max(Math.max(timeToNextMetadataUpdate, timeToNextReconnectAttempt), waitForMetadataFetch); if (metadataTimeout == 0) { // 选择一个当前链接建立最少的node Node node = leastLoadedNode(now); // 更新元数据 maybeUpdate(now, node); } return metadataTimeout; } // 判断当前是否可以发送请求，可以的话将 metadata 请求加入到发送列表中 private void maybeUpdate(long now, Node node) { String nodeConnectionId = node.idString(); // 通道是否已经准备完毕 if (canSendRequest(nodeConnectionId)) { this.metadataFetchInProgress = true; MetadataRequest metadataRequest; if (metadata.needMetadataForAllTopics()) metadataRequest = MetadataRequest.allTopics(); else metadataRequest = new MetadataRequest(new ArrayList<>(metadata.topics())); ClientRequest clientRequest = request(now, nodeConnectionId, metadataRequest); doSend(clientRequest, now); } else if (connectionStates.canConnect(nodeConnectionId, now)) { // 初始化链接 initiateConnect(node, now); } else { this.lastNoNodeAvailableMs = now; } } } // 主要做的初始化与Broker 的连接 private void initiateConnect(Node node, long now) { String nodeConnectionId = node.idString(); try { this.connectionStates.connecting(nodeConnectionId, now); selector.connect(nodeConnectionId, new InetSocketAddress(node.host(), node.port()), this.socketSendBuffer, this.socketReceiveBuffer); } }`

这里会进入一个 `selector.connect` 方法，在这个里面进行与 `Broker` 的连接

java

复制代码

`public void connect(String id, InetSocketAddress address, int sendBufferSize, int receiveBufferSize) throws IOException { // 经典的 NIO 的实现（之前netty中聊过） SocketChannel socketChannel = SocketChannel.open(); socketChannel.configureBlocking(false); Socket socket = socketChannel.socket(); socket.setKeepAlive(true); // 发送的buffer if (sendBufferSize != Selectable.USE_DEFAULT_BUFFER_SIZE) socket.setSendBufferSize(sendBufferSize); // 接受的buffer if (receiveBufferSize != Selectable.USE_DEFAULT_BUFFER_SIZE) socket.setReceiveBufferSize(receiveBufferSize); // 设置TCP的状态 socket.setTcpNoDelay(true); boolean connected; try { // 直接连接服务端（这也是netty讲过的） connected = socketChannel.connect(address); } catch (UnresolvedAddressException e) { socketChannel.close(); throw new IOException("Can't resolve address: " + address, e); } // 注册接受事件（这个事件只能客户端注册） SelectionKey key = socketChannel.register(nioSelector, SelectionKey.OP_CONNECT); }`

##### 2.2 第二次 poll 调用 #####

这个时候已经与我们的 `Broker` 建立了连接，我们下一步要找在哪里发送的请求，也就是 `write` 方法。

当我们第二次调用 `poll` 方法时， 再次到达 `maybeUpdate` 方法时，这个时候我们会走 `canSendRequest(nodeConnectionId)` 分支

java

复制代码

`// 由于上述我们已经建立了连接，这里已经可以发送了 if (canSendRequest(nodeConnectionId)) { this.metadataFetchInProgress = true; MetadataRequest metadataRequest; if (metadata.needMetadataForAllTopics()) metadataRequest = MetadataRequest.allTopics(); else metadataRequest = new MetadataRequest(new ArrayList<>(metadata.topics())); ClientRequest clientRequest = request(now, nodeConnectionId, metadataRequest); doSend(clientRequest, now); }`

**一开始博主以为这里的 `doSend` 已经完成了发送，实际上并没有，大家一定要注意这一点**

java

复制代码

`private void doSend(ClientRequest request, long now) { request.setSendTimeMs(now); this.inFlightRequests.add(request); selector.send(request.request()); } public void send(Send send) { KafkaChannel channel = channelOrFail(send.destination()); this.send = send; this.transportLayer.addInterestOps(SelectionKey.OP_WRITE); }`

这里的 `doSend` 主要将我们本次请求放在 `inFlight` 这个组件里面（`inFlight：主要控制当前发送的请求量`）

另外，这里注册了一个 `SelectionKey.OP_WRITE` 事件，这个我们之前 `Netty` 大结局时讲过，有兴趣的可以去看一下

这里简单说一下，注册了 `SelectionKey.OP_WRITE` 事件，当内核的缓存区有空闲时，会触发该事件。

当注册完事件后，我们会调用 `this.selector.poll(Utils.min(timeout, metadataTimeout, requestTimeoutMs))` 方法，正式将我们的请求发送至 `Broker`

这里当前所有准备好的 `key` 拿出来，执行 `pollSelectionKeys` 去执行每一个 `key`

java

复制代码

`public void poll(long timeout) throws IOException { /* check ready keys */ long startSelect = time.nanoseconds(); // 执行 nioSelector.selectNow() 方法，得到所有触发的 key int readyKeys = select(timeout); long endSelect = time.nanoseconds(); currentTimeNanos = endSelect; this.sensors.selectTime.record(endSelect - startSelect, time.milliseconds()); if (readyKeys > 0 || !immediatelyConnectedKeys.isEmpty()) { pollSelectionKeys(this.nioSelector.selectedKeys(), false); pollSelectionKeys(immediatelyConnectedKeys, true); } }`

循环去遍历每一个 `key`，由于我们前面注册了 `OP_WRITE`，这里一定有一个写的 `key`，通过 `channel.write()` 方法写入到内核并发送至服务端

java

复制代码

`private void pollSelectionKeys(Iterable<SelectionKey> selectionKeys, boolean isImmediatelyConnected) { Iterator<SelectionKey> iterator = selectionKeys.iterator(); while (iterator.hasNext()) { SelectionKey key = iterator.next(); iterator.remove(); KafkaChannel channel = channel(key); try { if (channel.ready() && key.isWritable()) { Send send = channel.write(); if (send != null) { this.completedSends.add(send); this.sensors.recordBytesSent(channel.id(), send.size()); } } }`

##### 2.3 第三次 poll 调用 #####

当我们第三次调用 `poll` 时，这个时候我们发送的请求的响应回来了

java

复制代码

`// 处理响应 long updatedNow = this.time.milliseconds(); List<ClientResponse> responses = new ArrayList<>(); // 处理 inFlight 组件 handleCompletedSends(responses, updatedNow); // 处理 metadata handleCompletedReceives(responses, updatedNow); // 处理断开的连接 handleDisconnections(responses, updatedNow); // 记录任何新完成的连接 handleConnections(); // 处理超时的连接 handleTimedOutRequests(responses, updatedNow);`

其他的我们暂时不研究，主要看 `handleCompletedReceives` 对于元数据的处理

java

复制代码

`private void handleCompletedReceives(List<ClientResponse> responses, long now) { for (NetworkReceive receive : this.selector.completedReceives()) { String source = receive.source(); ClientRequest req = inFlightRequests.completeNext(source); Struct body = parseResponse(receive.payload(), req.request().header()); if (!metadataUpdater.maybeHandleCompletedReceive(req, now, body)) responses.add(new ClientResponse(req, now, false, body)); } } // 如果当前的是 metadata 响应，则更新其元数据 public boolean maybeHandleCompletedReceive(ClientRequest req, long now, Struct body) { short apiKey = req.request().header().apiKey(); if (apiKey == ApiKeys.METADATA.id && req.isInitiatedByNetworkClient()) { handleResponse(req.request().header(), body, now); return true; } return false; } // 拿到当前响应里面的各种信息，进行封装保存 private void handleResponse(RequestHeader header, Struct body, long now) { this.metadataFetchInProgress = false; MetadataResponse response = new MetadataResponse(body); Cluster cluster = response.cluster(); // check if any topics metadata failed to get updated Map<String, Errors> errors = response.errors(); if (cluster.nodes().size() > 0) { this.metadata.update(cluster, now); } else { this.metadata.failedUpdate(now); } }`

这里可以给大家看一下题主拉取的元数据的信息

json

复制代码

`{ "brokers":[ { "node_id":"2", "host":"12.23.14.13", "port":"9192", "rack":"null" }, { "node_id":"1", "host":"12.23.14.14", "port":"9192", "rack":"null" } ], "controller_id":"2", "topic_metadata":[ { "topic_error_code":"0", "topic":"activity_test", "is_internal":"false", "partition_metadata":[ { "partition_error_code":"0", "partition_id":"1", "leader":"1", "replicas":[ "1", "2" ], "isr":[ "1", "2" ] }, { "partition_error_code":"0", "partition_id":"0", "leader":"2", "replicas":[ "2", "1" ], "isr":[ "2", "1" ] } ] } ] }`

到这里，我们的元数据正式拿到并封装到我们的 `Metadata` 中

然后我们的 `Prodecer` 就可以拿到 `Metadata` 的信息，进行后续业务逻辑

#### 3、验证 ####

我们口说无凭，利用代码来验证下真正的流程逻辑

![format_png 1][]

我们在每一个流程里面都加了相应的日志代码，如果我们的验证是没有问题的，那么输出的日志也应该按照我们预想的一样

启动我们的程序：

![format_png 2][]

我们可以明显的看到，程序的逻辑符合预计的执行

*  `Producer` 堵塞，唤醒 `Sender` 线程让其更新 `Metadata`
 *  `Sender` 第一次调用 `poll` 方法，初始化与 `Broker` 的连接
 *  `Sender` 第二次调用 `poll` 方法，向 `Selector` 注册写事件并将请求发送到 `Broker`
 *  `Sender` 第三次调用 `poll` 方法，处理元数据的响应信息
 *  `Producer` 不阻塞，`Sender` 已更新完元数据信息，继续执行其业务逻辑

### 五、总结 ###

本章主要讲解了：

*  `kafka` 为什么不用 `Netty` 的原因
 *  `Metadata` 是什么
 *  `Metadata` 的请求和获取
    
     *  `Producer` 的堵塞与唤醒
     *  初始化 `Broker` 的连接
     *  注册写事件并发送请求至 `Broker`
     *  接受响应处理响应的 `Metadata` 信息

下一章我们将会更新 `Sender` 如何拉取 `RecordAccumulator` 的数据并将其发送至 `Broker`

喜欢 `kafka` 的可以点个关注吆，后续会继续更新其源码文章。

我是爱敲代码的小黄，我们下次再见。

[format_png]: https://img-blog.csdnimg.cn/img_convert/f991140528a2f940eaf0ab9bf7227984.webp?x-oss-process=image/format,png
[format_png 1]: https://img-blog.csdnimg.cn/img_convert/8285a7446c2ae7665a95c7f44d7a4c48.webp?x-oss-process=image/format,png
[format_png 2]: https://img-blog.csdnimg.cn/img_convert/4c9292ddf0943d6c626e03ed6415a605.webp?x-oss-process=image/format,png