首页 > 编程语言 >Kafka源码分析(四) - Server端-请求处理框架

Kafka源码分析(四) - Server端-请求处理框架

时间:2024-05-06 18:22:05浏览次数:28  
标签:... 请求 PRODUCE Kafka 源码 线程 Server Processor Acceptor

系列文章目录

https://zhuanlan.zhihu.com/p/367683572

一. 总体结构

先给一张概览图:

服务端请求处理过程涉及到两个模块:kafka.networkkafka.server

1.1 kafka.network

该包是kafka底层模块,提供了服务端NIO通信能力基础。

有4个核心类:SocketServer、Acceptor、Processor、RequestChannel。各自角色如下:

  • SocketServer:服务端的抽象,是服务端通信的入口;

  • Acceptor:Reactor通信模式中处理连接ACCEPT事件的线程/线程池所执行的任务;

  • Processor:Reactor通信模式中处理连接可读/可写事件的线程/线程池所执行的任务;

  • RequestChannel:请求队列,存储已经解析好的请求以等待处理;

对于上层模块而言,该基础模块有两个输入和一个输出

  1. 输入:IP+端口号,该模块会对目标端口实现监听;

  2. 输出:解析好的请求,通过RequestChannel进行输出;

  3. 输入:待发送的Response,通过Processor.responseQueue来完成输入;

1.2 kafka.server

该包在kafka.network的基础上实现各种请求的处理逻辑,主要包含KafkaServer和KafkaApis两个类。其中:

  • KafkaServer:Kafka服务端的抽象,统一维护Kafka服务端的各流程和状态;

  • KakfaApis:维护了各类请求对应的业务逻辑,通过KafkaServer.apis字段组合到KafkaServer之中;

二. Server的端口监听

整体流程如图:

接下来按调用顺序依次分析各方法

2.1 KafkaServer.startup()

关于端口监听的核心逻辑分4步,代码如下(用注释说明各部分的目的):

def startup() {
      // 省略无关代码
      ... ...

      // 1. 创建SocketServer
      socketServer = new SocketServer(config, metrics, time, credentialProvider)

      // 2. 启动端口监听
      // (在这里完成了Acceptor的创建和端口ACCEPT事件的监听)
      // (startupProcessors = false表示暂不启动Processor处理线程)
      socketServer.startup(startupProcessors = false)

      // 3. 启动请求处理过程中的相关依赖
      // (这也是第2步中不启动Processor处理线程的原因,有依赖项需要处理)
      ... ...

      // 4. 启动端口可读/可写事件处理线程(即Processor线程)
      socketServer.startProcessors()

      // 省略无关代码
      ... ...
}

2.2 SocketServer.startup(Boolean)

代码及说明性注释如下:

def startup(startupProcessors: Boolean = true) {
    this.synchronized {
      // 省略无关代码
      ... ...

      // 1. 创建Accetpor和Processor的实例,
      // 同时页完成了Acceptor对端口ACCEPT事件的监听
      createAcceptorAndProcessors(config.numNetworkThreads, config.listeners)

      // 2. [可选]启动各Acceptor对应的Processor线程
      if (startupProcessors) {
        startProcessors()
      }
    }
}

2.3 ScocketServer.createAcceptorAndProcessor()

直接上注释版的代码,流程分3步:

// 入参解释
// processorsPerListener: 对于每个IP:Port, 指定Reactor模式子线程池大小, 
//                        即处理端口可读/可写事件的线程数(Processor线程);
// endpoints: 接收请求的IP:Port列表;
def createAcceptorAndProcessors(processorsPerListener: Int,
                                endpoints: Seq[EndPoint]): Unit = synchronized {
    // 省略无关代码
    ... ...

    endpoints.foreach { endpoint =>
      // 省略无关代码
      ... ...

      // 1. 创建Acceptor对象
      // 在此步骤中调用Acceptor.openServerSocket, 完成了对端口ACCEPT事件的监听
      val acceptor = new Acceptor(endpoint, sendBufferSize, recvBufferSize, brokerId, connectionQuotas)

      // 2. 创建了与acceptor对应的Processor对象列表
      // (这里并未真正启动Processor线程)
      addProcessors(acceptor, endpoint, processorsPerListener)

      // 3. 启动Acceptor线程
      KafkaThread.nonDaemon(s"kafka-socket-acceptor-$listenerName-$securityProtocol-${endpoint.port}", acceptor).start()

      // 省略无关代码
      ... ...
    }
  }

