ZooKeeper到底为Kafka的做了什么牺牲？

原创

JavaEdge

修改于 2020-09-17 10:30:05

9390

修改于 2020-09-17 10:30:05

文章被收录于专栏：JavaEdgeJavaEdge

全是干货的技术号：本文已收录在github，欢迎 star/fork： https://github.com/Wasabi1234/Java-Interview-Tutorial

Kafka使用zk实现和RocketMQ的NameServer相似的功能。

1 Kafka的zk有什么作用？

首先我们来看一下Kafka在ZooKeeper都保存了哪些信息：

0.8.x的旧版本的情况，最新版本的Kafka已经将消费位置管理等一些原本依赖ZooKeeper实现的功能，替换成了其他的实现方式。
绿色框形是临时节点，其它是持久节点。

1.1 ids 子树（临时节点）

保存的是Kafka的Broker信息，/brokers/ids/[0…N]，每个临时节点对应一个在线Broker，Broker启动后会创建一个临时节点，代表Broker已经加入集群，可提供服务了，节点名称就是BrokerID，节点内保存了包括Broker的地址、版本号、启动时间等信息。若Broker宕机或与zk集群失联，该临时节点也会消失。

1.2 topics 子树

topic

保存topic和partition信息。/brokers/topics/节点下的每个子节点都是一个topic，节点的名称就是topic名称。

partitions

每个topic节点下面都包含 一个固定 的partitions节点，pattitions节点的子节点就是主题下的所有分区，节点名称即分区编号。

state（临时节点）

位于每个分区节点下，其保存着 分区当前的leader和所有ISR的BrokerID。该state临时节点由该分区当前的Leader Broker创建。

若该分区的Leader Broker宕机，对应state节点也会消失，直至新Leader被选举出来，再次创建state节点。

这份元数据同时被缓存到每一个Broker。

Kafka主要使用zk保存其元数据、监控Broker和分区的存活状态，并利用zk进行选举。

2 Kafka client如何定位Broker？

先根据topic和queue，在topics树中找到分区对应的state临时节点。从中取得该Leader的BrokerID，再去ids 子树，找到BrokerID对应的临时节点，即可获取到Broker真正的物理地址。

Kafka client不会直接连zk，而是在需要时，通过RPC请求从Broker拉取所关心的topic的元数据，然后保存到client的元数据缓存，以便client的生产和消费。 zk上的元数据都是通过Broker中转给各个Kafka client。

Kafka client（客户端）真正与Broker（服务端）数据通信是在NetworkClient#poll实现，经过如下调用链：

该方法里面，Kafka构造了一个更新元数据请求：

该方法创建的并不是一个真正的更新元数据的MetadataRequest，而是一个用于构造MetadataRequest的构造器MetadataRequest.Builder，等到真正要发送请求之前，Kafka才会调用Builder.buid()方法把这个MetadataRequest构建出来然后发送出去。其实不仅是元数据的请求，所有的请求都是这样处理的。

调用sendInternalMetadataRequest()方法时，这个请求也并没有被真正发出去，依然是保存在待发送的队列中，然后择机来异步批量发送。

请求的具体数据内容封装在MetadataRequestData 类

3 Broker如何处理客户端的更新元数据请求？

Broker处理所有RPC请求的入口方法

KafkaApis#handleTopicMetadataRequest

处理更新元数据的方法

handleTopicMetadataRequest(RequestChannel.Request)：

 def handleTopicMetadataRequest(request: RequestChannel.Request): Unit = {
 	...
    // 需要获取哪些topic的元数据
    val topics = if (metadataRequest.isAllTopics)
      metadataCache.getAllTopics()
    // 不会披露未经Describe授权的主题的存在，因此甚至都没有检查它们是否存在
    val unauthorizedForDescribeTopicMetadata =
      // 对于所有主题，请勿包括未经授权的主题
      // 在旧版本的协议中，每次都获取所有主题的元数据
      if ((requestVersion == 0 && (metadataRequest.topics == null || metadataRequest.topics.isEmpty)) || metadataRequest.isAllTopics)
        Set.empty[MetadataResponseTopic]
      else
        unauthorizedForDescribeTopics.map(topic =>
          metadataResponseTopic(Errors.TOPIC_AUTHORIZATION_FAILED, topic, false, util.Collections.emptyList()))
        // 从元数据缓存过滤出相关主题的元数据
        getTopicMetadata(metadataRequest.allowAutoTopicCreation, authorizedTopics, request.context.listenerName,
          errorUnavailableEndpoints, errorUnavailableListeners)

    var clusterAuthorizedOperations = Int.MinValue
    if (request.header.apiVersion >= 8) {
      // 获取集群授权的操作
      if (metadataRequest.data.includeClusterAuthorizedOperations) {
      	...
      // 获取主题授权操作
      if (metadataRequest.data.includeTopicAuthorizedOperations) {
      	...
    val completeTopicMetadata = topicMetadata ++ unauthorizedForCreateTopicMetadata ++ unauthorizedForDescribeTopicMetadata
    // 获取所有Broker列表
    val brokers = metadataCache.getAliveBrokers
    // 构建Response并发送
    sendResponseMaybeThrottle(request, requestThrottleMs =>
    ...
  }