Kafka 源码解析：网络交互模型

class SocketServer(val config: KafkaConfig,
                   val metrics: Metrics,
                   val time: Time,
                   val credentialProvider: CredentialProvider) extends Logging with KafkaMetricsGroup {

    /** 封装服务器对应的多张网卡，kafka 可以同时监听这些 IP 和端口，每个 EndPoint 对应一个 Acceptor */
    private val endpoints: Map[ListenerName, EndPoint] = config.listeners.map(l => l.listenerName -> l).toMap
    /** 每个 Acceptor 对应的 Processor 对应的线程数 */
    private val numProcessorThreads = config.numNetworkThreads
    /** broker 节点上 Processor 线程总数 */
    private val totalProcessorThreads = numProcessorThreads * endpoints.size
    /** 请求队列中缓存的最大请求个数 */
    private val maxQueuedRequests = config.queuedMaxRequests
    /** 每个 IP 允许创建的最大连接数 */
    private val maxConnectionsPerIp = config.maxConnectionsPerIp
    /** 针对特定 IP 指定的允许创建的最大连接数，会覆盖 maxConnectionsPerIp 配置 */
    private val maxConnectionsPerIpOverrides = config.maxConnectionsPerIpOverrides
    /** Processor 线程与 Handler 线程之间交换数据的通道 */
    val requestChannel = new RequestChannel(totalProcessorThreads, maxQueuedRequests)
    /** Acceptor 对象集合，每个 EndPoint 对应一个 Acceptor */
    private[network] val acceptors = mutable.Map[EndPoint, Acceptor]()
    /** Processor 对象集合，封装所有的 Processor 对象 */
    private val processors = new Array[Processor](totalProcessorThreads)
    /** 用于控制每个 IP 上的最大连接数 */
    private var connectionQuotas: ConnectionQuotas = _

    // ... 省略方法定义

}

class RequestChannel(val numProcessors: Int, // Processor 线程总数
                     val queueSize: Int // 请求队列的大小
                    ) extends KafkaMetricsGroup {

    /** 响应监听器列表，当 Handler 往响应队列写回响应数据时唤醒对应的 Processor 线程进行处理 */
    private var responseListeners: List[Int => Unit] = Nil
    /** 请求队列，所有的 Processor 共用一个 */
    private val requestQueue = new ArrayBlockingQueue[RequestChannel.Request](queueSize)
    /** 响应队列，每个 Processor 对应一个响应队列 */
    private val responseQueues = new Array[BlockingQueue[RequestChannel.Response]](numProcessors)

    // ... 省略方法定义

}

def startup() {
    synchronized {

        // 创建控制 IP 最大连接数的 ConnectionQuotas 对象
        connectionQuotas = new ConnectionQuotas(maxConnectionsPerIp, maxConnectionsPerIpOverrides)

        // 指定 socket send buffer 的大小（对应 socket.send.buffer.bytes 配置）
        val sendBufferSize = config.socketSendBufferBytes
        // 指定 socket receive buffer 的大小（对应 socket.receive.buffer.bytes 配置）
        val recvBufferSize = config.socketReceiveBufferBytes
        // 获取 broker 节点 ID
        val brokerId = config.brokerId

        var processorBeginIndex = 0
        // 遍历为每个 EndPoint，创建并绑定对应的 Acceptor 和 Processor
        config.listeners.foreach { endpoint =>
            val listenerName = endpoint.listenerName
            val securityProtocol = endpoint.securityProtocol
            val processorEndIndex = processorBeginIndex + numProcessorThreads

            // 按照指定的 processor 线程数，为每个 EndPoint 创建对应数量的 Processor 对象，
            // 编号区间 [processorBeginIndex, processorEndIndex)
            for (i <- processorBeginIndex until processorEndIndex)
                processors(i) = this.newProcessor(i, connectionQuotas, listenerName, securityProtocol)

            // 为当前 EndPoint 创建并绑定一个 Acceptor 对象
            val acceptor = new Acceptor(endpoint, sendBufferSize, recvBufferSize, brokerId,
                processors.slice(processorBeginIndex, processorEndIndex), connectionQuotas)
            acceptors.put(endpoint, acceptor)

            // 启动 Acceptor 线程
            Utils.newThread(s"kafka-socket-acceptor-$listenerName-$securityProtocol-${endpoint.port}", acceptor, false).start()

            // 主线程等待 Acceptor 线程启动完成
            acceptor.awaitStartup()

            processorBeginIndex = processorEndIndex
        }
    }

    info("Started " + acceptors.size + " acceptor threads")
}

