Kafka-服务端-网络层-源码流程
整体架构如下所示:
responseQueue不在RequestChannel中,在Processor中,每个Processor内部有一个responseQueue
- 客户端发送的请求被Acceptor转发给Processor处理
- 处理器将请求放到RequestChannel的requestQueue中
- KafkaRequestHandler取出requestQueue中的请求
- 调用KafkaApis进行业务逻辑处理
- KafkaApis将响应结果放到对应的Processor的responseQueue中
- processor从responseQueue中取出响应结果
- processor将响应结果返回给客户端
KafkaServer是Kafka服务端的主类,KafkaServer中和网络成相关的服务组件包括SocketServer、KafkaApis和KafkaRequestHandlerPool。SocketServer主要关注网络层的通信协议,具体的业务处理逻辑则交给KafkaRequestHandler和KafkaApis来完成。
class KafkaServer(val config: KafkaConfig) {def startup() {socketServer = new SocketServer(config, metrics, time, credentialProvider)socketServer.startup(startupProcessors = false)/* start processing requests */apis = new KafkaApis(socketServer.requestChannel, ...)requestHandlerPool = new KafkaRequestHandlerPool(config.brokerId, ...)}}
SocketServer
def startup(startupProcessors: Boolean = true) {this.synchronized {...createAcceptorAndProcessors(config.numNetworkThreads, config.listeners)if (startupProcessors) {startProcessors()}}private def createAcceptorAndProcessors(processorsPerListener: Int,endpoints: Seq[EndPoint]): Unit = synchronized {...endpoints.foreach { endpoint =>...val acceptor = new Acceptor(endpoint, ...)addProcessors(acceptor, endpoint, processorsPerListener)KafkaThread.nonDaemon(s"kafka-socket-acceptor-$listenerName-$securityProtocol-${endpoint.port}", acceptor).start()acceptor.awaitStartup()acceptors.put(endpoint, acceptor)}}
可以看出SocketServer.startup()中会根据listener的个数创建相同个数的acceptor,每个acceptor关联数个processor。这是一种典型的Reactor模式,acceptor负责与客户端建立连接,并将连接分发给processor,processor负责所分连接后续的所有读写交互。
Acceptor
def run() {serverChannel.register(nioSelector, SelectionKey.OP_ACCEPT)startupComplete()try {var currentProcessor = 0while (isRunning) {try {val ready = nioSelector.select(500)if (ready > 0) {val keys = nioSelector.selectedKeys()val iter = keys.iterator()while (iter.hasNext && isRunning) {try {val key = iter.nextiter.remove()if (key.isAcceptable) {val processor = synchronized {currentProcessor = currentProcessor % processors.sizeprocessors(currentProcessor)}accept(key, processor)} elsethrow new IllegalStateException("Unrecognized key state for acceptor thread.")// round robin to the next processor thread, mod(numProcessors) will be done latercurrentProcessor = currentProcessor + 1} catch {case e: Throwable => error("Error while accepting connection", e)}}}}catch {// We catch all the throwables to prevent the acceptor thread from exiting on exceptions due// to a select operation on a specific channel or a bad request. We don't want// the broker to stop responding to requests from other clients in these scenarios.case e: ControlThrowable => throw ecase e: Throwable => error("Error occurred", e)}}} finally {debug("Closing server socket and selector.")CoreUtils.swallow(serverChannel.close(), this, Level.ERROR)CoreUtils.swallow(nioSelector.close(), this, Level.ERROR)shutdownComplete()}}
上面是Acceptor的run()方法,可以看出,Acceptor在通道上注册了SelectionKey.OP_ACCEPT事件(OP_READ、OP_WRITE、OP_CONNECT、OP_ACCEPT,客户端监听OP_CONNECT事件,负责发起连接,服务端监听OP_CONNECT事件,负责建立连接),负责与客户端建立连接。并将建立的连接通过轮询的方式指派给processor。
Processor
每个Processor都会分到数个与客户端的连接。Processor的处理逻辑如下所示:
override def run() {startupComplete()try {while (isRunning) {try {// 在新分到的客户端连接上注册OP_READ事件configureNewConnections()// 从responseQueue中取响应,赋值给KafkaChannel的send,等待poll时发送processNewResponses()// selector轮询各种事件,读取请求或者发送响应poll()// 封装selector.completedReceives中的请求,放入requestQueueprocessCompletedReceives()// 处理selector.completedSends响应(移除inflightResponses中的记录;执行响应的回调函数)processCompletedSends()processDisconnected()} catch {...}}} finally {...}}
Processor线程的名字中有kafka-network字样,可以通过jstack -l pid | grep kafka-network进行筛选。
KafkaRequestHandlerPool
KafkaServer会创建请求处理线程池(KafkaRequestHandlerPool),在请求处理线程池中会创建并启动多个请求处理线程(KafkaRequestHandler)。KafkaRequestHandler会获取RequestChannel.requestQueue中的请求进行处理,在内部实际处理会交给KafkaApis完成。
class KafkaRequestHandlerPool(val brokerId: Int, ...) {...for (i <- 0 until numThreads) {createHandler(i)}def createHandler(id: Int): Unit = synchronized {runnables += new KafkaRequestHandler(..., requestChannel, apis, time)KafkaThread.daemon("kafka-request-handler-" + id, runnables(id)).start()}
}
KafkaRequestHandler的run()方法如下:
class KafkaRequestHandler(id: Int,...) extends Runnable with Logging {...def run() {while (!stopped) {val req = requestChannel.receiveRequest(300)req match {case RequestChannel.ShutdownRequest =>shutdownComplete.countDown()returncase request: RequestChannel.Request =>try {request.requestDequeueTimeNanos = endTimeapis.handle(request)} catch {case e: FatalExitError =>shutdownComplete.countDown()Exit.exit(e.statusCode)case e: Throwable => error("Exception when handling request", e)} finally {request.releaseBuffer()}case null => // continue}}shutdownComplete.countDown()}}
相关文章:
Kafka-服务端-网络层-源码流程
整体架构如下所示: responseQueue不在RequestChannel中,在Processor中,每个Processor内部有一个responseQueue 客户端发送的请求被Acceptor转发给Processor处理处理器将请求放到RequestChannel的requestQueue中KafkaRequestHandler取出reque…...
百日筑基第十一天-看看SpringBoot
百日筑基第十一天-看看SpringBoot 创建项目 Spring 官方提供了 Spring Initializr 的方式来创建 Spring Boot 项目。网址如下: https://start.spring.io/ 打开后的界面如下: 可以将 Spring Initializr 看作是 Spring Boot 项目的初始化向导ÿ…...
Generative Modeling by Estimating Gradients of the Data Distribution
Generative Modeling by Estimating Gradients of the Data Distribution 本文介绍宋飏提出的带噪声扰动的基于得分的生成模型。首先介绍基本的基于得分的生成模型的训练方法(得分匹配)和采样方法(朗之万动力学)。然后基于流形假…...
vector与list的简单介绍
1. 标准库中的vector类的介绍: vector是表示大小可以变化的数组的序列容器。 就像数组一样,vector对其元素使用连续的存储位置,这意味着也可以使用指向其元素的常规指针上的偏移量来访问其元素,并且与数组中的元素一样高效。但与数…...
四种线程池的使用,优缺点分析
池化思想:线程池、字符串常量池、数据库连接池 提高资源的利用率 下面是手动创建线程和执行任务过程,可见挺麻烦的,而且线程利用率不高。 手动创建线程对象执行任务执行完毕,释放线程对象 线程池的优点: 提高线程的…...
什么是 BEM 规范
BEM(Block, Element, Modifier)是一种 CSS 命名规范,旨在提高代码的可读性和可维护性。BEM 规范通过明确的命名规则来定义组件和组件的各个部分,使开发者能够更容易地理解和维护代码。 BEM 命名规范的基本概念 Block(…...
【Node.JS】入门
文章目录 Node.js的入门涉及对其基本概念、特点、安装、以及基本使用方法的了解。以下是对Node.js入门的详细介绍: 一、Node.js基本概念和特点 定义:Node.js是一个基于Chrome V8引擎的JavaScript运行环境,它使得JavaScript能够运行在服务器…...
Amazon SageMaker 机器学习之旅的助推器
一、前言 在当今的数字化时代,人工智能和机器学习已经成为推动社会进步的重要引擎。亚马逊云科技在 2023 re:Invent 全球大会上,宣布推出五项 Amazon SageMaker 新功能: Amazon SageMaker HyperPod 通过为大规模分布式训练提供专用的基础架构…...
TransMIL:基于Transformer的多实例学习
MIL是弱监督分类问题的有力工具。然而,目前的MIL方法通常基于iid假设,忽略了不同实例之间的相关性。为了解决这个问题,作者提出了一个新的框架,称为相关性MIL,并提供了收敛性的证明。基于此框架,还设计了一…...
3.用户程序与驱动交互
驱动程序请使用第二章https://blog.csdn.net/chenhequanlalala/article/details/140034424 用户app与驱动交互最常见的做法是insmod驱动后,生成一个设备节点,app通过open,read等系统调用去操作这个设备节点,这里先用mknode命令调…...
尽量不写一行if...elseif...写出高质量可持续迭代的项目代码
背景 无论是前端代码还是后端代码,都存在着定位困难,不好抽离,改造困难的问题,造成代码开发越来越慢,此外因为代码耦合较高,总是出现改了一处地方,然后影响其他地方,要么就是要修改…...
xcrun: error: unable to find utility “simctl“, not a developer tool or in PATH
目录 前言 一、问题详情 二、解决方案 1.确认Xcode已安装 2.安装Xcode命令行工具 3.指定正确的开发者目录 4. 确认命令行工具路径 5. 更新PATH环境变量 前言 今天使用cocoapods更新私有库的时候,遇到了"xcrun: error: unable to find utility &…...
【linux高级IO(一)】理解五种IO模型
💓博主CSDN主页:杭电码农-NEO💓 ⏩专栏分类:Linux从入门到精通⏪ 🚚代码仓库:NEO的学习日记🚚 🌹关注我🫵带你学更多操作系统知识 🔝🔝 Linux高级IO 1. 前言2. 重谈对…...
前端引用vue/element/echarts资源等引用方法Blob下载HTML
前端引用下载vue/element/echarts资源等引用方法 功能需求 需求是在HTML页面中集成Vue.js、Element Plus(Element UI的Vue 3版本)、ECharts等前端资源,使用Blob下载HTML。 解决方案概述 直接访问线上CDN地址:简单直接,…...
昇思MindSpore学习笔记2-01 LLM原理和实践 --基于 MindSpore 实现 BERT 对话情绪识别
摘要: 通过识别BERT对话情绪状态的实例,展现在昇思MindSpore AI框架中大语言模型的原理和实际使用方法、步骤。 一、环境配置 %%capture captured_output # 实验环境已经预装了mindspore2.2.14,如需更换mindspore版本,可更改下…...
uniapp实现图片懒加载 封装组件
想要的效果就是窗口滑动到哪里,哪里的图片进行展示 主要原理使用IntersectionObserver <template><view><image error"HandlerError" :style"imgStyle" :src"imageSrc" :id"randomId" :mode"mode&quo…...
持续交付:自动化测试与发布流程的变革
目录 前言1. 持续交付的概念1.1 持续交付的定义1.2 持续交付的核心原则 2. 持续交付的优势2.1 提高交付速度2.2 提高软件质量2.3 降低发布风险2.4 提高团队协作 3. 实施持续交付的步骤3.1 构建自动化测试体系3.1.1 单元测试3.1.2 集成测试3.1.3 功能测试3.1.4 性能测试 3.2 构建…...
VBA常用的字符串内置函数
前言 在VBA程序中,常用的内置函数可以按照功能分为字符串函数、数字函数、转换函数等等,本节主要会介绍常用的字符串的内置函数,包括Len()、Left()、Mid()、Right()、Split()、String()、StrConV()等。 本节的练习数据表以下表为例ÿ…...
大数据面试题之Spark(7)
目录 Spark实现wordcount Spark Streaming怎么实现数据持久化保存? Spark SQL读取文件,内存不够使用,如何处理? Spark的lazy体现在哪里? Spark中的并行度等于什么 Spark运行时并行度的设署 Spark SQL的数据倾斜 Spark的exactly-once Spark的…...
AI绘画 Stable Diffusion图像的脸部细节控制——采样器全解析
大家好,我是画画的小强 我们在运用AI绘画 Stable Diffusion 这一功能强大的AI绘图工具时,我们往往会发现自己对提示词的使用还不够充分。在这种情形下,我们应当如何调整自己的策略,以便更加精确、全面地塑造出理想的人物形象呢&a…...
liunx离线安装Firefox
在Linux系统中离线安装Firefox浏览器,您需要先从Mozilla的官方网站下载Firefox的安装包,然后通过终端进行安装。以下是详细的步骤: 准备工作 下载Firefox安装包: 首先,在一台可以上网的电脑上访问Firefox官方下载页面…...
UNet进行病理图像分割
数据集链接:https://pan.baidu.com/s/1IBe_P0AyHgZC39NqzOxZhA?pwdnztc 提取码:nztc UNet模型 import torch import torch.nn as nnclass conv_block(nn.Module):def __init__(self, ch_in, ch_out):super(conv_block, self).__init__()self.conv nn…...
初二数学基础差从哪开始补?附深度解析!
有时候,当你推不开一扇门的时候,不要着急,试着反方向拉一下,或者横向拉一下。下面是小偏整理的初二数学基础差从哪开始补2021年,感谢您的每一次阅读。 初二数学基础差从哪开始补2021年 第一个问题是很多同学都…...
【C语言】return 关键字
在C语言中,return是一个关键字,用于从函数中返回值或者结束函数的执行。它是函数的重要组成部分,负责将函数的计算结果返回给调用者,并可以提前终止函数的执行。 主要用途和原理: 返回值给调用者: 当函数执…...
华为机试HJ13句子逆序
华为机试HJ13句子逆序 题目: 将一个英文语句以单词为单位逆序排放。例如“I am a boy”,逆序排放后为“boy a am I”所有单词之间用一个空格隔开,语句中除了英文字母外,不再包含其他字符 想法: 将输入的字符串通过…...
代码随想录day40 动态规划(5)
52. 携带研究材料(第七期模拟笔试) (kamacoder.com) 完全背包,可重复放入物品,需要用一维滚动数组从前往后遍历。 由于第0个物品和后面物品的转移方程没有区别,可以不额外初始化dp数组,直接用元素全0的d…...
FFmpeg 命令行 音视频格式转换
📚:FFmpeg 提供了丰富的命令行选项和功能,可以用来处理音视频文件、流媒体等,掌握命令行的使用,可以有效提高工作效率。 目录 一、视频转换和格式转换 🔵 将视频文件转换为另一种格式 🔵 指定…...
Jmeter使用JSON Extractor提取多个变量
1.当正则不好使时,用json extractor 2.提取多个值时,默认值必填,否则读不到变量...
c++ 设计模式 的课本范例(下)
(19) 桥接模式 Bridge,不是采用类继承,而是采用类组合,一个类的数据成员是类对象,来扩展类的功能。源码如下: class OS // 操作系统负责绘图 { public:virtual ~OS() {}virtual void draw(cha…...
结合数据索引结构看SQL的真实执行过程
引言 关于数据库设计与优化的前几篇文章中,我们提到了数据库设计优化应该遵守的指导原则、数据库底层的索引组织结构、数据库的核心功能组件以及SQL的解析、编译等。这些其实都是在为SQL的优化、执行的理解打基础。 今天这篇文章,我们以MySQL中InnoDB存…...
spark shuffle——shuffle管理
ShuffleManager shuffle系统的入口。ShuffleManager在driver和executor中的sparkEnv中创建。在driver中注册shuffle,在executor中读取和写入数据。 registerShuffle:注册shuffle,返回shuffleHandle unregisterShuffle:移除shuff…...
HTMLCSS(入门)
HTML <html> <head><title>第一个页面</title></head><body>键盘敲烂,工资过万</body> </html> <!DOCTYPE>文档类型声明,告诉浏览器使用哪种HTML版本显示网页 <!DOCTYPE html>当前页面采取…...
富格林:曝光可信策略制止亏损
富格林指出,相信大家都对黄金投资的价值空间有目共睹,现如今黄金市场波动频繁,因此不少投资者也开始加入该市场试图赢得额外的财富。但作为新手投资者贸贸然地进场操作,亏损的几率是很大的,因此要学会掌握正规平台曝光…...
Android --- Service
出自于此,写得很清楚。关于Android Service真正的完全详解,你需要知道的一切_android service-CSDN博客 出自【zejian的博客】 什么是Service? Service(服务)是一个一种可以在后台执行长时间运行操作而没有用户界面的应用组件。 服务可由其他应用组件…...
Vue3从入门到精通(三)
vue3插槽Slots 在 Vue3 中,插槽(Slots)的使用方式与 Vue2 中基本相同,但有一些细微的差异。以下是在 Vue3 中使用插槽的示例: // ChildComponent.vue <template><div><h2>Child Component</h2&…...
【FreeRTOS】同步与互斥通信-有缺陷的互斥案例
目录 同步与互斥通信同步与互斥的概念同步与互斥并不简单缺陷分析汇编指令优化过程 - 关闭中断时间轴分析 思考时刻 参考《FreeRTOS入门与工程实践(基于DshanMCU-103).pdf》 同步与互斥通信 同步与互斥的概念 一句话理解同步与互斥:我等你用完厕所,我再…...
Docker 安装 Python
Docker 安装 Python 在当今的软件开发领域,Docker 已成为一项关键技术,它允许开发人员将应用程序及其依赖环境打包到一个可移植的容器中。Python,作为一种广泛使用的高级编程语言,经常被部署在 Docker 容器中。本文将详细介绍如何在 Docker 中安装 Python,以及如何配置环…...
外泌体相关基因肝癌临床模型预测——2-3分纯生信文章复现——4.预后相关外泌体基因确定单因素cox回归(2)
内容如下: 1.外泌体和肝癌TCGA数据下载 2.数据格式整理 3.差异表达基因筛选 4.预后相关外泌体基因确定 5.拷贝数变异及突变图谱 6.外泌体基因功能注释 7.LASSO回归筛选外泌体预后模型 8.预后模型验证 9.预后模型鲁棒性分析 10.独立预后因素分析及与临床的…...
C++: Map数组的遍历
在C中,map是一个关联容器,它存储的元素是键值对(key-value pairs),其中每个键都是唯一的,并且自动根据键来排序。遍历map的方式有几种,但最常用的两种是使用迭代器(iterator…...
【Windows】Bootstrap Studio(网页设计)软件介绍及安装步骤
软件介绍 Bootstrap Studio 是一款专为前端开发者设计的强大工具,主要用于快速创建现代化的响应式网页和网站。以下是它的主要特点和功能: 直观的界面设计 Bootstrap Studio 提供了直观的用户界面,使用户能够轻松拖放元素来构建网页。界面…...
二维舵机颜色追踪,使用树莓派+opencv+usb摄像头+两个舵机实现颜色追踪,采用pid调控
效果演示 二维云台颜色追踪 使用树莓派opencvusb摄像头两个舵机实现颜色追踪,采用pid调控 import cv2 import time import numpy as np from threading import Thread from servo import Servo from pid import PID# 初始化伺服电机 pan Servo(pin19) tilt Serv…...
c进阶篇(四):内存函数
内存函数以字节为单位更改 1.memcpy memcpy 是 C/C 中的一个标准库函数,用于内存拷贝操作。它的原型通常定义在 <cstring> 头文件中,其作用是将一块内存中的数据复制到另一块内存中。 函数原型:void *memcpy(void *dest, const void…...
新手入门:无服务器函数和FaaS简介
无服务器(Serverless)架构的价值在于其成本效益、弹性和扩展性、简化的开发和部署流程、高可用性和可靠性以及使开发者能够专注于业务逻辑。通过自动化资源调配和按需计费,无服务器架构能够降低成本并适应流量变化,同时简化开发流…...
基于Transformer的端到端的目标检测 | 读论文
本文正在参加 人工智能创作者扶持计划 提及到计算机视觉的目标检测,我们一般会最先想到卷积神经网络(CNN),因为这算是目标检测领域的开山之作了,在很长的一段时间里人们都折服于卷积神经网络在图像处理领域的优势&…...
6.8应用进程跨网络通信
《计算机网络》第7版,谢希仁 理解socket通信...
redis布隆过滤器原理及应用场景
目录 原理 应用场景 优点 缺点 布隆过滤器(Bloom Filter)是一种空间效率很高的随机数据结构,它利用位数组和哈希函数来判断一个元素是否存在于集合中。 原理 数据结构: 位数组:一个由0和1组成的数组,初始…...
vue+openlayers之几何图形交互绘制基础与实践
文章目录 1.实现效果2.实现步骤3.示例页面代码3.基本几何图形绘制的关键代码 1.实现效果 绘制点、线、多边形、圆、正方形、长方形 2.实现步骤 引用openlayers开发库。加载天地图wmts瓦片地图。在页面上添加几何图形绘制的功能按钮,使用下拉列表(sel…...
「多模态大模型」解读 | 突破单一文本模态局限
编者按:理想状况下,世界上的万事万物都能以文字的形式呈现,如此一来,我们似乎仅凭大语言模型(LLMs)就能完成所有任务。然而,理想很丰满,现实很骨感——数据形态远不止文字一种&#…...
Redis深度解析:核心数据类型与键操作全攻略
文章目录 前言redis数据类型string1. 设置单个字符串数据2.设置多个字符串类型的数据3.字符串拼接值4.根据键获取字符串的值5.根据多个键获取多个值6.自增自减7.获取字符串的长度8.比特流操作key操作a.查找键b.设置键值的过期时间c.查看键的有效期d.设置key的有效期e.判断键是否…...
C语言 指针和数组——指针的算术运算
目录 指针的算术运算 指针加上一个整数 指针减去一个整数 指针相减 指针的关系比较运算 小结 指针的算术运算 指针加上一个整数 指针减去一个整数 指针相减 指针的关系比较运算 小结 指针变量 – 指针类型的变量,保存地址型数据 指针变量与其他类型…...