文档章节

Apache Kafka源码剖析:第5篇 业务API处理

强子1985
 强子1985
发布于 2017/08/13 15:31
字数 433
阅读 59
收藏 1

之前说过了,请求到达业务线程池后,会被处理,但是如何被处理呢?这就是接下来要说的。

-----------------------------------------------------------------------------------------------

业务线程属于 Kafka的API层,对请求的处理通过调用KafkaAPIs中的方法实现!

1 KafkaRequestHandler

首先我们得知道这个业务线程池是怎么创建的

回到KafkaServer.scala

        requestHandlerPool = new KafkaRequestHandlerPool(config.brokerId, socketServer.requestChannel, apis, time,
          config.numIoThreads)
class KafkaRequestHandlerPool(val brokerId: Int,
                              val requestChannel: RequestChannel,
                              val apis: KafkaApis,
                              time: Time,
                              numThreads: Int) extends Logging with KafkaMetricsGroup {

  /* a meter to track the average free capacity of the request handlers */
  private val aggregateIdleMeter = newMeter("RequestHandlerAvgIdlePercent", "percent", TimeUnit.NANOSECONDS)

  this.logIdent = "[Kafka Request Handler on Broker " + brokerId + "], "
  val runnables = new Array[KafkaRequestHandler](numThreads)
  for(i <- 0 until numThreads) {//创建这么多个runnable,放到线程里执行
    runnables(i) = new KafkaRequestHandler(i, brokerId, aggregateIdleMeter, numThreads, requestChannel, apis, time)
    Utils.daemonThread("kafka-request-handler-" + i, runnables(i)).start()
  }

取出请求执行的代码

def run() {
    while (true) {
      try {
        var req : RequestChannel.Request = null
        while (req == null) {
          // We use a single meter for aggregate idle percentage for the thread pool.
          // Since meter is calculated as total_recorded_value / time_window and
          // time_window is independent of the number of threads, each recorded idle
          // time should be discounted by # threads.
          val startSelectTime = time.nanoseconds
          req = requestChannel.receiveRequest(300)
          val endTime = time.nanoseconds
          if (req != null)
            req.requestDequeueTimeNanos = endTime
          val idleTime = endTime - startSelectTime
          aggregateIdleMeter.mark(idleTime / totalHandlerThreads)
        }

        if (req eq RequestChannel.AllDone) {
          debug("Kafka request handler %d on broker %d received shut down command".format(id, brokerId))
          latch.countDown()
          return
        }
        trace("Kafka request handler %d on broker %d handling request %s".format(id, brokerId, req))
        apis.handle(req)
      } catch {
        case e: FatalExitError =>
          latch.countDown()
          Exit.exit(e.statusCode)
        case e: Throwable => error("Exception when handling request", e)
      }
    }
  }

这样就不难理解了吧,

可见,API层使用kafkaRequestHandlerPool来管理所有的KafkaRequestHandler线程,它是1个简易版的线程池,其中创建了多个KafkaRequestHandler线程。

 

KafkaApis

是Kafka服务器处理请求的入口类,负责将KafkaRequestHandler.Request分发到不同的handle*()方法里执行,见图:

因为函数太多,这里就不展开,后面碰到的时候再详细展开!

 

© 著作权归作者所有

共有 人打赏支持
强子1985

强子1985

粉丝 862
博文 929
码字总数 642915
作品 8
南京
架构师
消息中间件—简谈Kafka中的NIO网络通信模型

文章摘要:很多人喜欢把RocketMQ与Kafka做对比,其实这两款消息队列的网络通信层还是比较相似的,本文就为大家简要地介绍下Kafka的NIO网络通信模型 前面写的两篇RocketMQ源码研究笔记系列: ...

癫狂侠
07/16
0
0
架构设计:系统间通信(31)——其他消息中间件及场景应用(下1)

接上文:《架构设计:系统间通信(30)——Kafka及场景应用(中3)》 5、场景应用——电商平台:浏览记录收集功能 事件/日志收集系统是大中型软件不得不面对的话题。目前第三方业务系统对 事...

yinwenjie
2016/05/19
0
0
Java后端工程师学习大纲

之前自己总结过的Java后端工程师技能树,其涵盖的技术点比较全面,并非一朝一夕能够全部覆盖到的。对于一些还没有入门或者刚刚入门的Java后端工程师,如果一下子需要学习如此多的知识,想必很...

JackFace
2016/07/08
567
0
Heron:来自Twitter的新一代流处理引擎应用篇

【导语】 本文对比了Heron和常见的流处理项目,包括Storm、Flink、Spark Streaming和Kafka Streams,归纳了系统选型的要点。此外实践了Heron的一个案例,以及讨论了Heron在这一年开发的新特性...

dev_csdn
2017/12/26
0
0
Kafka分区分配计算(分区器Partitions)

KafkaProducer在调用send方法发送消息至broker的过程中,首先是经过拦截器Inteceptors处理,然后是经过序列化Serializer处理,之后就到了Partitions阶段,即分区分配计算阶段。在某些应用场景...

u013256816
2017/12/03
0
0

没有更多内容

加载失败,请刷新页面

加载更多

权限框架Shiro学习之表结构设计

权限框架Shiro学习之表结构设计 Shiro是一款优秀的开源安全框架,学习Shiro大家可以参考张开涛老师的博客:跟我学Shiro,当然也可参考我之前的笔记:Shiro实现身份认证、Shiro实现授权。 在学...

TyCoding
29分钟前
1
0
find命令和文件名后缀

9月18日任务 2.23/2.24/2.25 find命令 2.26 文件名后缀 which which 搜索可执行文件,必须在PATH环境变量目录中!!否则无法搜到! [root@centos7 ~]# which lsalias ls='ls --color=auto'...

robertt15
32分钟前
1
0
阿里Java程序员必备的Intellij IDEA 插件

善用Intellij插件可大幅提升我们的效率,以下是我用过不错的Intellij插件,分享给大家希望能帮到大家。 1. .ignore 生成各种ignore文件,一键创建git ignore文件的模板,免得自己去写 2. lom...

我是你大哥
42分钟前
1
0
为什么Java大神,都在看Spring Boot和Spring Cloud的书?

如果你是一名Java开发人员,并且最近正打算学习Spring Boot和Spring Cloud框架并寻找一些关于它们的最好的书籍,那么,你今天就来对地方了。 本文,我们将讨论一些学习Spring Boot和Spring ...

Java小铺
今天
12
0
springboot logback日志配置

springboot 如果不使用外部tomcat的话,日志是需要自己配置的,不然的话就只有控制台的日志,但是日志又是我们在项目上了生产环境,出问题时,检查问题的唯一途径,所以我们要配置详细的日志...

曾大大胖
今天
2
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部