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

系列文章目录

Kafka源码分析-目录

一. 总体结构

先给一张概览图:
在这里插入图片描述

服务端请求处理过程涉及到两个模块: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. 创建SocketServersocketServer = 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)elsenew 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 {// 省略部分代码... ...// 上文提到的请求类型对应的idpublic 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 ecase e: Throwable => handleError(request, e)} finally {request.apiLocalCompleteTimeNanos = time.nanoseconds}
    }
    
  7. Processor从自己的ResponseQueue中拉取待发送的Respnose;

  8. Processor将Response发给客户端;

五. 总结

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

另外也可以在目录中找到同系列的其他文章:
Kafka源码分析系列-目录(收藏关注不迷路)
感谢阅读。

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.mzph.cn/web/1504.shtml

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

【Django】django.core.exceptions.AppRegistryNotReady: Apps aren‘t loaded yet.

其中django后台manage.py入口程序报错,检索很多问题解决方案,这里记录下个人问题原因 1.django启动异常问题详情 django.core.exceptions.AppRegistryNotReady: Apps aren’t loaded yet. 2.问题原因 Python第三方包安装版本不一致或缺少依赖包&…

Flink窗口机制

1.窗口的概念 时间是为窗口服务的。窗口是什么?为什么会有窗口呢? (1)Flink要处理的数据,一般是从Kafka过来的流式数据,如果只是单纯地统计流的数据量,是没办法统计的。 (2&#xff…

C语言程序设计:简易版的printf函数实现

简易版的printf函数实现 功能说明 (1)使用putchar函数、va可变参完成printf函数基本功能的实现; (2)函数说明: 实现对下列数据类型的输出,并返回成功输出打印的字符个数: 整数&…

在CSDN创作了6个月,我收获了什么?文末送书~

作者主页:阿玥的小东东主页! 正在学习:python和C/C 期待大家的关注哦 目录 一次很好的机会,让我开始了CSDN之旅 首先来看看我的几位领路人 创作动力 1W粉丝 在CSDN我收获了什么? 很高的展现量 认证创作者身份 社…

Java Web常见框架

1、Spring Framework Spring是一个非常强大的框架,用于构建企业级应用程序。它提供了全面的编程和配置模型。 2、Spring Boot 基于Spring,它简化了配置过程,使得启动和运行基于Spring的应用程序变得更快,更容易。 3、Hibernat…

【Linux】系统安全及应用

目录 一、账号安全基本措施 1.系统账号清理 2.密码安全控制 3.历史命令安全管理 4.限制su切换用户 1)将信任的用户加入到wheel组中 2)修改su的PAM认证配置文件 5.ssh远程登录输入三次密码错误则锁定用户 二、Linux中的PAM安全认证 1.su命令的…

Redis入门到通关之数据结构解析-动态字符串SDS

文章目录 Redis数据结构-动态字符串动态扩容举例二进制安全SDS优点与C语言中的字符串的区别 Redis数据结构-动态字符串 我们都知道 Redis 中保存的Key是字符串,value 往往是字符串或者字符串的集合。可见字符串是 Redis 中最常用的一种数据结构。 不过 Redis 没有…

Android Studio超级详细讲解下载、安装配置教程(建议收藏)

博主介绍:✌专注于前后端、机器学习、人工智能应用领域开发的优质创作者、秉着互联网精神开源贡献精神,答疑解惑、坚持优质作品共享。本人是掘金/腾讯云/阿里云等平台优质作者、擅长前后端项目开发和毕业项目实战,深受全网粉丝喜爱与支持✌有…

ES是什么?ES的使用场景有哪些?分词器??

一、ES是什么??? 1、Elasticsearch 是一个基于 Apache Lucene 构建的开源分布式搜索引擎和分析引擎。同时还可以被视为一种特殊的数据库,具体而言,它是一种分布式、面向文档的NoSQL数据库,专为全文搜索和数…

