提示： 
这里需要注意的是，由于UDP协议是一种不可靠的传输协议，所以Kafka系统采用TCP协议作为服务间的通信协议。

提示： 
Reactor是一种事件模型，可以将请求提交到一个或者多个服务程序中进行处理。 
当收到Client的请求后，Server处理程序使用多路分发策略，由一个非阻塞的线程来接收所有的请求，然后将这些请求转发到对应的工作线程中进行处理。

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() 
    } 
  }

class BlockingChannel( val host: String,  
val port: Int,  
val readBufferSize: Int,  
val writeBufferSize: Int,  
val readTimeoutMs: Int ) extends Logging { 
private var connected = false 
private var channel: SocketChannel = null 
private var readChannel: ReadableByteChannel = null 
private var writeChannel: GatheringByteChannel = null 
private val lock = new Object() 
private val connectTimeoutMs = readTimeoutMs 
private var connectionId: String = "" 
def connect() = lock synchronized  { 
if(!connected) { 
try { 
channel = SocketChannel.open() 
if(readBufferSize > 0) 
channel.socket.setReceiveBufferSize(readBufferSize) 
if(writeBufferSize > 0) 
channel.socket.setSendBufferSize(writeBufferSize) 
channel.configureBlocking(true) 
channel.socket.setSoTimeout(readTimeoutMs) 
channel.socket.setKeepAlive(true) 
channel.socket.setTcpNoDelay(true) 
channel.socket.connect(new InetSocketAddress(host, port), connectTimeoutMs) 
writeChannel = channel 
// Need to create a new ReadableByteChannel from input stream because SocketChannel doesn't implement read with timeout 
// See: http://stackoverflow.com/questions/2866557/timeout-for-socketchannel-doesnt-work 
readChannel = Channels.newChannel(channel.socket().getInputStream) 
connected = true 
val localHost = channel.socket.getLocalAddress.getHostAddress 
val localPort = channel.socket.getLocalPort 
val remoteHost = channel.socket.getInetAddress.getHostAddress 
val remotePort = channel.socket.getPort 
connectionId = localHost + ":" + localPort + "-" + remoteHost + ":" + remotePort 
// settings may not match what we requested above 
val msg = "Created socket with SO_TIMEOUT = %d (requested %d), SO_RCVBUF = %d (requested %d), SO_SNDBUF = %d (requested %d), connectTimeoutMs = %d." 
debug(msg.format(channel.socket.getSoTimeout, 
readTimeoutMs, 
channel.socket.getReceiveBufferSize,  
readBufferSize, 
channel.socket.getSendBufferSize, 
writeBufferSize, 
connectTimeoutMs)) 
} catch { 
case _: Throwable => disconnect() 
} 
} 
} 
def disconnect() = lock synchronized { 
if(channel != null) { 
swallow(channel.close()) 
swallow(channel.socket.close()) 
channel = null 
writeChannel = null 
} 
// closing the main socket channel *should* close the read channel 
// but let's do it to be sure. 
if(readChannel != null) { 
swallow(readChannel.close()) 
readChannel = null 
} 
connected = false 
} 
def isConnected = connected 
def send(request: RequestOrResponse): Long = { 
if(!connected) 
throw new ClosedChannelException() 
val send = new RequestOrResponseSend(connectionId, request) 
send.writeCompletely(writeChannel) 
} 
def receive(): NetworkReceive = { 
if(!connected) 
throw new ClosedChannelException() 
val response = readCompletely(readChannel) 
response.payload().rewind() 
response 
} 
private def readCompletely(channel: ReadableByteChannel): NetworkReceive = { 
val response = new NetworkReceive 
while (!response.complete()) 
response.readFromReadableChannel(channel) 
response 
} 
}

class RequestChannel(val numProcessors: Int, val queueSize: Int) extends KafkaMetricsGroup { 
private var responseListeners: List[(Int) => Unit] = Nil 
private val requestQueue = new ArrayBlockingQueue[RequestChannel.Request](queueSize) 
private val responseQueues = new Array[BlockingQueue[RequestChannel.Response]](numProcessors) 
for(i <- 0 until numProcessors) 
responseQueues(i) = new LinkedBlockingQueue[RequestChannel.Response]() 
newGauge( 
"RequestQueueSize", 
new Gauge[Int] { 
def value = requestQueue.size 
} 
) 
newGauge("ResponseQueueSize", new Gauge[Int]{ 
def value = responseQueues.foldLeft(0) {(total, q) => total + q.size()} 
}) 
for (i <- 0 until numProcessors) { 
newGauge("ResponseQueueSize", 
new Gauge[Int] { 
def value = responseQueues(i).size() 
}, 
Map("processor" -> i.toString) 
) 
} 
/** 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) 
} 
/** Send a response back to the socket server to be sent over the network */ 
def sendResponse(response: RequestChannel.Response) { 
responseQueues(response.processor).put(response) 
for(onResponse <- responseListeners) 
onResponse(response.processor) 
} 
/** No operation to take for the request, need to read more over the network */ 
def noOperation(processor: Int, request: RequestChannel.Request) { 
responseQueues(processor).put(RequestChannel.Response(processor, request, null, RequestChannel.NoOpAction)) 
for(onResponse <- responseListeners) 
onResponse(processor) 
} 
/** Close the connection for the request */ 
def closeConnection(processor: Int, request: RequestChannel.Request) { 
responseQueues(processor).put(RequestChannel.Response(processor, request, null, RequestChannel.CloseConnectionAction)) 
for(onResponse <- responseListeners) 
onResponse(processor) 
} 
/** Get the next request or block until specified time has elapsed */ 
def receiveRequest(timeout: Long): RequestChannel.Request = 
requestQueue.poll(timeout, TimeUnit.MILLISECONDS) 
/** Get the next request or block until there is one */ 
def receiveRequest(): RequestChannel.Request = 
requestQueue.take() 
/** Get a response for the given processor if there is one */ 
def receiveResponse(processor: Int): RequestChannel.Response = { 
val response = responseQueues(processor).poll() 
if (response != null) 
response.request.responseDequeueTimeMs = Time.SYSTEM.milliseconds 
response 
} 
def addResponseListener(onResponse: Int => Unit) { 
responseListeners ::= onResponse 
} 
def shutdown() { 
requestQueue.clear() 
} 
}