kafka在设计上大量使用了Selector+Channel+Buffer的设计原理.所以在开始之前简单介绍一下NIO 的Selector+Channel+Buffer
NIO 的Selector+Channel+Buffer
Buffers（缓冲区）
Java NIO中的Buffer用于和NIO通道进行交互。
缓冲区本质上是一块可以写入数据，然后可以从中读取数据的内存。这块内存被包装成NIO Buffer对象，并提供了一组方法，用来方便的访问该块内存
标准的IO基于字节流和字符流进行操作的，而NIO是基于通道（Channel）和缓冲区（Buffer）进行操作，数据总是从通道读取到缓冲区中，或者从缓冲区写入到通道中。
Channels(通道)
Java NIO的通道类似流，但又有些不同：
既可以从通道中读取数据，又可以写数据到通道。但流的读写通常是单向的。
通道可以异步地读写。
如下面图示是Buffers与Channel交互:
image.png
Selectors（选择器）
选择器用于监听多个通道的事件（比如：连接打开，数据到达）。Selector（选择器）是Java NIO中能够检测一到多个NIO通道，并能够知晓通道是否为诸如读写事件做好准备的组件。这样，一个单独的线程可以管理多个channel，从而管理多个网络连接
下面是单线程中使用一个Selector处理3个Channel的图示：
image.png
Non-blocking IO（非阻塞IO）
当线程从通道读取数据到缓冲区时，线程还是可以进行其他事情。当数据被写入到缓冲区时，线程可以继续处理它。从缓冲区写入通道也类似。
Broker请求处理流程
下面通过重要环节的源码分析,来梳理请求处理的整个过程(kafka2.3)
- KafkaServer Kafka的网络层入口类是SocketServer。
kafka.Kafka是Kafka Broker的入口类，kafka.Kafka.main()是Kafka Server的main()方法，即Kafka Broker的启动入口。我们跟踪代码，即沿着方法调用栈kafka.Kafka.main() -> kafkaServerStartable.startup() -> KafkaServer().startup可以从main()方法入口一直跟踪到SocketServer即网络层对象的创建，这意味着Kafka Server启动的时候会初始化并启动SocketServer。
def main(args: Array[String]): Unit = {
try {
val serverProps = getPropsFromArgs(args)
val kafkaServerStartable = KafkaServerStartable.fromProps(serverProps)
// 部分省略 ...
kafkaServerStartable.startup()
kafkaServerStartable.awaitShutdown()
}
catch {
case e: Throwable =>
fatal("Exiting Kafka due to fatal exception", e)
Exit.exit(1)
}
Exit.exit(0)
}
class KafkaServerStartable(val staticServerConfig: KafkaConfig, reporters: Seq[KafkaMetricsReporter]) extends Logging {
private val server = new KafkaServer(staticServerConfig, kafkaMetricsReporters = reporters)
...
def startup() {
try server.startup()
catch {
...
}
}
}
SocketServer处理与代理之间的新连接、请求和响应。
Kafka支持两种类型的请求数据层面:处理来自集群中的客户端和其他代理的请求。
线程模型是每个监听器有一个Acceptor线程，用来处理新的连接。可以通过在KafkaConfig中为“ listeners”指定多个“、”分隔的endpoint来配置多个监听端口。
Acceptor有N个处理器线程（每个线程都有自己的selector并从套接字中读取请求）和M处理程序线程(它处理请求并将响应返回给处理器线程进行编写)控制层面:处理来自控制器的请求。这是可选的，可以通过指定“control.plan .listener.name”来配置。如果没有配置，控制器请求由数据层面处理。
线程模型是处理新连接的接受线程Acceptor有一个处理器线程(它有自己的选择器并从套接字中读取请求)和1处理程序线程，它处理请求并将响应生成回处理器线程进行编写SocketServer的startup方法,创建Control和Data层面的Acceptor和Processor线程并启动所有的processor线程
def startup(startupProcessors: Boolean = true) {
this.synchronized {
connectionQuotas = new ConnectionQuotas(config, time)
//控制层面
createControlPlaneAcceptorAndProcessor(config.controlPlaneListener)
//数据层面
createDataPlaneAcceptorsAndProcessors(config.numNetworkThreads, config.dataPlaneListeners)
if (startupProcessors) {
//在控制层面启动Processor线程
startControlPlaneProcessor()
//在数据层面启动Processor线程
startDataPlaneProcessors()
}
}
}
private def createDataPlaneAcceptorsAndProcessors(dataProcessorsPerListener: Int,
endpoints: Seq[EndPoint]): Unit = synchronized {
endpoints.foreach { endpoint =>
connectionQuotas.addListener(config, endpoint.listenerName)
//每一个endPoint创建一个Acceptor,创建多个Processor放入processor线程数组
val dataPlaneAcceptor = createAcceptor(endpoint, DataPlaneMetricPrefix)
addDataPlaneProcessors(dataPlaneAcceptor, endpoint, dataProcessorsPerListener)
}
}
- Acceptor的构造方法中，首先通过openServerSocket()打开自己负责的EndPoint的Socket，即打开端口并启动监听。
然后，Acceptor会负责构造并管理的一个Processor的ArrayBuffer。其实，每一个Processor都是一个独立线程 - Acceptor线程的run()方法，是不断监听对应ServerChannel上的连接请求(ACCEPT)，如果有新的连接请求，使用的轮询方式将通道分配给Processor.
新连接交付给Processor的具体的调用是在方法assignNewConnection方法中
private[kafka] class Acceptor(val endPoint: EndPoint,
val sendBufferSize: Int,
val recvBufferSize: Int,
brokerId: Int,
connectionQuotas: ConnectionQuotas,
metricPrefix: String) extends AbstractServerThread(connectionQuotas) with KafkaMetricsGroup {
private val nioSelector = NSelector.open()
val serverChannel = openServerSocket(endPoint.host, endPoint.port)
private val processors = new ArrayBuffer[Processor]()
/**
* Accept loop that checks for new connection attempts
*/
def run() {
//将ServerChannel注册到Selector,并监听ACCEPT事件
serverChannel.register(nioSelector, SelectionKey.OP_ACCEPT)
startupComplete()
try {
var currentProcessorIndex = 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).foreach { socketChannel =>
var retriesLeft = synchronized(processors.length)
var processor: Processor = null
do {
retriesLeft -= 1
processor = synchronized {
processors(currentProcessorIndex)
}
currentProcessorIndex += 1
// 此处调用assignNewConnection
} 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 {
// ...
}
}
}
//...
}
- assignNewConnection中通过processor.accept的调用,将SocketChannel放入每个processor自己维护的新连接的队列,后面processor会从队列取出做后续处理
private def assignNewConnection(socketChannel: SocketChannel, processor: Processor, mayBlock: Boolean): Boolean = {
//调用processor.accept
if (processor.accept(socketChannel, mayBlock, blockedPercentMeter)) {
// ...
true
} else
false
}
- 每一个Processor都维护了一个单独的Selector对象，这个Selector只负责这个Processor上所有channel的监听。这样最大程度上保证了不同Processor线程之间的完全并行和业务隔离.同时每一个processor维护一个responseQueue,用于KafkaRequestHandler交互,在下面的流程会提到
private[kafka] class Processor(val id: Int,
time: Time,
maxRequestSize: Int,
requestChannel: RequestChannel,
connectionQuotas: ConnectionQuotas,
connectionsMaxIdleMs: Long,
failedAuthenticationDelayMs: Int,
listenerName: ListenerName,
securityProtocol: SecurityProtocol,
config: KafkaConfig,
metrics: Metrics,
credentialProvider: CredentialProvider,
memoryPool: MemoryPool,
logContext: LogContext,
connectionQueueSize: Int = ConnectionQueueSize) extends AbstractServerThread(connectionQuotas) with KafkaMetricsGroup {
// 维护一个新连接队列,在run方法里会取出处理
private val newConnections = new ArrayBlockingQueue[SocketChannel](connectionQueueSize)
//每一个processor维护一个responseQueue
private val responseQueue = new LinkedBlockingDeque[RequestChannel.Response]()
// processor都维护了一个单独的Selector
private val selector = createSelector(
ChannelBuilders.serverChannelBuilder(listenerName,
listenerName == config.interBrokerListenerName,
securityProtocol,
config,
credentialProvider.credentialCache,
credentialProvider.tokenCache,
time))
// Visible to override for testing
protected[network] def createSelector(channelBuilder: ChannelBuilder): KSelector = {
channelBuilder match {
case reconfigurable: Reconfigurable => config.addReconfigurable(reconfigurable)
case _ =>
}
new KSelector(
maxRequestSize,
connectionsMaxIdleMs,
failedAuthenticationDelayMs,
metrics,
time,
"socket-server",
metricTags,
false,
true,
channelBuilder,
memoryPool,
logContext)
}
override def run() {
//表示初始化流程已经结束，通过这个CountDownLatch代表初始化已经结束，这个Processor已经开始正常运行了
startupComplete()
try {
while (isRunning) {
try {
// setup any new connections that have been queued up
configureNewConnections()
// register any new responses for writing
//处理响应队列,这个响应队列是Handler线程处理以后的结果，会交付给RequestChannel.responseQueue.同时调用unmute，开始接受请求
processNewResponses()
//调用KSelector.poll(),进行真正的数据读写
poll()
//调用Selector.mute,不再接受Read请求,发送响应之前,不可以再接收任何请求
processCompletedReceives()
processCompletedSends()
processDisconnected()
closeExcessConnections()
} catch {
// ...
}
}
} finally {
// ...
}
}
}
- run方法中configureNewConnections是processor从自己维护的newConnections队列取出新连接,并将其注册到selector并监听OR_READ事件。configureNewConnections 内部调用register()方法，会将新接收的新连接SocketChannel注册到服务器端的Selector，并监听OP_READ事件,如果发生读请求,可以取出对应的request进行后续处理
private def configureNewConnections() {
var connectionsProcessed = 0
while (connectionsProcessed < connectionQueueSize && !newConnections.isEmpty) {
// 取出新连接SocketChannel
val channel = newConnections.poll()
try {
// 将SocketChannel注册到selector
selector.register(connectionId(channel.socket), channel)
connectionsProcessed += 1
} catch {
case e: Throwable =>
// ...
}
}
}
public void register(String id, SocketChannel socketChannel) throws IOException {
ensureNotRegistered(id);
registerChannel(id, socketChannel, SelectionKey.OP_READ);
this.sensors.connectionCreated.record();
}
- RequestChannel 负责消息从网络层转接到业务层，以及将业务层的处理结果交付给网络层进而返回给客户端。每一个SocketServer只有一个RequestChannel对象，在SocketServer中构造。RequestChannel构造方法中初始化了requestQueue，用来存放网络层接收到的请求，这些请求即将交付给业务层进行处理。同时，初始化了responseQueues，为每一个Processor建立了一个response队列，用来存放这个Processor的一个或者多个Response，这些response即将交付给网络层返回给客户端。
class RequestChannel(val queueSize: Int, val metricNamePrefix : String) extends KafkaMetricsGroup {
import RequestChannel._
val metrics = new RequestChannel.Metrics
private val requestQueue = new ArrayBlockingQueue[BaseRequest](queueSize)
private val processors = new ConcurrentHashMap[Int, Processor]()
/** 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)
}
}
}
- Processor.processCompletedReceives()通过遍历completedReceives，对于每一个已经完成接收的数据，对数据进行解析和封装，交付给RequestChannel，RequestChannel会交付给具体的业务处理层进行处理。其中RequestChannel拿到请求数据,会调用RequestChannel.sendRequest方法,将请求put到requestQueue中,以供后续的处理请求线程处理
private def processCompletedReceives() {
selector.completedReceives.asScala.foreach { receive =>
try {
openOrClosingChannel(receive.source) match {
case Some(channel) =>
else {
val nowNanos = time.nanoseconds()
if (channel.serverAuthenticationSessionExpired(nowNanos)) {
// ...
} else {
//将请求通过RequestChannel.requestQueue交付给Handler
requestChannel.sendRequest(req)
selector.mute(connectionId)//不再接受Read请求,发送响应之前，不可以再接收任何请求
handleChannelMuteEvent(connectionId, ChannelMuteEvent.REQUEST_RECEIVED)
}
}
} catch {
// ...
}
}
}
- KafkaRequestHandler请求处理线程和KafkaRequestHandlerPool线程池
KafkaRequestHandler 主要关注run方法,该方法的具体逻辑是从RequestChannel取出processor之前put请求,调用KafkaApi针对不同请求类型分别处理
class KafkaRequestHandler(id: Int,
brokerId: Int,
val aggregateIdleMeter: Meter,
val totalHandlerThreads: AtomicInteger,
val requestChannel: RequestChannel,
apis: KafkaApis,
time: Time) extends Runnable with Logging {
def run() {
while (!stopped) {
//从RequestChannel.requestQueue中取出请求
val req = requestChannel.receiveRequest(300)
req match {
case RequestChannel.ShutdownRequest =>
shutdownComplete.countDown()
return
case request: RequestChannel.Request =>
try {
// 调用KafkaApi.handle()，将请求交付给业务
apis.handle(request)
} catch {
// 异常处理 ...
} finally {
request.releaseBuffer()
}
case null => // continue
}
}
shutdownComplete.countDown()
}
- KafkaRequestHandlerPool构造方法中初始化并启动了多个KafkaRequestHandler线程对象，线程池大小通过Kafka配置文件配置项num.io.threads进行配置。
KafkaRequestHandlerPool线程池中的所有KafkaRequestHandler，通过竞争方式从RequestChannel.requestQueue中获取请求进行处理。由于requestQueue的类型是ArrayBlockingQueue，通过调用ArrayBlockingQueue.poll()方法取出请求.
class KafkaRequestHandlerPool(val brokerId: Int,
val requestChannel: RequestChannel,
val apis: KafkaApis,
time: Time,
numThreads: Int,
requestHandlerAvgIdleMetricName: String,
logAndThreadNamePrefix : String) extends Logging with KafkaMetricsGroup {
private val threadPoolSize: AtomicInteger = new AtomicInteger(numThreads)
//初始化由KafkaRequestHandler线程构成的线程数组
val runnables = new mutable.ArrayBuffer[KafkaRequestHandler](numThreads)
for (i <- 0 until numThreads) {
createHandler(i)
}
def createHandler(id: Int): Unit = synchronized {
runnables += new KafkaRequestHandler(id, brokerId, aggregateIdleMeter, threadPoolSize, requestChannel, apis, time)
KafkaThread.daemon(logAndThreadNamePrefix + "-kafka-request-handler-" + id, runnables(id)).start()
}
// ...
}
- KafkaApis类似一个工具类，解析用户请求并将请求交付给业务层，我们可以把它看做Kafka的API层。从上面KafkaRequestHandler.run()方法可以看到，这是通过调用KafkaApis.handle()方法完成的
def handle(request: RequestChannel.Request) {
request.header.apiKey match {
case ApiKeys.PRODUCE => handleProduceRequest(request)
case ApiKeys.FETCH => handleFetchRequest(request)
case ApiKeys.LIST_OFFSETS => handleListOffsetRequest(request)
case ApiKeys.METADATA => handleTopicMetadataRequest(request)
case ApiKeys.LEADER_AND_ISR => handleLeaderAndIsrRequest(request)
//其它ApiKeys，略
//异常处理略
}
}
- 我们以ApiKeys.PRODUCE 的流程来分析后续流程,handleProduceRequest方法中有两个重要的方法sendResponseCallback()和replicaManager.appendRecords() .其中sendResponseCallback回调函数中调用requestChannel.sendResponse()将response交付给RequestChannel
def handleProduceRequest(request: RequestChannel.Request) {
val produceRequest = request.body[ProduceRequest]
// 回调函数,内部将业务层处理的最终结果发送到对应processor负责的响应队列
def sendResponseCallback(responseStatus: Map[TopicPartition, PartitionResponse]) {
// Send the response immediately. In case of throttling, the channel has already been muted.
if (produceRequest.acks == 0) {
// 通过RequestChannel将response放入processor的响应队列,调用requestChannel.sendResponse()将response交付给RequestChannel
sendResponse(request, Some(new ProduceResponse(mergedResponseStatus.asJava, maxThrottleTimeMs)), None)
}
}
// appendRecords方法是records写入的逻辑
replicaManager.appendRecords(
timeout = produceRequest.timeout.toLong,
requiredAcks = produceRequest.acks,
internalTopicsAllowed = internalTopicsAllowed,
isFromClient = true,
entriesPerPartition = authorizedRequestInfo,
responseCallback = sendResponseCallback,
recordConversionStatsCallback = processingStatsCallback)
// ...
}
}
- 最后,在上文讲解Processor的时候说过，Procossor.processNewResponses()就是从requestChannel.responseQueues取出属于自己的连接上的响应，准备返回给客户端
一图胜千言,最后通过一张图来回顾整个Broker请求处理流程
整体流程图示如下:
image.png
参考自
https://blog.csdn.net/zhanyuanlin/article/details/76906583
http://ifeve.com/channels/