贪吃蛇游戏实现(VS编译环境)

贪吃蛇游戏 🥕个人主页:开敲🍉 🔥所属专栏:C语言🍓 🌼文章目录🌼 0. 前言 1. 游戏背景 2. 实现后游戏画面展示 3. 技术要求 4. Win32 API介绍 4.1 Win32 API 4.2 控制台程序 4.…

Java之类和对象

一面向对象的初步认知 1.什么是面向对象 Java是一门纯面向对象的语言(Object Oriented Program,简称OOP),在面向对象的世界里,一切皆为对象。面向对象是解决问题的一种思想,主要依靠对象之间的交互完成一件事情。用面向对象的思想…

嵌入式物联网实战开发笔记-乐鑫ESP32开发环境ESP-IDF搭建【doc.yotill.com】

乐鑫ESP32入门到精通项目开发参考百例下载: 链接:百度网盘 请输入提取码 提取码:4e33 3.1 ESP-IDF 简介 ESP-IDF(Espressif IoT Development Framework)是乐鑫(Espressif Systems)为 ESP 系列…

如何用JS校验HTTP和HTTPS地址

在日常开发过程中,我们有时候对某些应用功能进行封装,但是在请求接口又不能写死,这个时候我们需要对他进行多方面考虑。 如何验证请求地址是HTTP还是HTTPS 方法一: function getBaseUrl (string) {let url;try {url new URL(s…

大型网站系统架构演化实例_2.使用缓存改善网站性能

1.使用缓存改善网站性能 网站访问的特点和现实世界的财富分配一样遵循二八定律:80%的业务访问集中在20%的数据上。既然大部分业务访问集中在一小部分数据上,那么如果把这一小部分数据缓存在内存中,就可以减少数据库的访问压力&#xf…

【Python】自定义修改pip下载模块默认的安装路径

因为电脑下载了Anaconda提供的默认Python 3.9 以及后期下载的python3.10所以在Pychram进行项目开发时,发现一些库怎么导入都导入不了,手动install也是失败,后期在cmd里面发现python以及pip配置有点儿混乱,导致执行命令时&#xff…

基于SpringCloudAlibaba+Sentinel的分布式限流设计

胡弦,视频号2023年度优秀创作者,互联网大厂P8技术专家,Spring Cloud Alibaba微服务架构实战派(上下册)和RocketMQ消息中间件实战派(上下册)的作者,资深架构师,技术负责人,极客时间训练营讲师,四…

碳循环、人类、遥感之间的关联

1. 碳与碳循环 碳是自然界中很常见的一种元素,它以多种形式广泛存在于大气和地壳之中。碳单质很早就被人认识和利用,碳的一系列化合物——有机物是生命的根本。 1.1 自然界中的碳 地球上最大的两个碳库是岩石圈和化石燃料,含碳量约占…

小米K8s运维-云原生方向(面经分享)

大家好,我是秋意零。今天分享一篇小米运维面经。 小米K8s运维-云原生方向 一面 2024年4月3日 | 10点 | 一面 | 40 min 左右 1)自我介绍 2)你熟悉Python多一点吗?还熟悉其它语言吗,拿出来写过的? 3&am…

搜索引擎中的倒排索引是什么

在搜索引擎领域,倒排索引是一种核心数据结构,它让搜索引擎能够以极高的效率找到包含用户查询关键词的所有网页。为了理解倒排索引的工作原理,我们可以将其与一种更直观、生活化的例子相比较:书店里的索引卡片系统。 假设你是一位…

在RISC-V64架构的CV1811C开发板上应用perf工具进行多线程程序性能分析及火焰图调试

CV1811C环境编译 SDK目录结构 . ├── build // 编译目录,存放编译脚本以及各board差异化配置 ├── buildroot-2021.05 // buildroot开源工具 ├── freertos // freertos系统 ├── fsbl // fsbl启动固件,prebuilt形式存在…