private[kafka] class Acceptor(val endPoint: EndPoint, // 对应的网卡信息
                              val sendBufferSize: Int, // socket send buffer size
                              val recvBufferSize: Int, // socket receive buffer size
                              brokerId: Int, // broker 节点 id
                              processors: Array[Processor], // 绑定的 Processor 线程集合
                              connectionQuotas: ConnectionQuotas // 控制 IP 连接数的对象
                             ) extends AbstractServerThread(connectionQuotas) with KafkaMetricsGroup {

    /** NIO Selector */
    private val nioSelector = NSelector.open()
    /** ServerSocketChannel 对象，监听对应网卡的指定端口 */
    val serverChannel: ServerSocketChannel = this.openServerSocket(endPoint.host, endPoint.port)

    // ... 省略方法定义

}

def run() {
    // 注册监听 OP_ACCEPT 事件
    serverChannel.register(nioSelector, SelectionKey.OP_ACCEPT)
    // 标记当前线程启动完成，以便 SocketServer 能够继续为其它网卡创建并绑定对应的 Acceptor 线程
    this.startupComplete()
    try {
        var currentProcessor = 0 // 当前生效的 processor 编号
        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()
                            // 如果是 OP_ACCEPT 事件，则调用 accept 方法进行处理
                            if (key.isAcceptable)
                                this.accept(key, processors(currentProcessor))
                            else
                                throw new IllegalStateException("Unrecognized key state for acceptor thread.")
                            // 基于轮询算法选择下一个 Processor 处理下一次请求，负载均衡
                            currentProcessor = (currentProcessor + 1) % processors.length
                        } catch {
                            case e: Throwable => error("Error while accepting connection", e)
                        }
                    }
                }
            } catch {
                case e: ControlThrowable => throw e
                case e: Throwable => error("Error occurred", e)
            }
        }
    } finally {
        debug("Closing server socket and selector.")
        this.swallowError(serverChannel.close())
        this.swallowError(nioSelector.close())
        this.shutdownComplete()
    }
}

def accept(key: SelectionKey, processor: Processor) {
    val serverSocketChannel = key.channel().asInstanceOf[ServerSocketChannel]
    // 创建 SocketChannel 对象
    val socketChannel = serverSocketChannel.accept()
    try {
        // 增加对应 IP 上的连接数，如果连接数超过阈值，则抛 TooManyConnectionsException 异常
        connectionQuotas.inc(socketChannel.socket().getInetAddress)
        // 配置 SocketChannel 对象，非阻塞模式
        socketChannel.configureBlocking(false)
        socketChannel.socket().setTcpNoDelay(true)
        socketChannel.socket().setKeepAlive(true)
        if (sendBufferSize != Selectable.USE_DEFAULT_BUFFER_SIZE)
            socketChannel.socket().setSendBufferSize(sendBufferSize)

        // 将 SocketChannel 交给 Processor 进行处理
        processor.accept(socketChannel)
    } catch {
        // 连接数过多，关闭当前通道上的连接，并将连接计数减 1
        case e: TooManyConnectionsException =>
            info("Rejected connection from %s, address already has the configured maximum of %d connections.".format(e.ip, e.count))
            this.close(socketChannel)
    }
}

/** Processor 与 Handler 线程之间传递请求数据的队列 */
val requestChannel: RequestChannel
/** 记录分配给当前 Processor 的待处理的 SocketChannel 对象 */
private val newConnections = new ConcurrentLinkedQueue[SocketChannel]()
/** 缓存未发送给客户端的响应，由于客户端不会进行确认，所以服务端在发送成功之后会将其移除 */
private val inflightResponses = mutable.Map[String, RequestChannel.Response]()

def accept(socketChannel: SocketChannel) {
    // 将 Acceptor 分配的 SocketChannel 对象缓存到同步队列中
    newConnections.add(socketChannel)
    // 唤醒 Processor 线程处理队列
    this.wakeup() // 本质上调用 NIO Server 的 wakeup 方法
}

synchronized {
    // 遍历启动分配给当前 Acceptor 的 Processor 线程
    processors.foreach { processor =>
        Utils.newThread(
            s"kafka-network-thread-$brokerId-${endPoint.listenerName}-${endPoint.securityProtocol}-${processor.id}",
            processor, false).start()
    }
}

