1. 这两个类都是  AbstractServerThead的实现类,超类是Runnable 可运行的。
2. 每个Acceptor持有num.network.threads个  Processor  线程, 假如配置了多个listeners，那么总共Processor线程数是 listeners*num.network.threads.
3. Acceptor  创建的是ServerSocketChannel通道,这个通道是用来监听新进来的TCP链接的通道,
   通过serverSocketChannel.accept()方法可以拿到SocketChannel通道用于传输数据。
4. 每个Processor  线程都有一个唯一的id，并且通过Acceptor拿到的SocketChannel会被暂时放入到newConnections队列中
5. 每个Processor  都创建了自己的Selector
6. Processor会不断的从自身的newConnections队列里面获取新SocketChannel，并注册读写事件,如果有数据传输过来,则会读取数据,并解析成Request请求。

def run(): Unit =
//将serverChannel 注册到nioselector上,并且对 Accept事件感兴趣：表示服务器监听到了客户连接，那么服务器可以接收这个连接了
serverChannel.register(nioSelector, SelectionKey.OP_ACCEPT)
try
var currentProcessorIndex = 0
while (isRunning)
try
//返回感兴趣的事件数量这里是感兴趣的是SelectionKey.OP_ACCEPT，监听到新的链接
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()
//只处理Accept事件，其他的事件则抛出异常，ServerSocketChannel是 监听Tcp的链接通道
if (key.isAcceptable)
//根据Key 拿到SocketChannle = serverSocketChannel.accept()，然后再遍历
accept(key).foreachsocketChannel =>

//将socketChannel分配给我们的 processor来处理，如果有多个socketChannel 则按照轮训分配的原则
//如果一个processor 中能够处理的newconnection 队列满了放不下了，则找下一个
// 如果所有的都放不下，则会一直循环直到有processor能够处理。

var retriesLeft = synchronized(processors.length)
var processor: Processor = null
do
retriesLeft -= 1
//轮训每个processors来处理
processor = synchronized
// adjust the index (if necessary) and retrieve the processor atomically for
// correct behaviour in case the number of processors is reduced dynamically
currentProcessorIndex = currentProcessorIndex % processors.length
processors(currentProcessorIndex)

currentProcessorIndex += 1
while (!assignNewConnection(socketChannel, processor, retriesLeft == 0))

else
throw new IllegalStateException("Unrecognized key state for acceptor thread.")
catch
case e: Throwable => error("Error while accepting connection", e)




catch
省略


finally
省略

serverChannel.register(nioSelector, SelectionKey.OP_ACCEPT)

// 阻塞查询Selector是否有监听到新的事件
val ready = nioSelector.select(500)
// 如果有事件,则查询具体的事件和通道
if(ready> 0>
//获取所有就绪事件准备处理
val keys = nioSelector.selectedKeys()

Unrecognized key state for acceptor thread

val serverSocketChannel = key.channel().asInstanceOf[ServerSocketChannel]
// 调用accept方法就可以拿到ScoketChannel了。
val socketChannel = serverSocketChannel.accept()
//设置为非阻塞模式 就可以在异步模式下调用connect(), read() 和write()了。
socketChannel.configureBlocking(false)

override def run(): Unit =
startupComplete()
try
while (isRunning)
try
// setup any new connections that have been queued up
// 将之前监听到的TCP链接(暂时保存在newConnections中) 开始注册监听OP_READ事件到每个Processor的 KSelector选择器中。
configureNewConnections()
// register any new responses for writing
processNewResponses()
//在不阻塞的情况下对每个连接执行任何 I/O 操作。这包括完成连接、完成断开连接、启动新发送或在进行中的发送或接收上取得进展。
// 当此调用完成时，用户可以使用completedSends() 、 completedReceives() 、 connected() 、 disconnected()检查已完成的发送、接收、连接或断开连接。
poll()
// 把请求解析后放到 requestChannels 队列中，异步处理
processCompletedReceives()
//处理已经发送完成的请求
processCompletedSends()
processDisconnected()
closeExcessConnections()
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. This behavior might need to be
// reviewed if we see an exception that needs the entire broker to stop. Usually the exceptions thrown would
// be either associated with a specific socket channel or a bad request. These exceptions are caught and
// processed by the individual methods above which close the failing channel and continue processing other
// channels. So this catch block should only ever see ControlThrowables.
case e: Throwable => processException("Processor got uncaught exception.", e)


finally
debug(s"Closing selector - processor $id")
CoreUtils.swallow(closeAll(), this, Level.ERROR)
shutdownComplete()

1. configureNewConnections(): 之前Acceptor监听到的SocketChannel保存在Procesor中的newConnections阻塞队列中, 现在开始将newConnections阻塞队列一个个取出来,向Procesor的Selector注册SocketChannel通道，并且感兴趣的事件为SelectionKey.OP_READ读事件。
2. processNewResponses() : 去Processor里面的无边界阻塞队列responseQueue里面获取RequestChannel.Response数据, 如果有数据并且需要返回Response的话, 则通过channel返回数据. 具体的Channel是根据connectionId 获取之前构建的KafkaChannel,  KafkaChannel则会通过监听SelectionKey.OP_WRITE。然后调用writeTo方法。
   至于responseQueue这个队列是什么时候入队的,我们后面再分析
3. poll(): 这个方法里面执行的就很多了, 这个方法底层调用的是selector.poll(); 将监听到的事件批量处理,它才是执行I/O请求的最终地方, 它正对每个连接执行任何的I/O操作,这包括了 完成连接、完成断开连接、启动新发送等等。
   像校验身份信息,还有handshake等等这些也都是在这里执行的。
4. processCompletedReceives(): 处理所有completedReceives(已完成接收的请求)进行接下来的处理, 处理的方式是解析一下收到的请求,最终调用了requestChannel.sendRequest(req). 也就是说所有的请求最终通过解析放入到了RequestChannel中的requestQueue阻塞队列中, 这个阻塞队列的大小为queued.max.requests默认500；表示的是在阻塞网络线程之前，数据平面允许的排队请求数
   PS: 这个completedReceives 是在 poll()方法中添加的元素。
5. processCompletedSends(): 它负责处理 Response 的回调逻辑，通过遍历completedSends(已完成发送)集合 可以从inflightResponses中移除并拿到response对象,然后再调用回调逻辑。
   PS: 这个completedSends 是在 poll()方法中添加的元素。
6. processDisconnected(): 处理断开链接的情况, connectionQuotas连接限流减掉这个链接，inflightResponses也移除对应连接。
7. closeExcessConnections():  关闭超限连接  ，当总连接数 > max.connections & & (inter.broker.listener.name!=listener|| listeners 数量==1) 则需要关闭一些连接.
   简单来说就是：就算Broker已经达到了最大连接数的限制了, 也应该允许 broker之间监听器上的连接, 这种情况下,将会关闭另外一个监听器上最近最少使用的连接。broker之间的监听器是配置inter.broker.listener.name 决定的
   所谓优先关闭，是指在诸多 TCP 连接中找出最近未被使用的那个。这里“未被使用”就是说，在最近一段时间内，没有任何 Request 经由这个连接被发送到 Processor 线程。