2.4 Acceptor.openServerSocket()

该方法中没什么特殊点,就是java NIO的标准流程:

def openServerSocket(host: String, port: Int): ServerSocketChannel = {
    // 1. 构建InetSocketAddress对象
    val socketAddress =
      if (host == null || host.trim.isEmpty)
        new InetSocketAddress(port)
      else
        new InetSocketAddress(host, port)

    // 2. 构建ServerSocketChannel对象, 并设置必要参数值
    val serverChannel = ServerSocketChannel.open()
    serverChannel.configureBlocking(false)
    if (recvBufferSize != Selectable.USE_DEFAULT_BUFFER_SIZE)
      serverChannel.socket().setReceiveBufferSize(recvBufferSize)

    // 3. 端口绑定, 实现事件监听
    try {
      serverChannel.socket.bind(socketAddress)
      info("Awaiting socket connections on %s:%d.".format(socketAddress.getHostString, serverChannel.socket.getLocalPort))
    } catch {
      case e: SocketException =>
        throw new KafkaException("Socket server failed to bind to %s:%d: %s.".format(socketAddress.getHostString, port, e.getMessage), e)
    }

    // 4. 返回ServerSocketChannel对象, 用于后续register到Selector中
    serverChannel
  }

2.5 SocketServer.startProcessor()

从这步开始,仅剩的工作就是启动Processor线程,代码都非常简单。比如本方法只是遍历Acceptor列表,并调用Acceptor.startProcessors()

def startProcessors(): Unit = synchronized {
  acceptors.values.asScala.foreach { _.startProcessors() }
  info(s"Started processors for ${acceptors.size} acceptors")
}

2.6 Acceptor.startProcessors()

该方法很简明,直接上代码

def startProcessors(): Unit = synchronized {
  if (!processorsStarted.getAndSet(true)) {
    startProcessors(processors)
  }
}

def startProcessors(processors: Seq[Processor]): Unit = synchronized {
  processors.foreach { processor =>
    KafkaThread.nonDaemon(s"kafka-network-thread-$brokerId-${endPoint.listenerName}-${endPoint.securityProtocol}-${processor.id}",
      processor).start()
  }
}

三. 请求/响应的格式

3.1 格式概述

请求和响应都由两部分组成:Header和Body。RequestHeader中包含ApiKey、ApiVersion、CorrelationId、ClientId;ResponseHeader中只包含CorrelationId字段。接下来逐个讲解这些字段。

  • ApiKey

    2字节整型,指明请求的类型;比如0代表Produce请求,1代表Fetch请求;具体id和请求类型之间的映射关系可在 org.apache.kafka.common.protocol.ApiKeys 中找到;

  • ApiVersion

    随着API的升级迭代,各类型请求的请求体格式可能有变更;这个2字节的整型指明了请求体结构的版本;

  • CorrelationId

    4字节整型,在Response中传回,Kafka Server端不处理,用于客户端内部关联业务数据;

  • ClientId

    可变长字符串,标识客户端;

3.2 请求体/响应体的具体格式

各业务操作(比如Produce、Fetch等)对应的请求体和响应体格式都维护在 org.apache.kafka.common.protocol.ApiKeys 中。接下来以Produce为例讲解ApiKeys是如何表达数据格式的。

ApiKeys是个枚举类,其核心属性如下:

public enum ApiKeys {
  // 省略部分代码
  ... ...

  // 上文提到的请求类型对应的id
  public final short id;

  // 业务操作名称
  public final String name;

  // 各版本请求体格式
  public final Schema[] requestSchemas;

  // 各版本响应体格式
  public final Schema[] responseSchemas;

