这个类上的注释阐述了kafka server的io线程模型

/**
 * An NIO socket server. The threading model is
 *   1 Acceptor thread that handles new connections
 *   Acceptor has N Processor threads that each have their own selector and read requests from sockets
 *   M Handler threads that handle requests and produce responses back to the processor threads for writing.
 */

一共三种线程。一个Acceptor线程负责处理新连接请求，会有N个Processor线程，每个都有自己的Selector，负责从socket中读取请求和将返回结果写回。然后会有M个Handler线程，负责处理请求，并且将结果返回给Processor。
将Acceptor和Processor线程分开的目的是为了避免读写频繁影响新连接的接收。

SocketServer初始化

SockerServer创建的时候通过server.properties和默认的配置中获取配置，如numNetworkThread(num.network.threads，也就是线程模型中的N)、

创建processor数组、acceptorMap(因为可能会在多个Endpoint接收请求)、memoryPool(SimpleMemoryPool里主要做的事情是统计监控ByteBuffer的使用)、requestChanne等 。

private val endpoints = config.listeners.map(l => l.listenerName -> l).toMap
private val numProcessorThreads = config.numNetworkThreads
private val maxQueuedRequests = config.queuedMaxRequests
private val totalProcessorThreads = numProcessorThreads * endpoints.size
private val maxConnectionsPerIp = config.maxConnectionsPerIp
private val maxConnectionsPerIpOverrides = config.maxConnectionsPerIpOverrides
this.logIdent = "[Socket Server on Broker " + config.brokerId + "], "
private val memoryPoolSensor = metrics.sensor("MemoryPoolUtilization")
private val memoryPoolDepletedPercentMetricName = metrics.metricName("MemoryPoolAvgDepletedPercent", "socket-server-metrics")
memoryPoolSensor.add(memoryPoolDepletedPercentMetricName, new Rate(TimeUnit.MILLISECONDS))
private val memoryPool = if (config.queuedMaxBytes > 0) new SimpleMemoryPool(config.queuedMaxBytes, config.socketRequestMaxBytes, false, memoryPoolSensor) else MemoryPool.NONE
val requestChannel = new RequestChannel(totalProcessorThreads, maxQueuedRequests)
private val processors = new Array[Processor](totalProcessorThreads)
private[network] val acceptors = mutable.Map[EndPoint, Acceptor]()
private var connectionQuotas: ConnectionQuotas = _

RequestChannel

因为Nio带来的异步特性，就是在一个连接上可以连续发送多个应用层的请求，每个请求得到是一个返回的Future。RequestChannel中将请求和返回结果放在各自的BlockingQueue中，也就是requestQueue和responseQueue，这里的request指客户端发来的请求。requestQueue的大小是queued.max.requests定义的，默认500。而每个RequestChannel中有numProcessor大小个responseQueue（无界的LinkedBlockingQueue)。
这样Handler从requestQueue中取request处理得到response然后put到responseQueue中。Processor则把接收到的byte转换成requestput到requestQueue中，并从responseQueue中拉response写回给对应的socket。

startup

startup中创建Processor、Acceptor。创建connectionQuotas, 限制每个客户端ip的最大连接数。

  val sendBufferSize = config.socketSendBufferBytes
  val recvBufferSize = config.socketReceiveBufferBytes
  val brokerId = config.brokerId
  var processorBeginIndex = 0
  config.listeners.foreach { endpoint =>
    val listenerName = endpoint.listenerName
    val securityProtocol = endpoint.securityProtocol
    val processorEndIndex = processorBeginIndex + numProcessorThreads
    for (i <- processorBeginIndex until processorEndIndex)
      processors(i) = newProcessor(i, connectionQuotas, listenerName, securityProtocol, memoryPool)
    val acceptor = new Acceptor(endpoint, sendBufferSize, recvBufferSize, brokerId,
      processors.slice(processorBeginIndex, processorEndIndex), connectionQuotas)
    acceptors.put(endpoint, acceptor)
    KafkaThread.nonDaemon(s"kafka-socket-acceptor-$listenerName-$securityProtocol-${endpoint.port}", acceptor).start()
    acceptor.awaitStartup()
    processorBeginIndex = processorEndIndex
  }
}