override def run() {
    // 标识当前线程启动完成
    this.startupComplete()
    while (isRunning) {
        try {
            // 1. 遍历获取分配给当前 Processor 的 SocketChannel 对象，注册 OP_READ 事件
            this.configureNewConnections()

            // 2. 遍历处理当前 Processor 的响应队列，依据响应类型进行处理
            this.processNewResponses()

            // 3. 发送缓存的响应对象给客户端
            this.poll()

            // 4.
            // 遍历处理 poll 操作放置在 Selector 的 completedReceives 队列中的请求，
            // 封装请求信息为 Request 对象，并记录到请求队列中等待 Handler 线程处理，
            // 同时标记当前 Selector 暂时不再接收新的请求
            this.processCompletedReceives()

            // 5.
            // 遍历处理 poll 操作放置在 Selector 的 completedSends 队列中的请求，
            // 将其从 inflightResponses 集合中移除，并标记当前 Selector 可以继续读取数据
            this.processCompletedSends()

            // 6.
            // 遍历处理 poll 操作放置在 Selector 的 disconnected 集合中的断开的连接，
            // 将连接对应的所有响应从 inflightResponses 中移除，同时更新对应 IP 的连接数
            this.processDisconnected()
        } catch {
            case e: ControlThrowable => throw e
            case e: Throwable =>
                error("Processor got uncaught exception.", e)
        }
    }

    debug("Closing selector - processor " + id)
    // 关闭所有的连接以及选择器
    this.swallowError(closeAll())
    this.shutdownComplete()
}

private def configureNewConnections() {
    while (!newConnections.isEmpty) {
        // 获取待处理 SocketChannel 对象
        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
            // 注册 OP_READ 事件
            selector.register(connectionId, channel)
        } catch {
            // 对于不致命的异常，则捕获并关闭对应的通道
            case NonFatal(e) =>
                val remoteAddress = channel.getRemoteAddress
                this.close(channel)
                error(s"Processor $id closed connection from $remoteAddress", e)
        }
    }
}

private def processNewResponses() {
    // 获取当前 Processor 的响应队列
    var curr = requestChannel.receiveResponse(id)
    while (curr != null) {
        try {
            // 依据响应类型对响应进行处理
            curr.responseAction match {
                // 暂时没有响应需要发送，如果对应的通道未被关闭，则继续注册 OP_READ 事件读取请求数据
                case RequestChannel.NoOpAction =>
                    curr.request.updateRequestMetrics()
                    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) // 注册 OP_READ 事件
                // 当前响应需要发送给请求方
                case RequestChannel.SendAction =>
                    // 发送该响应，并将响应对象记录到 inflightResponses 集合中
                    this.sendResponse(curr)
                // 需要关闭当前连接
                case RequestChannel.CloseConnectionAction =>
                    curr.request.updateRequestMetrics()
                    trace("Closing socket connection actively according to the response code.")
                    // 关闭连接
                    this.close(selector, curr.request.connectionId)
            }
        } finally {
            // 获取下一个待处理的响应
            curr = requestChannel.receiveResponse(id)
        }
    }
}

private def processCompletedReceives() {
    // 遍历处理接收到的请求
    selector.completedReceives.asScala.foreach { receive =>
        try {
            // 获取请求对应的通道
            val openChannel = selector.channel(receive.source)
            // 创建通道对应的 Session 对象，用于权限控制
            val session = {
                // Only methods that are safe to call on a disconnected channel should be invoked on 'channel'.
                val channel = if (openChannel != null) openChannel else selector.closingChannel(receive.source)
                RequestChannel.Session(new KafkaPrincipal(KafkaPrincipal.USER_TYPE, channel.principal.getName), channel.socketAddress)
            }
            // 封装请求信息为 Request 对象
            val req = RequestChannel.Request(
                processor = id,
                connectionId = receive.source,
                session = session,
                buffer = receive.payload,
                startTimeMs = time.milliseconds,
                listenerName = listenerName,
                securityProtocol = securityProtocol)
            // 将请求对象放入请求队列中，等待 Handler 线程处理
            requestChannel.sendRequest(req)
            // 取消注册的 OP_READ 事件，处理期间不再接收新的请求（即不读取新的请求数据）
            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)
        }
    }
}

private def processCompletedSends() {
    // 遍历处理已经完全发送出去的请求
    selector.completedSends.asScala.foreach { send =>
        // 因为当前响应已经发送成功，从 inflightResponses 中移除，不需要客户端确认
        val resp = inflightResponses.remove(send.destination).getOrElse {
            throw new IllegalStateException(s"Send for ${send.destination} completed, but not in `inflightResponses`")
        }
        resp.request.updateRequestMetrics()
        // 注册 OP_READ 事件，继续读取请求数据
        selector.unmute(send.destination)
    }
}

private def processDisconnected() {
    // 遍历处理已经断开的连接
    selector.disconnected.asScala.foreach { connectionId =>
        val remoteHost = ConnectionId.fromString(connectionId).getOrElse {
            throw new IllegalStateException(s"connectionId has unexpected format: $connectionId")
        }.remoteHost
        // 将连接对应的所有响应从 inflightResponses 中移除
        inflightResponses.remove(connectionId).foreach(_.request.updateRequestMetrics())
        // 对应的通道已经被关闭，所以需要减少对应 IP 上的连接数
        connectionQuotas.dec(InetAddress.getByName(remoteHost))
    }
}

