文档章节

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

强子1985
 强子1985
发布于 2017/08/13 03:22
字数 1076
阅读 46
收藏 0

这一节,主要聊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的代码,既然请求已经放到一个队列里了,那么就看业务线程如何处理了,下一节讲这个

 

© 著作权归作者所有

共有 人打赏支持
强子1985

强子1985

粉丝 875
博文 1091
码字总数 799514
作品 8
南京
架构师
私信 提问
kafka系列文章索引(结束)

apache kafka在数据处理中特别是日志和消息的处理上会有很多出色的表现,这里写个索引,关于kafka的文章暂时就更新到这里,最近利用空闲时间在对 kafka做一些功能性增强,并java化,虽然现在...

老先生二号
2017/05/28
0
0
源码圈 365 胖友的书单整理

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

芋道源码掘金Java群217878901
2017/09/21
0
0
【写书评赠书活动】《Unix内核源码剖析》

赠书活动不能停!本次图灵教育携手开源中国一起赠书啦! 本期活动的赠书是超级计算机“京”的L1缓存设计者青柳隆宏经典著作《Unix内核源码剖析》! 图书介绍 书名:《Unix内核源码剖析》 书号...

图灵教育
2014/03/19
6.2K
52
apache kafka技术分享系列(目录索引)

目录索引: Kafka使用场景 1.为何使用消息系统 2.我们为何需要搭建ApacheKafka分布式系统 3.消息队列中点对点与发布订阅区别 kafka开发与管理: 1)apachekafka消息服务 2)kafak安装与使用 ...

dannyhe
2015/09/06
453
1
Java后端工程师学习大纲

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

JackFace
2016/07/08
567
0

没有更多内容

加载失败,请刷新页面

加载更多

eslint rules 规则

'rules': { "comma-dangle": ["error", "never"], //是否允许对象中出现结尾逗号 "no-cond-assign": 2, //条件语句的条件中不允许出现赋值运算符 "no-console": 2, //不允许出现console语句 ...

agenyun
45分钟前
1
0
类型判断时instanceof和equals的不同用法

接口设计时为了避免序列化的麻烦,将接口定义为参数为map<String,String>类型的接口,但是现在调用时需要转换当前的实体Bean为Map,接口接收方再把Map转换为另一个Bean实体。过程中的需要对类...

wangtx
51分钟前
3
0
vue 组件间传值(个人精编)

1.父组件向子组件传值 1⃣️.子组件标签绑定需要传递的参数名2⃣️.子组件页面使用props 接收参数 2.子组件向父组件传值  1⃣️.子组件使用$emit来触发一个自定义事件,并传递一个参...

MrBoyce
今天
1
0
(荷兰)彼得·冯·门施著:博物馆学研究的目的

博物馆学研究的目的 (荷)彼得·冯·门施 尽管诸多关于博物馆学认知目的的不同看法可以被归纳为数个主要群体,但没有一个群体可以被称为“学派”。一般来说,学派是由于博物馆学研究目的的不...

乔老哥
今天
3
0
Vue slot的用法

之前看官方文档,由于自己理解的偏差,不知道slot是干嘛的,看到小标题,使用Slot分发内容,就以为 是要往下派发内容。然后就没有理解插槽的概念。其实说白了,使用slot就是先圈一块地,将来...

peakedness丶
今天
2
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部