Acceptor创建过程中启动了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()
  val serverChannel = openServerSocket(endPoint.host, endPoint.port)
  this.synchronized {
    processors.foreach { processor =>
      KafkaThread.nonDaemon(s"kafka-network-thread-$brokerId-${endPoint.listenerName}-${endPoint.securityProtocol}-${processor.id}",
        processor).start()
    }
  }

Acceptor和Processor启动后各自执行自己的loop。

Acceptor只负责接收新连接，并采用round-robin的方式交给各个Processor

/**
  * Accept loop that checks for new connection attempts
  */
 def run() {
   serverChannel.register(nioSelector, SelectionKey.OP_ACCEPT)
   startupComplete()
   try {
     var currentProcessor = 0
     while (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.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)
             }
           }
         }
       }
       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 e
         case e: Throwable => error("Error occurred", e)
       }
     }
   } finally {
     debug("Closing server socket and selector.")
     swallowError(serverChannel.close())
     swallowError(nioSelector.close())
     shutdownComplete()
   }
 }

/*
   * Accept a new connection
   */
  def accept(key: SelectionKey, processor: Processor) {
    val serverSocketChannel = key.channel().asInstanceOf[ServerSocketChannel]
    val socketChannel = serverSocketChannel.accept()
    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)
    } 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的循环。

• 设置新连接
• 如果有Response尝试写回
• 带timeout的poll一次
• 接收Request
• 处理已经发送成功的Response
• 处理已经断开的连接

override def run() {
   startupComplete()
   while (isRunning) {
     try {
       // setup any new connections that have been queued up
       configureNewConnections()
       // register any new responses for writing
       processNewResponses()
       poll()
       processCompletedReceives()
       processCompletedSends()
       processDisconnected()
     } catch {
       // We catch all the throwables here to prevent the processor thread from exiting. We do this because
       // letting a processor exit might cause a bigger impact on the broker. Usually the exceptions thrown would
       // be either associated with a specific socket channel or a bad request. We just ignore the bad socket channel
       // or request. This behavior might need to be reviewed if we see an exception that need the entire broker to stop.
       case e: ControlThrowable => throw e
       case e: Throwable =>
         error("Processor got uncaught exception.", e)
     }
   }
   debug("Closing selector - processor " + id)
   swallowError(closeAll())
   shutdownComplete()
 }

处理新连接 configureNewConnections

Acceptor传过来的新socket放在了一个ConcorrentLinkedQueue中，
congiureNewConnections()负责获取ip端口号等信息然后注册到Processor自己的selector上。这个selector是Kafka封装了一层的KSelector

/**
   * 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)
      }
    }
  }

processNewResponses

从requestChannel中poll待写回的Response，这里是将Channel的send变量设置为Response.responseSend等待Selector处理

private def processNewResponses() {
    var curr = requestChannel.receiveResponse(id)
    while (curr != null) {
      try {
        curr.responseAction match {
          case RequestChannel.NoOpAction =>
            // There is no response to send to the client, we need to read more pipelined requests
            // that are sitting in the server's socket buffer
            updateRequestMetrics(curr.request)
            trace("Socket server received empty response to send, registering for read: " + curr)
            val channelId = curr.request.connectionId
            if (selector.channel(channelId) != null || selector.closingChannel(channelId) != null)
                selector.unmute(channelId)
          case RequestChannel.SendAction =>
            val responseSend = curr.responseSend.getOrElse(
              throw new IllegalStateException(s"responseSend must be defined for SendAction, response: $curr"))
            sendResponse(curr, responseSend)
          case RequestChannel.CloseConnectionAction =>
            updateRequestMetrics(curr.request)
            trace("Closing socket connection actively according to the response code.")
            close(selector, curr.request.connectionId)
        }
      } finally {
        curr = requestChannel.receiveResponse(id)
      }
    }
  }

Selector.send

/**
    * Queue the given request for sending in the subsequent {@link #poll(long)} calls
    * @param send The request to send
    */
   public void send(Send send) {
       String connectionId = send.destination();
       if (closingChannels.containsKey(connectionId))
           this.failedSends.add(connectionId);
       else {
           KafkaChannel channel = channelOrFail(connectionId, false);
           try {
               channel.setSend(send);
           } catch (CancelledKeyException e) {
               this.failedSends.add(connectionId);
               close(channel, false);
           }
       }
   }