  // 省略部分代码
  ... ...
}

其中PRODUCE枚举项的定义如下

PRODUCE(0, "Produce", ProduceRequest.schemaVersions(), ProduceResponse.schemaVersions())

可以看到各版本的请求格式维护在 ProduceRequest.schemaVersions(),代码如下

public static Schema[] schemaVersions() {
  return new Schema[] {PRODUCE_REQUEST_V0, PRODUCE_REQUEST_V1, PRODUCE_REQUEST_V2, PRODUCE_REQUEST_V3,
    PRODUCE_REQUEST_V4, PRODUCE_REQUEST_V5, PRODUCE_REQUEST_V6};
}

这里只是简单返回了一个Schema数组。一个Schema对象代表了一种数据格式。请求头中的ApiVersion指明了请求体的格式对应数组的第几项(从0开始)。

接下来我们看看Schema是如何表达数据格式的。其结构如下

Schema有两个字段:fields和fieldsByName。其中fields是体现数据格式的关键,它指明了字段的排序和各字段类型;而fieldsByName只是按字段名重新组织的Map,用于根据名称查找对应字段。

BoundField只是Field的简单封装。Field有两个核心字段:name和type。其中name表示字段名称,type表示字段类型。常见的Type如下:

Type.BOOLEAN;
Type.INT8;
Type.INT16;
Type.INT32;

// 可通过org.apache.kafka.common.protocol.types.Type查看全部类型
... ...

回到PRODUCE API,通过查看Schema的定义,能看到其V0版本的请求体和响应体的结构如下:

四. 请求的处理流程

  1. Acceptor监听到ACCEPT事件(TCP创建连接"第一次握手"的SYN);

  2. Acceptor将将连接注册到Processor列表内的其中一个,由该Processor监听这个连接的后续可读可写事件;

  3. Processor接收到完整请求后,会将Request追加到RequestChannel中进行排队,等待后续处理;

  4. KafkaServer中有个requestHandlerPool的字段,KafkaRequestHandlerPool类型,代表请求处理线程池;KafkaRequestHandler就是其中的线程,会从RequestChannel拉请求进行处理;

  5. KafkaRequestHandler将拉到的Request传入KafkaApis.handle(Request)方法进行处理;

  6. KafkaApis根据不同的ApiKey调用不同的方法进行处理,处理完毕后会将Response最终写入对应的Processor的ResponseQueue中等待发送;KafkaApis.handle(Request)的方法结构如下:

def handle(request: RequestChannel.Request) {
  try {
    // 省略部分代码
    ... ...
    request.header.apiKey match {
      case ApiKeys.PRODUCE => handleProduceRequest(request)
      case ApiKeys.FETCH => handleFetchRequest(request)
      case ApiKeys.LIST_OFFSETS => handleListOffsetRequest(request)
      case ApiKeys.METADATA => handleTopicMetadataRequest(request)
      case ApiKeys.LEADER_AND_ISR => handleLeaderAndIsrRequest(request)
      // 省略部分代码
      ... ...
    }
  } catch {
    case e: FatalExitError => throw e
    case e: Throwable => handleError(request, e)
  } finally {
    request.apiLocalCompleteTimeNanos = time.nanoseconds
  }
}
  1. Processor从自己的ResponseQueue中拉取待发送的Respnose;

  2. Processor将Response发给客户端;

五. 总结

才疏学浅,未能窥其十之一二,随时欢迎各位交流补充。若文章质量还算及格,可以点赞收藏加以鼓励,后续我继续更新。


知乎主页:https://www.zhihu.com/people/hao_zhihu
关注收藏不迷路,第一时间接收技术文章推送


微信公众号:
微信二维码

标签:...,请求,PRODUCE,Kafka,源码,线程,Server,Processor,Acceptor
From: https://www.cnblogs.com/zhanghao2244/p/18174173

