文档章节

Apache Kafka源码剖析:第3篇 Acceptor&Processor细节

强子大叔的码田
 强子大叔的码田
发布于 2017/08/13 03:22
字数 1076
阅读 289
收藏 2

#程序员薪资揭榜#你做程序员几年了?月薪多少?发量还在么?>>>

这一节,主要聊Acceptor。

主要功能是:接收请求,创建socket连接,并且分配给Processor处理。

/**
 * Thread that accepts and configures new connections. There is one of these per endpoint.
 */
private[kafka] class Acceptor(val endPoint: EndPoint,
                              val sendBufferSize: Int,
                              val recvBufferSize: Int,
                              brokerId: Int,
                              processors: Array[Processor],
                              connectionQuotas: ConnectionQuotas) extends AbstractServerThread(connectionQuotas) with KafkaMetricsGroup {

  private val nioSelector = NSelector.open()//注册监听socket的Selector对象!!!
  val serverChannel = openServerSocket(endPoint.host, endPoint.port)//监听套接字!!!

  this.synchronized {//启动其管辖的Processor线程
    processors.foreach { processor =>
      Utils.newThread(s"kafka-network-thread-$brokerId-${endPoint.listenerName}-${endPoint.securityProtocol}-${processor.id}",
        processor, false).start()
    }
  }

接下来,它的run方法是Acceptor的核心逻辑,我们看看具体实现:

 /**
   * Accept loop that checks for new connection attempts
   */
  def run() {
    serverChannel.register(nioSelector, SelectionKey.OP_ACCEPT)//注册ACCEPT事件
    startupComplete()
    try {
      var currentProcessor = 0
      while (isRunning) {
        try {
          val ready = nioSelector.select(500)//最多等待500毫秒的时间,看是否有socket过来!
          if (ready > 0) {
            val keys = nioSelector.selectedKeys()
            val iter = keys.iterator()
            while (iter.hasNext && isRunning) {
              try {
                val key = iter.next
                iter.remove()
                if (key.isAcceptable)
                  accept(key, processors(currentProcessor))
                else
                  throw new IllegalStateException("Unrecognized key state for acceptor thread.")

                // round robin to the next processor thread
                currentProcessor = (currentProcessor + 1) % processors.length//看来也是采用轮询的方案
              } catch {
                case e: Throwable => error("Error while accepting connection", e)
              }
            }
          }
        }

小贴士:

我们类比下Thrift的方案

 /**
   * A round robin load balancer for choosing selector threads for new
   * connections.
   */
  protected static class SelectorThreadLoadBalancer {
    private final Collection<? extends SelectorThread> threads;
    private Iterator<? extends SelectorThread> nextThreadIterator;

    public <T extends SelectorThread> SelectorThreadLoadBalancer(Collection<T> threads) {
      if (threads.isEmpty()) {
        throw new IllegalArgumentException("At least one selector thread is required");
      }
      this.threads = Collections.unmodifiableList(new ArrayList<T>(threads));
      nextThreadIterator = this.threads.iterator();
    }

    public SelectorThread nextThread() {
      // Choose a selector thread (round robin)
      if (!nextThreadIterator.hasNext()) {
        nextThreadIterator = threads.iterator();
      }
      return nextThreadIterator.next();
    }
  }
}
一旦到了最后,就回绕到第1个

可见,殊途同归,不解释!

接下来,重点就是accept函数

  /*
   * Accept a new connection
   */
  def accept(key: SelectionKey, processor: Processor) {

为了顺利进来,我们先打个断点如下:

stop in kafka.network.SocketServer$Acceptor.accept
stop in kafka.network.SocketServer$Acceptor.run
stop at kafka.network.SocketServer:335
stop at kafka.network.SocketServer:265
/*
   * Accept a new connection
   */
  def accept(key: SelectionKey, processor: Processor) {
    val serverSocketChannel = key.channel().asInstanceOf[ServerSocketChannel]
    val socketChannel = serverSocketChannel.accept()//调用accept函数获取 socket句柄
    try {
      connectionQuotas.inc(socketChannel.socket().getInetAddress)
      socketChannel.configureBlocking(false)
      socketChannel.socket().setTcpNoDelay(true)
      socketChannel.socket().setKeepAlive(true)
      if (sendBufferSize != Selectable.USE_DEFAULT_BUFFER_SIZE)
        socketChannel.socket().setSendBufferSize(sendBufferSize)

      debug("Accepted connection from %s on %s and assigned it to processor %d, sendBufferSize [actual|requested]: [%d|%d] recvBufferSize [actual|requested]: [%d|%d]"
            .format(socketChannel.socket.getRemoteSocketAddress, socketChannel.socket.getLocalSocketAddress, processor.id,
                  socketChannel.socket.getSendBufferSize, sendBufferSize,
                  socketChannel.socket.getReceiveBufferSize, recvBufferSize))

      processor.accept(socketChannel)//交给Processor处理,这个已经是通过轮询选中的
    } catch {
      case e: TooManyConnectionsException =>
        info("Rejected connection from %s, address already has the configured maximum of %d connections.".format(e.ip, e.count))
        close(socketChannel)
    }
  }

我们看Processor怎么处理的

/**
   * Queue up a new connection for reading
   */
  def accept(socketChannel: SocketChannel) {
    newConnections.add(socketChannel)
    wakeup()
  }

这个newConnections是个什么?

private val newConnections = new ConcurrentLinkedQueue[SocketChannel]()

是1个队列,恩,类比下Thrift怎么玩的

看到了吧,套路都一样。。。

那么, 这个新的连接是怎么被Processor处理的呢?

看代码

奥秘就在这里,我们再看看Thrift的玩法

真的没啥可说的,就这么回事吧

好,回到Kafka,我们知道Processor主要完成读取请求和写回响应。

Processor不参与具体的业务逻辑操作。

 

通过acceptor.accept创建的socket,通过processor.accept传给processor处理,

/**
   * Register any new connections that have been queued up
   */
  private def configureNewConnections() {
    while (!newConnections.isEmpty) {
      val channel = newConnections.poll()
      try {
        debug(s"Processor $id listening to new connection from ${channel.socket.getRemoteSocketAddress}")
        val localHost = channel.socket().getLocalAddress.getHostAddress
        val localPort = channel.socket().getLocalPort
        val remoteHost = channel.socket().getInetAddress.getHostAddress
        val remotePort = channel.socket().getPort
        val connectionId = ConnectionId(localHost, localPort, remoteHost, remotePort).toString
        selector.register(connectionId, channel)//注册读事件
      } catch {
        // We explicitly catch all non fatal exceptions and close the socket to avoid a socket leak. The other
        // throwables will be caught in processor and logged as uncaught exceptions.
        case NonFatal(e) =>
          val remoteAddress = channel.getRemoteAddress
          // need to close the channel here to avoid a socket leak.
          close(channel)
          error(s"Processor $id closed connection from $remoteAddress", e)
      }
    }
  }

到这里,就注册了读事件,然后看Processor怎么处理读事件的!

 private def processCompletedReceives() {
    selector.completedReceives.asScala.foreach { receive =>
      try {
        val openChannel = selector.channel(receive.source)
        // Only methods that are safe to call on a disconnected channel should be invoked on 'openOrClosingChannel'.
        val openOrClosingChannel = if (openChannel != null) openChannel else selector.closingChannel(receive.source)
        val session = RequestChannel.Session(new KafkaPrincipal(KafkaPrincipal.USER_TYPE, openOrClosingChannel.principal.getName), openOrClosingChannel.socketAddress)

        val req = RequestChannel.Request(processor = id, connectionId = receive.source, session = session,
          buffer = receive.payload, startTimeNanos = time.nanoseconds,
          listenerName = listenerName, securityProtocol = securityProtocol)
        requestChannel.sendRequest(req)//发给业务线程池,是通过requestChannel
        selector.mute(receive.source)
      } catch {
        case e @ (_: InvalidRequestException | _: SchemaException) =>
          // note that even though we got an exception, we can assume that receive.source is valid. Issues with constructing a valid receive object were handled earlier
          error(s"Closing socket for ${receive.source} because of error", e)
          close(selector, receive.source)
      }
    }
  }
  /** Send a request to be handled, potentially blocking until there is room in the queue for the request */
  def sendRequest(request: RequestChannel.Request) {
    requestQueue.put(request)
  }

可见,把请求放入了队列,跟Thrift一模一样的

接下来,看这个队列如何被业务线程获取拿任务处理的!

在此之前,有1个细节

这个跟Thrift完全是一模一样啊

手法如出一辙。

回到kafka的代码,既然请求已经放到一个队列里了,那么就看业务线程如何处理了,下一节讲这个

 

© 著作权归作者所有

强子大叔的码田

强子大叔的码田

粉丝 923
博文 1630
码字总数 1283214
作品 9
南京
架构师
私信 提问
加载中

评论(0)

源码圈 365 胖友的书单整理

🙂🙂🙂关注微信公众号:【芋道源码】有福利: RocketMQ / MyCAT / Sharding-JDBC 所有源码分析文章列表 RocketMQ / MyCAT / Sharding-JDBC 中文注释源码 GitHub 地址 您对于源码的疑问...

芋道源码掘金Java群217878901
2017/09/21
0
0
精尽 Kafka 学习指南【优秀学习指南汇总】

视频 炼石成金 《中间件之 Kafka》 一共有 19P 。概念部分讲的蛮细的。 尚硅谷 《大数据视频Kafka视频教程》 一共 24P 。讲的还不错的。 书生小四 《Kafka 流处理平台》 一共 1 小时 16 分钟...

osc_9g4x7r4s
2019/02/23
12
0
王家林大咖清华大学新书Spark第二版已上市:前浪致 Spark + AI 后浪

王家林大咖清华大学新书Spark第二版已上市:前浪 致 Spark + AI 后浪 大咖心声 新书图片 新书介绍 编辑推荐 内容简介 作者简介 新书目录 第二版前言 第一版前言 新书案例讲解 第二版网购链接...

段智华
昨天
0
0
Apache Beam实战指南 | 手把手教你玩转KafkaIO与Flink

https://mp.weixin.qq.com/s?biz=MzU1NDA4NjU2MA==&mid=2247492538&idx=2&sn=9a2bd9fe2d7fd681c10ebd368ef81c9c&chksm=fbea5a75cc9dd3636c148ebe6e296621d0c07132938a62f0b3643f34af414b3fd8......

osc_azsn5lm2
2018/09/05
22
0
阿里大牛力荐6篇实战文档:JVM+多线程+Kafka+Redis+Nginx+MySQL

深入理解Java虚拟机:JVM高级特性与最佳实践 第一部分 走近Java 第1章 走近Java 第二部分 自动内存管理机制 第2章 Java内存区域与内存溢出异常 第3章 垃圾收集器与内存分配策略 第4章 虚拟机...

Java微服务
2019/07/16
0
0

没有更多内容

加载失败,请刷新页面

加载更多

微服务项目搭建,到底要不要聚合工程?

这是一个入门问题,做微服务项目,首先就是要搭建 Project,代码采用什么样的形式来组织,这是我们面临的第一个问题。 要扯清楚这个问题,首先对 Maven 的使用不能含糊,小伙伴们可以在公众号...

osc_y12wmf09
22分钟前
15
0
C语言探索之旅 | 第一部分第六课:变量的世界(三),显示变量内容

作者 谢恩铭,慕课网精英讲师 Oscar老师。 转载请注明出处。 内容简介 用 printf 显示变量内容 用 scanf 提取程序中的输入 总结 第一部分第七课预告 1. 用 printf 显示变量内容 变量相关的内...

osc_zjs1puzi
23分钟前
16
0
php 操作RabbitMQ

基本流程图 如果exchange 没有绑定queue,则消息将会被丢弃 如果创建exchange,queue,并且已经绑定了,则可以直接使用 为了防止脚本出问题 可以配合supervisor 安装 从网站 https://packag...

php开源社区
24分钟前
18
0
Kotlin Coroutines Flow 系列(五) 其他的操作符

八. Flow 其他的操作符 8.1 Transform operators transform 在使用 transform 操作符时,可以任意多次调用 emit ,这是 transform 跟 map 最大的区别: fun main() = runBlocking { (1...

osc_0l1onu3j
25分钟前
13
0
挽救数据库性能的30条黄金法则

原文: 挽救数据库性能的30条黄金法则 1. 优化查询,应尽量避免全表扫描,应该在用于检索数据和排序数据的字段上建立索引,如where子句用于搜索,order by子句用于排序,所以在这两个子句涉及...

osc_gxvh47u5
25分钟前
16
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部