Kafka 服务端（Broker）采用 Reactor 的架构思想，通过1 个 Acceptor，N 个 Processor(N默认为3)，M 个 KafkaRequestHandler（M默认为8），来处理客户端请求，这种模式结合了多线程和事件驱动的设计，优点是能够有效地利用系统资源，可以实现高效地处理请求，无需为每个连接或请求创建新的线程，减少了线程上下文切换的开销，以实现高并发和高吞吐量。
文章目录
- 服务端整体架构
- 执行流程
- 源码剖析
- Acceptor 线程处理
- Processor 线程处理
- KafkaRequestHandlerPool 处理请求
服务端整体架构
Kafka 服务端的网络结构主要包含以下三层：
- 网络连接层：Acceptor 线程接收客户端的连接请求并创建网络连接。
- 请求转发层：Acceptor 线程以轮询的方式分发给Processor 线程，从而实现负载均衡的效果，Processor
线程将请求放到请求队列中。 - 请求处理层：KafkaRequestHandler线程不断地从请求队列中获取请求，解析请求,调用KafkaAPIs获取对应的操作结果,并将结果返回给客户端。
执行流程
Acceptor线程在初始化的时候会往selector注册 OP_ACCEPT事件，表示可以接受客户端的连接请求，当客户端有请求连接过来时，根据selectionkey可以得到socketChannel，再将socketChannel以轮询的方式交给Processor线程（默认有3个Processor线程）处理。- Processor线程收到Acceptor线程分发的连接后，会先将连接放入自己的队列
newConnections中，然后在selector注册OP_READ事件，表示可以读取客户端的请求，当客户端发送消息过来时，Processor线程就会处理OP_READ事件，然后Processor线程会将客户端的请求连接放入requestChannel的RequestQueue（请求队列被所有Processor线程共享）里并取消OP_READ事件的监听。 KafkaRequestHandler线程（默认会创建8个线程）会从RequestQueue取出请求进行处理，通过KafkaApis调用得到响应结果，将处理后的响应结果放入responseQueues中（每个Processor线程对应一个responseQueues)。- Processor线程往selector
注册OP_WRITE事件，表示可以将响应结果发送给客户端，当Processor线程检测到有OP_WRITE事件时，Processor线程就会从对应的responseQueues中取出响应结果，并通过selector.poll()方法将响应结果发送给对应的客户端且取消OP_WRITE事件的监听，最后Processor线程就会重新注册OP_READ事件，准备下一个请求的处理。
源码剖析
服务端的核心代码都在kafka.scala这个类，首先是main入口方法，该方法主要设置参数，然后调用启动方法
def main(args: Array[String]): Unit = {try {//启动服务端的时候,在这里解析参数val serverProps = getPropsFromArgs(args)val kafkaServerStartable = KafkaServerStartable.fromProps(serverProps)//启动的核心代码方法kafkaServerStartable.startup//...}
kafka的服务端核心方法都在startup()里面
def startup() {//启动服务server.startup()//...}
创建SocketServer，startup启动后，会创建Acceptor线程和三个Processor线程并启动
//Kafka 服务端的功能 都是在这里面实现
def startup() {//创建NIO的服务端socketServer = new SocketServer(config, metrics, kafkaMetricsTime)socketServer.startup()
} def startup() {this.synchronized {// 设置发送和接收的缓冲区大小val sendBufferSize = config.socketSendBufferBytesval recvBufferSize = config.socketReceiveBufferBytes//获取当前broker主机idval brokerId = config.brokerIdvar processorBeginIndex = 0//endpoints表示Kafka配置文件config/server.properties中的信息//正常情况下，只有一个服务实例endpoints.values.foreach { endpoint =>val protocol = endpoint.protocolType//processorEndIndex = 0 + 3val processorEndIndex = processorBeginIndex + numProcessorThreads//创建了三个Processor的线程for (i <- processorBeginIndex until processorEndIndex)//默认新建3个Processor线程processors(i) = newProcessor(i, connectionQuotas, protocol)//Acceptor类的主构造函数会启动3个Processor线程val acceptor = new Acceptor(endpoint, sendBufferSize, recvBufferSize, brokerId,processors.slice(processorBeginIndex, processorEndIndex), connectionQuotas)acceptors.put(endpoint, acceptor)// Utils是线程工具类，启动acceptor线程，Utils.newThread("kafka-socket-acceptor-%s-%d".format(protocol.toString, endpoint.port), acceptor, false).start()acceptor.awaitStartup()processorBeginIndex = processorEndIndex}}}
Acceptor 线程处理
Utils.newThread启动acceptor线程的start()方法后，就会执行该线程的run方法
-
首先
serverChannel向nioSelector注册OP_ACCEPT事件，nioSelector就会监听serverChannel是否有新的连接请求 -
若有新的连接请求到来，根据该连接的key创建
SocketChannel，然后通过轮询的方式分发给不同的processors线程处理，从而保证processor线程的负载均衡。
def run() {//ServerChannel往Selector注册OP_ACCEPT事件，表示可以接收客户端的请求，//Selector就会检查ServerChannel是否有新的请求到达serverChannel.register(nioSelector, SelectionKey.OP_ACCEPT)startupComplete()try {var currentProcessor = 0//死循环，不断轮询while (isRunning) {try {//selecotr 查看是否有新的注册事件val ready = nioSelector.select(500)//大于0，说明有新事件到来if (ready > 0) {//获取事件的keyval keys = nioSelector.selectedKeys()//遍历注册的所有keyval iter = keys.iterator()while (iter.hasNext && isRunning) {try {val key = iter.next//遍历完就删除iter.remove()//如果事件是OP_ACCEPT，就会调用accept()方法接收请求if (key.isAcceptable)// 创建SocketChannel，将其分发给Processor线程处理accept(key, processors(currentProcessor))elsethrow new IllegalStateException("Unrecognized key state for acceptor thread.")// 轮询遍历下一个Processor线程currentProcessor = (currentProcessor + 1) % processors.length} catch {case e: Throwable => error("Error while accepting connection", e)}}}}}
根据key封装socketChannel，分发给processor线程处理，processor线程将socketChannel放入自己的队列newConnections中，该队列是由ConcurrentLinkedQueue实现的队列，然后唤醒processor的 selector处理
def accept(key: SelectionKey, processor: Processor) {//根据SelectionKey获取到serverSocketChannelval serverSocketChannel = key.channel().asInstanceOf[ServerSocketChannel]//获取到一个socketChannelval socketChannel = serverSocketChannel.accept()try {connectionQuotas.inc(socketChannel.socket().getInetAddress)socketChannel.configureBlocking(false)socketChannel.socket().setTcpNoDelay(true)socketChannel.socket().setKeepAlive(true)// processor调用accept方法对socketChannel进行处理processor.accept(socketChannel)}}
def accept(socketChannel: SocketChannel) {//将接收的 SocketChannel放入到自己的队列newConnections.add(socketChannel)// 唤醒 Processor 的 selector 进行处理wakeup()}
Processor 线程处理
在上面的startup()中已经创建了3个Processor线程，然后在Acceptor的主构造函数中进行启动
//主构造函数，new出来的时候会被运行
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 {this.synchronized {//启动在startup()创建的3个Processor线程processors.foreach { processor =>Utils.newThread("kafka-network-thread-%d-%s-%d".format(brokerId, endPoint.protocolType.toString, processor.id), processor, false).start()}}
Processor启动之后就会执行run方法
override def run() {startupComplete()while (isRunning) {try {//读取队列中的每个SocketChannel,都往Selector上面注册OP_READ事件configureNewConnections()//处理响应，并注册OP_WRITE事件processNewResponses()//读取和发送请求的代码应该都是在这个方法完成，用于处理OP_READ事件与OP_WRITE事件poll()//处理接收到的新请求，将这些请求放入requestChannel请求队列中并取消OP_READ事件processCompletedReceives()//处理已经发送出去的响应并重新监听OP_READ事件processCompletedSends()processDisconnected()} swallowError(closeAll())shutdownComplete()}
不断获取连接队列里所有的SocketChannel，解析参数得到ConnectionId，再往selector注册OP_READ事件，注册之后就可以读取客户端的请求。
private def configureNewConnections() {
//当连接队列不为空while (!newConnections.isEmpty) {//不断获取连接队列里面的SocketChannelval channel = newConnections.poll()try {//解析SocketChannel，获取对应的参数val localHost = channel.socket().getLocalAddress.getHostAddressval localPort = channel.socket().getLocalPortval remoteHost = channel.socket().getInetAddress.getHostAddressval remotePort = channel.socket().getPortval connectionId = ConnectionId(localHost, localPort, remoteHost, remotePort).toString//往selector注册OP_READ事件selector.register(connectionId, channel)} }}
从这段代码可以知道kafka对SocketChannel进行了封装，封装成KakaChannel，并将SelectionKey和KakaChannel进行二者的绑定，除此之外，Kafka还实现了channel复用，将connectionId和KakaChannel放入map中，避免每次发起请求都新建channel，减少了资源的消耗。
public void register(String id, SocketChannel socketChannel) throws ClosedChannelException {//往自己的Selector上面注册OP_READ事件SelectionKey key = socketChannel.register(nioSelector, SelectionKey.OP_READ);//kafka对SocketChannel进行了封装，封装成KakaChannelKafkaChannel channel = channelBuilder.buildChannel(id, key, maxReceiveSize);//将key和channel进行绑定key.attach(channel);//channels护了多个网络连接，实现channel复用this.channels.put(id, channel);}
将客户端的请求放入请求队列中，并取消OP_READ事件
private def processCompletedReceives() {//遍历每一个请求selector.completedReceives.asScala.foreach { receive =>try {val channel = selector.channel(receive.source)val session = RequestChannel.Session(new KafkaPrincipal(KafkaPrincipal.USER_TYPE, channel.principal.getName),channel.socketAddress)//对于获取到的请求进行解析，得到requestval req = RequestChannel.Request(processor = id, connectionId = receive.source, session = session, buffer = receive.payload, startTimeMs = time.milliseconds, securityProtocol = protocol)//将request放入请求队列中requestChannel.sendRequest(req)//取消OP_READ事件selector.mute(receive.source)} }}
KafkaRequestHandlerPool 处理请求
接下来就会通过KafkaRequestHandler线程去处理请求队列中的请求。回到最开始的 startup()，
def startup() {//主要用于处理请求队列里面的请求requestHandlerPool = new KafkaRequestHandlerPool(config.brokerId, socketServer.requestChannel, apis, config.numIoThreads)//...
}新建的KafkaRequestHandlerPool，会在主构造函数创建8个KafkaRequestHandler
class KafkaRequestHandlerPool(val brokerId: Int,val requestChannel: RequestChannel,val apis: KafkaApis,numThreads: Int) extends Logging with KafkaMetricsGroup {val threads = new Array[Thread](numThreads)val runnables = new Array[KafkaRequestHandler](numThreads)//默认启动8个线程，一般情况下可以根据消息的吞吐量去设置这个参数for(i <- 0 until numThreads) {//创建线程runnables(i) = new KafkaRequestHandler(i, brokerId, aggregateIdleMeter, numThreads, requestChannel, apis)threads(i) = Utils.daemonThread("kafka-request-handler-" + i, runnables(i))//线程启动，就会执行run方法threads(i).start()}}
KafkaRequestHandler启动之后就会执行run方法，将客户端的请求交由KafkaAPIs进行最终的处理。
def run() {while(true) {try {var req : RequestChannel.Request = nullwhile (req == null) {val startSelectTime = SystemTime.nanoseconds//获取request对象req = requestChannel.receiveRequest(300)val idleTime = SystemTime.nanoseconds - startSelectTimeaggregateIdleMeter.mark(idleTime / totalHandlerThreads)}//将请求交给KafkaApis进行处理apis.handle(req)} }}
def handle(request: RequestChannel.Request) {//处理生产者发送过来的请求case ApiKeys.PRODUCE => handleProducerRequest(request)}
def handleProducerRequest(request: RequestChannel.Request) {//获取到生产发送过来的请求信息val produceRequest = request.body.asInstanceOf[ProduceRequest]val numBytesAppended = request.header.sizeOf + produceRequest.sizeOf//按照分区的方式去遍历数据val (existingAndAuthorizedForDescribeTopics, nonExistingOrUnauthorizedForDescribeTopics) = produceRequest.partitionRecords.asScala.partition {//对发送过来的数据进行权限等判断。case (topicPartition, _) => authorize(request.session, Describe, new Resource(auth.Topic, topicPartition.topic)) && metadataCache.contains(topicPartition.topic)}//判断是否有写权限。val (authorizedRequestInfo, unauthorizedForWriteRequestInfo) = existingAndAuthorizedForDescribeTopics.partition {case (topicPartition, _) => authorize(request.session, Write, new Resource(auth.Topic, topicPartition.topic))}//把接收的数据追加到磁盘上replicaManager.appendMessages(produceRequest.timeout.toLong,produceRequest.acks,internalTopicsAllowed,authorizedMessagesPerPartition,sendResponseCallback)}
数据存储到磁盘后，调用sendResponseCallback()回调函数处理响应。
def sendResponseCallback(responseStatus: Map[TopicPartition, PartitionResponse]) {//...quotas.produce.recordAndMaybeThrottle(request.session.sanitizedUser,request.header.clientId,numBytesAppended,produceResponseCallback)}}
继续调用回调函数produceResponseCallback()，根据ack的值进行处理
- acks=0：生产者不会等待任何来自broker的确认。
- acks=1(默认)：生产者会等待leader broker接收到消息并确认（但不保证所有副本都已同步）。
- acks=all 或 acks=-1：生产者会等待所有同步副本都确认接收到消息。
def produceResponseCallback(delayTimeMs: Int) {//acks = 0,表示生产者不关心数据的处理结果，所以不需要返回响应信息if (produceRequest.acks == 0) {//...} else {//acks不为0，表明生产者需要响应消息//封装请求头val respHeader = new ResponseHeader(request.header.correlationId)//封装请求体，也就是响应消息val respBody = request.header.apiVersion match {case 0 => new ProduceResponse(mergedResponseStatus.asJava)case version@(1 | 2) => new ProduceResponse(mergedResponseStatus.asJava, delayTimeMs, version)}//将响应消息发送给客户端requestChannel.sendResponse(new RequestChannel.Response(request, new ResponseSend(request.connectionId, respHeader, respBody)))}}
将响应放入processor对应的responseQueue中，默认情况下有3个responseQueue。
def sendResponse(response: RequestChannel.Response) {//先从数组获取Processor对应的队列，再将响应放到这个队列responseQueues(response.processor).put(response)for(onResponse <- responseListeners)onResponse(response.processor)}
接着服务端需要响应客户端，回到processor的run方法
override def run() {//处理响应，并注册OP_WRITE事件processNewResponses()//处理已经发送出去的响应并重新监听OP_READ事件processCompletedSends()}
处理responseQueues中的响应可以分为三种类型：
- NoOpAction：对于不需要返回响应的请求，重新注册OP_READ监听事件。
- SendAction：需要发送响应的情况，接下来注册OP_WRITE监听事件，并最终通过selector.poll()方法将响应结果发送给客户端。
- CloseConnectionAction：需要关闭的响应。
private def processNewResponses() {//通过Process线程的id获取Response对象var curr = requestChannel.receiveResponse(id)while (curr != null) {try {curr.responseAction match {//对于不需要返回响应的请求case RequestChannel.NoOpAction =>curr.request.updateRequestMetrics//重新监听OP_READ事件selector.unmute(curr.request.connectionId)//需要发送响应的情况case RequestChannel.SendAction =>//注册OP_WRITE事件,发送响应sendResponse(curr)// 需要关闭的响应,关闭连接case RequestChannel.CloseConnectionAction =>curr.request.updateRequestMetricsclose(selector, curr.request.connectionId)}} finally {curr = requestChannel.receiveResponse(id)}}}
正常情况下处理已经发送出去的响应，将响应从响应队列中移除，并重新监听OP_READ事件，准备处理客户端的下一个请求。
private def processCompletedSends() {selector.completedSends.asScala.foreach { send =>//移除响应队列的响应val resp = inflightResponses.remove(send.destination).getOrElse {//...}resp.request.updateRequestMetrics()selector.unmute(send.destination)}}