override def run() {
    while (true) {
        try {
            var req: RequestChannel.Request = null
            while (req == null) {
                val startSelectTime = time.nanoseconds
                // 从请求队列中获取 Processor 封装的请求
                req = requestChannel.receiveRequest(300)
                val idleTime = time.nanoseconds - startSelectTime
                aggregateIdleMeter.mark(idleTime / totalHandlerThreads)
            }

            // 如果是 AllDone 请求，则退出当前线程
            if (req eq RequestChannel.AllDone) {
                debug("Kafka request handler %d on broker %d received shut down command".format(id, brokerId))
                return
            }
            req.requestDequeueTimeMs = time.milliseconds
            trace("Kafka request handler %d on broker %d handling request %s".format(id, brokerId, req))
            // 处理请求，将响应写回到对应 Processor 的响应队列中，并唤醒 Processor 线程
            apis.handle(req)
        } catch {
            case e: Throwable => error("Exception when handling request", e)
        }
    }
}

def handle(request: RequestChannel.Request) {
    try {
        trace("Handling request:%s from connection %s;securityProtocol:%s,principal:%s".
                format(request.requestDesc(true), request.connectionId, request.securityProtocol, request.session.principal))
        // 依据请求类型分发请求
        ApiKeys.forId(request.requestId) match {
            // 处理 ProduceRequest 请求
            case ApiKeys.PRODUCE => handleProducerRequest(request)
            // 处理 FetchRequest 请求
            case ApiKeys.FETCH => handleFetchRequest(request)
            // 处理 ListOffsetRequest 请求
            case ApiKeys.LIST_OFFSETS => handleOffsetRequest(request)
            // 处理 MetadataRequest 请求
            case ApiKeys.METADATA => handleTopicMetadataRequest(request)
            // 处理 LeaderAndIsrRequest 请求
            case ApiKeys.LEADER_AND_ISR => handleLeaderAndIsrRequest(request)
            // 处理 StopReplicaRequest 请求
            case ApiKeys.STOP_REPLICA => handleStopReplicaRequest(request)
            // 处理 UpdateMetadataRequest 请求
            case ApiKeys.UPDATE_METADATA_KEY => handleUpdateMetadataRequest(request)
            // 处理 ControlledShutdownRequest 请求
            case ApiKeys.CONTROLLED_SHUTDOWN_KEY => handleControlledShutdownRequest(request)
            // 处理 OffsetCommitRequest 请求
            case ApiKeys.OFFSET_COMMIT => handleOffsetCommitRequest(request)
            // 处理 OffsetFetchRequest 请求
            case ApiKeys.OFFSET_FETCH => handleOffsetFetchRequest(request)
            // 处理 GroupCoordinatorRequest 请求
            case ApiKeys.GROUP_COORDINATOR => handleGroupCoordinatorRequest(request)
            // 处理 JoinGroupRequest 请求
            case ApiKeys.JOIN_GROUP => handleJoinGroupRequest(request)
            // 处理 HeartbeatRequest 请求
            case ApiKeys.HEARTBEAT => handleHeartbeatRequest(request)
            // 处理 LeaveGroupRequest 请求
            case ApiKeys.LEAVE_GROUP => handleLeaveGroupRequest(request)
            // 处理 SyncGroupRequest 请求
            case ApiKeys.SYNC_GROUP => handleSyncGroupRequest(request)
            // 处理 DescribeGroupsRequest 请求
            case ApiKeys.DESCRIBE_GROUPS => handleDescribeGroupRequest(request)
            // 处理 ListGroupsRequest 请求
            case ApiKeys.LIST_GROUPS => handleListGroupsRequest(request)
            // 处理 SaslHandshakeRequest 请求
            case ApiKeys.SASL_HANDSHAKE => handleSaslHandshakeRequest(request)
            // 处理 ApiVersionsRequest 请求
            case ApiKeys.API_VERSIONS => handleApiVersionsRequest(request)
            // 处理 CreateTopicsRequest 请求
            case ApiKeys.CREATE_TOPICS => handleCreateTopicsRequest(request)
            // 处理 DeleteTopicsRequest 请求
            case ApiKeys.DELETE_TOPICS => handleDeleteTopicsRequest(request)
            case requestId => throw new KafkaException("Unknown api code " + requestId)
        }
    } catch {
        // ... 省略异常处理
    } finally
        request.apiLocalCompleteTimeMs = time.milliseconds
}