作者：石臻臻, CSDN博客之星Top5、Kafka Contributor 、nacos Contributor、华为云 MVP ,腾讯云TVP, 滴滴Kafka技术专家 、 LogiKM PMC(改名KnowStream)。
LogiKM(改名KnowStream) 是滴滴开源的Kafka运维管控平台, 有兴趣一起参与参与开发的同学,但是怕自己能力不够的同学,可以联系我,当你导师带你参与开源！ 。
文章目录
最近有个同学问了我一个非常有意思的问题, 今天我根据这个问题来给大家好好分析一下。
前提知识
我们都知道, 每个Broker都可以配置多个监听器, 用来用于网络分流。
相关知识请看：一文搞懂Kafka中的listeners和advertised.listeners以及其他通信配置
然后, 我们客户端中需要配置bootstrap.servers=xxxx:port 来连接到集群中。
然后当Kafka集群Broker数量很多的时候,我们不可能在bootstrap.servers配置所有的地址
所以Kafka是允许你只配置其中部分地址的, 它会通过自身的元信息更新机制,去获取Kafka集群中的所有地址。然后如果需要去跟某一台Broker发起连接的话,就去元信息里面获取地址。
问题图述
那么问题来了, 既然一台Broker能够配置多个Listener, 也就意味着有多个地址, 那么客户端在跟具体的Broker发起请求的话, 应该选择哪一个Listener？是遍历吗？
上图客户端部分获取到的 Broker列表EndPoint应该是什么呢？
问题源码探究
首先, 客户端(生产者、消费者)去获取集群元信息是通过元信息更新器MetadataUpdater
具体的元信息更新器流程请看 客户端发起元信息更新请求.
我们重点看一下, 获取元信息返回之后,是如何解析Broker集群列表的,确定一下是不是把集群所有的 EndPoint都获取了,还是只获取了一部分。
解析返回的元信息
直接定位到关键代码
Metadata#handleMetadataResponse
private MetadataCache handleMetadataResponse(MetadataResponse metadataResponse, boolean isPartialUpdate, long nowMs) {
//省略部分
Map<Integer, Node> nodes = metadataResponse.brokersById();
if (isPartialUpdate)
return this.cache.mergeWith(metadataResponse.clusterId(), nodes, partitions,
unauthorizedTopics, invalidTopics, internalTopics, metadataResponse.controller(),
(topic, isInternal) -> !topics.contains(topic) && retainTopic(topic, isInternal, nowMs));
else
return new MetadataCache(metadataResponse.clusterId(), nodes, partitions,
unauthorizedTopics, invalidTopics, internalTopics, metadataResponse.controller());
}
源码调试
本次启动的Kafka集群网络相关配置如下
①. server0.properties
listeners = PLAINTEXT://localhost:9090
②. server1.properties
listeners = PLAINTEXT://localhost:9091,TEXT://localhost:9099
listener.security.protocol.map=PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL,TEXT:PLAINTEXT
③. server2.properties
listeners = PLAINTEXT://localhost:9090
上面的配置, 只有 server1. 中的监听器配置了2个。
- PLAINTEXT://localhost:9091
- TEXT://localhost:9099
然后启动一个KafkaProducer客户端, bootstrap.servers=localhost:9099 .
通过Debug发现, KafkaProducer客户端获取到的元信息集群列表只有
localhost:9099 (id: 1 rack: null)
也就是说,客户端拿到的信息是在对应Broker处理请求那里就已经做好的筛选了。
为了搞清楚为什么这里只拿到了一个EndPoint信息, 我们需要去看看Broker是如何处理请求的。
注意：这里发出去的请求是 UPDATE_METADATA
所以, 从客户端发出UPDATE_METADATA请求之后, 服务端是如何处理的呢？
处理handleTopicMetadataRequest请求
在分析这个请求之前, 我想再补充一点关于服务端网络通信模型的知识.
网络通信模型
了解更多请看：图解Kafka的服务端的网络通信模型
先看一张服务端网络模型架构图
在Kafka启动的时候, 会根据Listener配置,启动对应个数的 Acceptor 和 Processor
比如在我们这个例子中, 有2个Listener配置, 那么就如下图所示(简化)
Acceptor：是专门用来监听连接过来的新链接请求的。
Kafka启动的时候会创建对应个数的Acceptor，这个Acceptor持有很多的信息, 比如ChannelBuilder, 这个ChannelBuilder持有 ListenerName, 如下图所示
也就是说, 不管哪个客户端从哪个监听器访问到服务端, 都是可以确定它对应的监听器名称的。
比如, 配置了下面2个监听器, 如果我客户端通过localhost:9099访问到了Broker, 那么跟这个客户端建立链接的Acceptor就是监听器名为：TEXT 的那个。
- PLAINTEXT://localhost:9091
- TEXT://localhost:9099
知道这么一个前提之后, 我们再来分析如何处理handleTopicMetadataRequest请求
服务端接受请求入队
当对应的Processor监听到请求过来的时候,会将请求解析一下并组装成Request,然后入队
Processor#processCompletedReceives
我们可以看到, 在组装Request的时候, 是有把listenerName传入的。
所以：Request持有ListenerName。
处理元信息更新请求 handleTopicMetadataRequest
KafkaApis#handleTopicMetadataRequest
关键代码
从代码中可以看到, 我们拿到了所有的Brokers的EndPoint, 包括多个监听器也都拿到了。
例子中, server1 配置了2个listener, 这里就有2个EndPoint。
但是真正把数据发往会客户端的时候, 是有根据listenerName做过滤的！
brokers.flatMap(_.getNode(request.context.listenerName)).asJava,
这个listenerName也就是我们上面一开始分析过的, Processor对应着监听器。
所以我们这个例子中, 监听器名是 TEXT 。
过滤完了之后,是不是发现只有一个复合要求,也就是他自己有一个TEXT监听器。
假如我们客户端发起请求的时候，bootstrap.servers=localhost:9091, 是不是就命中的PLAINTEXT监听器。
因为这个例子中每个Broker都配置了PLAINTEXT监听器, 所以最终会返回3个EndPoint。
结论
客户端对服务端发起请求的时候, 会根据命中的服务端的监听器, 然后根据这个命中的服务端监听器名listenerName，过滤集群中其他Broker同样是配置了这个监听器名称的EndPoint。
同样用一张图来回答一下最开始的问题图述的问题
因为图片里面的case，是从listener2 监听到的请求, 那么所有Broker的EndPoint也要用Listener2的监听器名称来进行过滤，也就会得到图中的结果。
问题
如果客户端bootstrap.servers配置了多个地址，并且这些地址对应的监听器名字还不一样会有啥后果？
举个例子：
服务端配置：
server1
listeners = PLAINTEXT1://IP1:9090,PLAINTEXT3://IP1:9092
server2
listeners = PLAINTEXT1://IP2:9090,PLAINTEXT2://IP2:9091,PLAINTEXT3://IP2:9092
server3
listeners = PLAINTEXT1://IP3:9090,PLAINTEXT2://IP3:9091,PLAINTEXT3://IP3:9092
客户端配置
bootstrap.servers=IP1:9090,IP2:9090,IP3:9091
首先,客户端发起请求的时候,是去bootstrap.servers获取一个最小负载的IP, 然后去获取元数据。
比如第一次更新的时候,我们去 IP1:9090请求元数据了。拿到的listenerName=PLAINTEXT1
这个时候我们拿到的Brokers是 {IP1:9090、IP2:9090、IP3:9090} .
当后续更新的时候,如果去IP3:9091获取数据的时候,拿到的listenerName=PLAINTEXT2
这个时候我们拿到的Brokers是 {IP2:9091、IP3:9091} .
所以：客户端配置bootstrap.servers的时候,尽量配置的地址都是属于同一个ListenerName 。