ZooKeeper到底为Kafka的做了什么牺牲?(下)

本文涉及的产品
服务治理 MSE Sentinel/OpenSergo,Agent数量 不受限
简介: ZooKeeper到底为Kafka的做了什么牺牲?(下)

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

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

KafkaApis#handleTopicMetadataRequest

  • 处理更新元数据的方法
  • image.png
  • 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 =>
    ...
  }

先根据请求中的topic列表

去本地元数据缓存MetadataCache中过滤出相应主题的元数据,即 topics 子树的子集

然后再去本地元数据缓存中获取所有Broker的集合, 即 ids 子树

最后把这两部分合在一起,作为响应返回给客户端。

Kafka在每个Broker中都维护了一份和zk中一样的元数据缓存,并非每次client请求元数据就去读一次zk。由于zk的Watcher机制,Kafka可感知到zk中的元数据变化,从而及时更新Broker的元数据缓存。

4 最佳实践

目前Kafka集群的可用性高度耦合zk,若zk集群不能提供服务,整个Kafka集群就无法服务了,Kafka的开发者也意识到了这个问题,目前正在讨论开发一个元数据服务来替代 zk。

若需部署大规模Kafka集群,推荐拆分成多个互相独立的小集群部署,每个小集群都使用一组独立的zk提供服务。这样,每个zk中存储的数据相对较少,且若某zk集群异常,只会影响一个小Kafka集群,尽量减小了影响范围。


参考


https://www.bcoder.top/2019/12/14/%E6%B6%88%E6%81%AF%E9%98%9F%E5%88%97%E4%B9%8BKafka%E5%8D%8F%E8%B0%83%E6%9C%8D%E5%8A%A1ZooKeeper/


目录
相关文章
|
3月前
|
消息中间件 数据可视化 Kafka
消息中间件系列教程(21) -Kafka- 集群搭建(自带Zookeeper)
消息中间件系列教程(21) -Kafka- 集群搭建(自带Zookeeper)
53 0
|
6天前
|
消息中间件 Kafka Docker
【docker专题_04】docker搭建kafka与zookeeper
【docker专题_04】docker搭建kafka与zookeeper
15 2
|
1月前
|
消息中间件 存储 Java
ZooKeeper 在 Kafka 中的应用
ZooKeeper 在 Kafka 中的应用
29 0
|
1月前
|
消息中间件 Java Kafka
使用Java编写代码安装Kafka及启动Zookeeper和Kafka
如何使用Java编写代码安装Kafka及启动Zookeeper和Kafka?其中包括安装前准备、修改配置文件、创建日志目录和数据目录等步骤,请提供详细实现过程。
33 0
|
2月前
|
消息中间件 Java Shell
Linux【脚本 03】shell脚本离线安装配置集结JDK+InfluxDB+Zookeeper+Kafka(安装文件及脚本源码网盘分享)
Linux【脚本 03】shell脚本离线安装配置集结JDK+InfluxDB+Zookeeper+Kafka(安装文件及脚本源码网盘分享)
26 0
|
2月前
|
消息中间件 Kafka Shell
Linux【脚本 02】shell脚本离线安装配置Zookeeper及Kafka并添加service服务和开机启动(脚本分析)
Linux【脚本 02】shell脚本离线安装配置Zookeeper及Kafka并添加service服务和开机启动(脚本分析)
32 0
|
3月前
|
消息中间件 Kafka Linux
kafka3.0创建topic出现zookeeper is not a recognized option
kafka3.0创建topic出现zookeeper is not a recognized option
35 0
|
22天前
|
消息中间件 安全 Kafka
2024年了,如何更好的搭建Kafka集群?
我们基于Kraft模式和Docker Compose同时采用最新版Kafka v3.6.1来搭建集群。
2024年了,如何更好的搭建Kafka集群?
|
1月前
|
消息中间件 存储 数据可视化
kafka高可用集群搭建
kafka高可用集群搭建
32 0
|
1月前
|
消息中间件 Kafka Linux
Apache Kafka-初体验Kafka(03)-Centos7下搭建kafka集群
Apache Kafka-初体验Kafka(03)-Centos7下搭建kafka集群
34 0

热门文章

最新文章