processCompletedReceives

Selector在接收到请求后，将数据放到一个List中，Processor取出后put到requestChannel的requestQueue中

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 = new RequestChannel.Request(processor = id, connectionId = receive.source, session = session,
          startTimeNanos = time.nanoseconds, listenerName = listenerName, securityProtocol = securityProtocol,
          memoryPool, receive.payload)
        requestChannel.sendRequest(req)
        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)
      }
    }
  }

processCompletedSends

在Selector发送完成Resposne后，从inflightResponse中remove掉这个connnection -> resposne的键值对，当前inflightResposne只用于验证response的正确性，就是一个Channel写的数据必须在发送后先记录在inflightResponse中

private def processCompletedSends() {
  selector.completedSends.asScala.foreach { send =>
    val resp = inflightResponses.remove(send.destination).getOrElse {
      throw new IllegalStateException(s"Send for ${send.destination} completed, but not in `inflightResponses`")
    }
    updateRequestMetrics(resp.request)
    selector.unmute(send.destination)
  }
}

processDisconnected

写失败的连接和由于各种原因close的连接，需要清理已经占用的内存空间，例如inflightResponses。

private def processDisconnected() {
    selector.disconnected.keySet.asScala.foreach { connectionId =>
      val remoteHost = ConnectionId.fromString(connectionId).getOrElse {
        throw new IllegalStateException(s"connectionId has unexpected format: $connectionId")
      }.remoteHost
      inflightResponses.remove(connectionId).foreach(response => updateRequestMetrics(response.request))
      // the channel has been closed by the selector but the quotas still need to be updated
      connectionQuotas.dec(InetAddress.getByName(remoteHost))
    }
  }

至此网络部分基本分析完成，后面有涉及到的要注意的地方会单独介绍。
startup完成后，KafkaServer继续完成其他的startup

Kafka Client端网络代码

clients包里分成主要Send、Receive、KafkaChannel和Selector四部分

Selectable是其中的网络操作的接口, Selector是具体的实现, 包括了发送请求、接收返回、建立连接、断开连接等操作。

/**
 * An interface for asynchronous, multi-channel network I/O
 */
public interface Selectable {
    /**
     * See {@link #connect(String, InetSocketAddress, int, int) connect()}
     */
    public static final int USE_DEFAULT_BUFFER_SIZE = -1;
    /**
     * Begin establishing a socket connection to the given address identified by the given address
     * @param id The id for this connection
     * @param address The address to connect to
     * @param sendBufferSize The send buffer for the socket
     * @param receiveBufferSize The receive buffer for the socket
     * @throws IOException If we cannot begin connecting
     */
    public void connect(String id, InetSocketAddress address, int sendBufferSize, int receiveBufferSize) throws IOException;
    /**
     * Wakeup this selector if it is blocked on I/O
     */
    public void wakeup();
    /**
     * Close this selector
     */
    public void close();
    /**
     * Close the connection identified by the given id
     */
    public void close(String id);
    /**
     * Queue the given request for sending in the subsequent {@link #poll(long) poll()} calls
     * @param send The request to send
     */
    public void send(Send send);
    /**
     * Do I/O. Reads, writes, connection establishment, etc.
     * @param timeout The amount of time to block if there is nothing to do
     * @throws IOException
     */
    public void poll(long timeout) throws IOException;
    /**
     * The list of sends that completed on the last {@link #poll(long) poll()} call.
     */
    public List<Send> completedSends();
    /**
     * The list of receives that completed on the last {@link #poll(long) poll()} call.
     */
    public List<NetworkReceive> completedReceives();
    /**
     * The connections that finished disconnecting on the last {@link #poll(long) poll()}
     * call. Channel state indicates the local channel state at the time of disconnection.
     */
    public Map<String, ChannelState> disconnected();
    /**
     * The list of connections that completed their connection on the last {@link #poll(long) poll()}
     * call.
     */
    public List<String> connected();
    ...
}

Send作为要发送数据的接口, 子类实现complete()方法用于判断是否已经发送完成，实现writeTo(GatheringByteChannel channel)方法来实现写入到Channel中，