相关文章

  • Unable to execute SonarScanner analysis: Fail to get bootstrap index from server
    1.背景编辑gitlab-ci流水线时,代码分析的job,maven使用sonar报错-mvncleanverifysonar:sonar-Dsonar.login=30c55d3b8d3d2569431fb39f3c488c90643a68442.错误信息[ERROR]Failedtoexecutegoalorg.sonarsource.scanner.maven:sonar-maven-plugin:3.11.0.3922:sonar(def......
  • SQL Server 查询数据库中所有表数据条数
    有的时间我想把数据库中表的记录统计一下,如果我们一个一个表的操作可以直接selectcount(*)fromtablename就可以然后一个个相加,但是如果有上百个表有没有更简单的方法呢,下面我总结了一些方法有需要的朋友可参考。如果是要得到中所有表的条数呢?我们来看几种最常见的方式:--方法......
  • web server apache tomcat11-31-websocket
    前言整理这个官方翻译的系列,原因是网上大部分的tomcat版本比较旧,此版本为v11最新的版本。开源项目从零手写实现tomcatminicat别称【嗅虎】心有猛虎,轻嗅蔷薇。系列文章webserverapachetomcat11-01-官方文档入门介绍webserverapachetomcat11-02-setup启动web......
  • sqlserver笔记
    明确的性能低的定义:在现有资源还没有达到最大吞吐量的前提下,系统如果不能满足合理的预期表现。最小化每个SQL的响应时间;合理增加吞吐量;减少网络延时优化磁盘IO、CPU能够协调、平衡的运行,合理的响应外部的请求,实现资源利用的最大化。影响性能的常见因素:1.数据库结构的设计--了解......
  • SSM教务管理系统设计与实现(附源码下载地址)
    @目录01项目背景02使用技术03运行环境04功能分析05数据库设计06项目工程结构07部分功能展示及源码7.1登录页7.2管理员端--首页7.3管理员端--课程管理7.4管理员端--学生管理7.5教师端--首页7.6教师端--个人信息7.7学生端--已修课程7.8学生端--公告管理08运行教程09......
  • 进入ThinkServer RQ940服务器的UEFI HII SAS RAID阵列配置界面
    内容导航 一、进入SASRAID阵列配置界面 二、硬盘选择操作 三、确认操作 一、进入SASRAID阵列配置界面 启动服务器,在出现ThinkServerlogo的时候按F2或者Delete键进入BIOS配置界面; 移至"Boot"选项卡,设置"Bootmodeselect"设置为"UEFI";  再移至"Advanced......
  • Kubernetes源码二次开发系列(一到三),玩转高阶K8s开发
    Kubernetes源码二次开发通常涉及以下步骤:环境准备:安装和配置Go环境,确保Go版本与Kubernetes项目兼容。获取源码:从GitHub克隆Kubernetes仓库到本地。设置工作环境:通过make命令设置编译环境。修改源码:根据需求修改Kubernetes源码。编译:使用makeall编译所有组......
  • GeoServer开发环境搭建
    前言本文用于记录GeoServer开发环境的搭建过程通过GeoServer发布计划可以看到,在2.23.x版本开始,会移除对jdk1.8的支持。那么当前我们会选择2.22.x版本进行研究环境JAVA:1.8或11MavenGitAction获取源码gitclonegit://github.com/geoserver/geoserver.gitgeoserver#......
  • im即时通讯源码/仿微信app源码+php即时通讯源码带红包+客服+禁言等系统php+uniapp开发
    即时通讯(IM)系统是现代互联网应用中不可或缺的一部分,它允许用户进行实时的文本、语音、视频交流。随着技术的发展,IM系统的功能越来越丰富,如红包、客服、禁言等。本文将探讨如何使用PHP语言开发一个功能完备的即时通讯系统,包括源码解析、系统架构、关键功能实现等。仓库源码:ym.fz......
  • std::move 和 std::forward源码解析
    std::move和std::forward仅仅是执行转换(cast)的函数(事实上是函数模板)。std::move无条件的将它的实参转换为右值,而std::forward只在特定情况满足时下进行转换。std::movetemplate<class_Ty>constexprremove_reference_t<_Ty>&&move(_Ty&&_Arg)noexcept{returnstatic......