Kafka网络通信架构：NIO模型与协议处理深度解析

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

/**
 * Kafka网络层采用Reactor模式的核心设计思想：
 * 
 * 1. Reactor（反应器）：SocketServer作为主反应器
 * 2. Acceptor：专门负责接受新连接的线程
 * 3. Processor：处理I/O事件的工作线程
 * 4. Handler：业务逻辑处理器（KafkaApis）
 * 
 * 优势：
 * - 单线程处理多个连接，避免线程切换开销
 * - 事件驱动模型，高效处理I/O密集型操作
 * - 线程数量可控，不随连接数线性增长
 */

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124

/**
 * SocketServer - Kafka的网络服务器核心实现
 * 基于NIO实现高性能的多连接处理
 */
class SocketServer(val config: KafkaConfig,
                  val metrics: Metrics,
                  val time: Time,
                  val credentialProvider: CredentialProvider,
                  val apiVersionManager: ApiVersionManager) extends Closeable with Logging {

  // 网络组件映射
  private val acceptors = mutable.Map[EndPoint, Acceptor]()
  private val processors = mutable.Map[Int, Processor]()
  private val dataPlaneRequestChannel: RequestChannel = _
  private val controlPlaneRequestChannel: RequestChannel = _
  
  // 连接管理
  private var connectionQuotas: ConnectionQuotas = _
  private var memoryPool: MemoryPool = _

  /**
   * 启动网络服务器
   * 按照特定顺序初始化各个网络组件
   */
  def startup(startProcessingRequests: Boolean = true,
             controlPlaneListener: Option[EndPoint] = None,
             config: KafkaConfig = this.config): Unit = {
    this.synchronized {
      info("启动SocketServer")
      
      // 1. 初始化连接配额管理器
      connectionQuotas = new ConnectionQuotas(config, time, metrics)
      
      // 2. 初始化内存池
      memoryPool = if (config.queuedMaxBytes > 0) {
        new SimpleMemoryPool(config.queuedMaxBytes, config.socketRequestMaxBytes, false, metrics)
      } else {
        MemoryPool.NONE
      }
      
      // 3. 创建请求通道
      createDataPlaneRequestChannel()
      createControlPlaneRequestChannel(controlPlaneListener)
      
      // 4. 为每个监听器创建网络处理组件
      val dataPlaneListeners = config.dataPlaneListeners
      val controlPlaneListenerOpt = config.controlPlaneListener
      
      dataPlaneListeners.foreach(createNetworkThreads)
      controlPlaneListenerOpt.foreach(createNetworkThreads)
      
      info(s"启动了 ${dataPlaneListeners.size} 个数据平面监听器")
      
      // 5. 启动请求处理（如果需要）
      if (startProcessingRequests) {
        enableRequestProcessing(Map.empty[ListenerName, CompletableFuture[Void]])
      }
    }
  }

  /**
   * 为指定监听器创建网络处理线程
   * 包括一个Acceptor线程和多个Processor线程
   */
  private def createNetworkThreads(endPoint: EndPoint): Unit = {
    val listenerName = endPoint.listenerName
    val securityProtocol = endPoint.securityProtocol
    
    info(s"为监听器 $listenerName 创建网络线程，地址: ${endPoint.host}:${endPoint.port}")
    
    // 创建Processor线程池
    val processors = new ArrayBuffer[Processor]()
    for (i <- 0 until config.numNetworkThreads) {
      val processor = new Processor(
        id = nextProcessorId(),
        requestChannel = if (endPoint == config.controlPlaneListener.get) 
                          controlPlaneRequestChannel else dataPlaneRequestChannel,
        maxRequestSize = config.socketRequestMaxBytes,
        listenerName = listenerName,
        securityProtocol = securityProtocol,
        config = config,
        metrics = metrics,
        time = time,
        credentialProvider = credentialProvider,
        memoryPool = memoryPool,
        logContext = logContext,
        connectionQuotas = connectionQuotas,
        apiVersionManager = apiVersionManager
      )
      
      processors += processor
      this.processors.put(processor.id, processor)
    }
    
    // 创建Acceptor线程
    val acceptor = new Acceptor(
      endPoint = endPoint,
      requestChannel = if (endPoint == config.controlPlaneListener.get) 
                        controlPlaneRequestChannel else dataPlaneRequestChannel,
      sendBufferSize = config.socketSendBufferBytes,
      recvBufferSize = config.socketReceiveBufferBytes,
      maxRequestSize = config.socketRequestMaxBytes,
      processors = processors.toBuffer,
      connectionQuotas = connectionQuotas,
      time = time,
      logContext = logContext
    )
    
    acceptors.put(endPoint, acceptor)
    
    // 启动线程
    Utils.newThread(s"kafka-socket-acceptor-${listenerName.value}-${endPoint.port}", 
                   acceptor, false).start()
    acceptor.awaitStartup()
    
    processors.foreach { processor =>
      Utils.newThread(s"kafka-network-thread-${config.brokerId}-${listenerName.value}-${processor.id}", 
                     processor, false).start()
      processor.awaitStartup()
    }
    
    info(s"为监听器 $listenerName 创建了 1 个Acceptor和 ${processors.size} 个Processor线程")
  }
}

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162

/**
 * Acceptor - 连接接受器
 * 专门负责接受新的客户端连接并分发给Processor
 */
private[network] class Acceptor(val endPoint: EndPoint,
                               val requestChannel: RequestChannel,
                               val sendBufferSize: Int,
                               val recvBufferSize: Int,
                               val maxRequestSize: Int,
                               processors: Buffer[Processor],
                               connectionQuotas: ConnectionQuotas,
                               time: Time,
                               logContext: LogContext) 
  extends AbstractServerThread(connectionQuotas) with KafkaMetricsGroup {

  // NIO组件
  private val nioSelector = NSelector.open()
  private val serverChannel = openServerSocket(endPoint.host, endPoint.port)
  
  // Processor分发策略
  private val processors = mutable.Set[Processor]() ++= processors
  private var currentProcessorIndex = 0

  /**
   * Acceptor主循环
   * 持续监听新连接并分发给Processor处理
   */
  def run(): Unit = {
    // 注册ServerSocketChannel到Selector，监听ACCEPT事件
    serverChannel.register(nioSelector, SelectionKey.OP_ACCEPT)
    startupComplete()
    
    try {
      while (isRunning) {
        try {
          // 等待连接事件，500ms超时
          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 =>
                    debug(s"接受新连接：${socketChannel.socket().getRemoteSocketAddress}")
                    
                    // 轮询分配给Processor
                    val processor = synchronized {
                      currentProcessorIndex = (currentProcessorIndex + 1) % processors.size
                      processors.toSeq(currentProcessorIndex)
                    }
                    
                    // 将连接移交给Processor处理
                    processor.accept(socketChannel, connectionQuotas, 
                                   maybeBlockingMode = true, maybeBlocked = false)
                  }
                } else {
                  throw new IllegalStateException("未识别的SelectionKey状态")
                }
              } catch {
                case e: Throwable => 
                  error("处理连接事件时发生错误", e)
              }
            }
          }
          
        } catch {
          case e: ControlThrowable => throw e
          case e: Throwable => 
            error("Acceptor线程中发生未预期的错误", e)
        }
      }
    } finally {
      debug("关闭Acceptor，清理资源")
      CoreUtils.swallow(serverChannel.close(), this, Level.ERROR)
      CoreUtils.swallow(nioSelector.close(), this, Level.ERROR)
      shutdownComplete()
    }
  }

  /**
   * 接受新的客户端连接
   * 配置Socket参数并进行连接配额检查
   */
  private def accept(key: SelectionKey): Option[SocketChannel] = {
    val serverSocketChannel = key.channel().asInstanceOf[ServerSocketChannel]
    val socketChannel = serverSocketChannel.accept()
    
    try {
      // 连接配额检查
      connectionQuotas.inc(socketChannel.socket().getInetAddress)
      
      // 配置Socket参数
      socketChannel.configureBlocking(false)
      socketChannel.socket().setTcpNoDelay(true)    // 禁用Nagle算法，降低延迟
      socketChannel.socket().setKeepAlive(true)     // 启用TCP keepalive
      
      // 配置发送缓冲区大小
      if (sendBufferSize != Selectable.USE_DEFAULT_BUFFER_SIZE) {
        socketChannel.socket().setSendBufferSize(sendBufferSize)
      }
      
      // 配置接收缓冲区大小
      if (recvBufferSize != Selectable.USE_DEFAULT_BUFFER_SIZE) {
        socketChannel.socket().setReceiveBufferSize(recvBufferSize)
      }
      
      debug(s"成功配置新连接：${socketChannel.socket().getRemoteSocketAddress}")
      Some(socketChannel)
      
    } catch {
      case e: TooManyConnectionsException =>
        info(s"拒绝连接来自 ${socketChannel.socket.getRemoteSocketAddress}：${e.getMessage}")
        close(socketChannel)
        None
        
      case e: Throwable =>
        error(s"配置新连接失败，来自 ${socketChannel.socket.getRemoteSocketAddress}", e)
        close(socketChannel)
        None
    }
  }

  /**
   * 打开服务器Socket
   * 配置服务器监听Socket的参数
   */
  private def openServerSocket(host: String, port: Int): ServerSocketChannel = {
    val socketAddress = if (host == null || host.trim.isEmpty) {
      new InetSocketAddress(port)
    } else {
      new InetSocketAddress(host, port)
    }
    
    val serverChannel = ServerSocketChannel.open()
    serverChannel.configureBlocking(false)
    
    // 配置Socket选项
    if (config.socketReceiveBufferBytes != Selectable.USE_DEFAULT_BUFFER_SIZE) {
      serverChannel.socket().setReceiveBufferSize(config.socketReceiveBufferBytes)
    }
    
    // 启用地址重用
    serverChannel.socket().setReuseAddress(true)
    
    try {
      serverChannel.socket().bind(socketAddress)
      info(s"成功绑定到地址 $socketAddress")
    } catch {
      case e: SocketException =>
        throw new KafkaException(s"Socket服务器无法绑定到 $socketAddress：${e.getMessage}", e)
    }
    
    serverChannel
  }
}

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347

/**
 * Processor - 网络I/O处理器
 * 使用单线程NIO模型处理多个连接的I/O操作
 */
private[network] class Processor(val id: Int,
                                val requestChannel: RequestChannel,
                                val maxRequestSize: Int,
                                val listenerName: ListenerName,
                                val securityProtocol: SecurityProtocol,
                                val config: KafkaConfig,
                                val metrics: Metrics,
                                val time: Time,
                                val credentialProvider: CredentialProvider,
                                val memoryPool: MemoryPool,
                                val logContext: LogContext,
                                val connectionQuotas: ConnectionQuotas,
                                val apiVersionManager: ApiVersionManager)
  extends AbstractServerThread(connectionQuotas) with KafkaMetricsGroup {

  // NIO选择器和连接管理
  private val selector = createSelector()
  private val newConnections = new ConcurrentLinkedQueue[SocketChannel]()
  private val inflightResponses = mutable.Map[String, RequestChannel.Response]()
  
  // 性能指标
  private val avgIdleMeter = newMeter("NetworkProcessorAvgIdlePercent", "percent", TimeUnit.NANOSECONDS)

  /**
   * Processor主循环
   * 处理网络I/O事件的核心逻辑
   */
  override def run(): Unit = {
    startupComplete()
    
    try {
      while (isRunning) {
        try {
          // 记录空闲时间开始
          val startSelectTime = time.nanoseconds()
          
          // 1. 配置新连接
          configureNewConnections()
          
          // 2. 处理响应
          processNewResponses()
          
          // 3. NIO事件轮询
          poll()
          
          // 4. 处理已完成的接收
          processCompletedReceives()
          
          // 5. 处理已完成的发送
          processCompletedSends()
          
          // 6. 处理断开的连接
          processDisconnected()
          
          // 记录空闲时间
          val endSelectTime = time.nanoseconds()
          avgIdleMeter.mark(endSelectTime - startSelectTime)
          
        } catch {
          case e: ControlThrowable => throw e
          case e: Throwable => 
            error("Processor循环中发生未预期的错误", e)
        }
      }
    } finally {
      debug("关闭Processor {}，清理资源", id)
      closeAll()
      shutdownComplete()
    }
  }

  /**
   * 配置新接受的连接
   * 将新连接注册到Selector并设置认证
   */
  private def configureNewConnections(): Unit = {
    var connectionsProcessed = 0
    
    while (connectionsProcessed < 20 && !newConnections.isEmpty) {
      val socketChannel = newConnections.poll()
      if (socketChannel != null) {
        try {
          debug(s"配置新连接：${socketChannel.socket().getRemoteSocketAddress}")
          
          // 生成连接ID
          val connectionId = generateConnectionId(socketChannel)
          
          // 创建KafkaChannel包装SocketChannel
          val kafkaChannel = buildKafkaChannel(connectionId, socketChannel)
          
          // 注册到Selector
          selector.register(connectionId, kafkaChannel)
          
          connectionsProcessed += 1
          
        } catch {
          case e: Throwable =>
            error("配置新连接时发生错误", e)
            close(socketChannel)
        }
      }
    }
  }

  /**
   * 构建KafkaChannel
   * 根据安全协议配置不同类型的通道
   */
  private def buildKafkaChannel(connectionId: String, 
                               socketChannel: SocketChannel): KafkaChannel = {
    
    try {
      val transportLayer = buildTransportLayer(connectionId, socketChannel)
      val authenticator = buildAuthenticator(connectionId, transportLayer)
      
      new KafkaChannel(
        id = connectionId,
        transportLayer = transportLayer,
        authenticator = authenticator,
        maxReceiveSize = maxRequestSize,
        memoryPool = memoryPool,
        metricGrpPrefix = listenerName.value
      )
      
    } catch {
      case e: Exception =>
        error(s"构建KafkaChannel失败，连接: $connectionId", e)
        throw e
    }
  }

  /**
   * 构建传输层
   * 根据安全协议选择不同的传输层实现
   */
  private def buildTransportLayer(connectionId: String, 
                                 socketChannel: SocketChannel): TransportLayer = {
    securityProtocol match {
      case SecurityProtocol.PLAINTEXT =>
        // 明文传输
        new PlaintextTransportLayer(connectionId, socketChannel)
        
      case SecurityProtocol.SSL =>
        // SSL/TLS加密传输
        buildSslTransportLayer(connectionId, socketChannel)
        
      case SecurityProtocol.SASL_PLAINTEXT =>
        // SASL认证 + 明文传输
        buildSaslTransportLayer(connectionId, socketChannel, false)
        
      case SecurityProtocol.SASL_SSL =>
        // SASL认证 + SSL加密传输
        buildSaslTransportLayer(connectionId, socketChannel, true)
        
      case _ =>
        throw new IllegalArgumentException(s"不支持的安全协议: $securityProtocol")
    }
  }

  /**
   * NIO事件轮询
   * 处理就绪的I/O事件
   */
  private def poll(): Unit = {
    try {
      // 轮询I/O事件，1000ms超时
      selector.poll(1000)
      
    } catch {
      case e: IOException =>
        error("NIO轮询发生I/O异常", e)
    }
  }

  /**
   * 处理已完成的数据接收
   * 将完整的请求提交到RequestChannel
   */
  private def processCompletedReceives(): Unit = {
    selector.completedReceives().asScala.foreach { receive =>
      try {
        val connectionId = receive.source()
        val channel = selector.channel(connectionId)
        
        // 解析请求头
        val requestHeader = RequestHeader.parse(receive.payload())
        val apiKey = requestHeader.apiKey()
        val apiVersion = requestHeader.apiVersion()
        
        debug(s"接收到完整请求：连接={}, API={}, 版本={}, 大小={}字节", 
             connectionId, apiKey, apiVersion, receive.payload().limit())
        
        // 验证API版本
        if (!apiVersionManager.isApiEnabled(apiKey, apiVersion)) {
          // 不支持的API版本，发送错误响应
          val errorResponse = new ApiVersionsResponse(
            new ApiVersionsResponseData()
              .setErrorCode(Errors.UNSUPPORTED_VERSION.code())
          )
          
          sendResponse(RequestChannel.Response.fromRequest(receive, errorResponse))
        } else {
          // 创建请求对象
          val request = new RequestChannel.Request(
            processor = id,
            context = new RequestContext(requestHeader, connectionId, channel.socketAddress,
                                       channel.principal(), listenerName, securityProtocol),
            startTimeNanos = time.nanoseconds(),
            memoryPool = memoryPool,
            buffer = receive.payload(),
            metrics = metrics,
            envelope = None
          )
          
          // 提交到请求通道
          requestChannel.sendRequest(request)
        }
        
      } catch {
        case e: InvalidRequestException =>
          warn(s"接收到无效请求，连接: ${receive.source()}", e)
          closeConnection(receive.source())
          
        case e: Throwable =>
          error(s"处理接收数据时发生异常，连接: ${receive.source()}", e)
          closeConnection(receive.source())
      }
    }
  }

  /**
   * 处理已完成的数据发送
   * 清理已发送完成的响应
   */
  private def processCompletedSends(): Unit = {
    selector.completedSends().asScala.foreach { send =>
      val connectionId = send.destination()
      
      // 从in-flight响应中移除
      inflightResponses.remove(connectionId) match {
        case Some(response) =>
          debug(s"响应发送完成：连接={}, 大小={}字节", 
               connectionId, response.responseSize())
          
          // 更新发送指标
          updateSendMetrics(response)
          
        case None =>
          warn(s"完成发送但未找到对应的响应：连接={}", connectionId)
      }
    }
  }

  /**
   * 处理断开的连接
   * 清理连接相关的资源和状态
   */
  private def processDisconnected(): Unit = {
    selector.disconnected().asScala.foreach { connectionId =>
      info(s"连接断开：{}", connectionId)
      
      // 清理in-flight响应
      inflightResponses.remove(connectionId)
      
      // 更新连接指标
      connectionQuotas.dec(getConnectionAddress(connectionId))
      
      // 清理认证状态
      clearAuthenticationState(connectionId)
    }
  }

  /**
   * 处理新的响应
   * 从RequestChannel获取响应并发送给客户端
   */
  private def processNewResponses(): Unit = {
    var currentResponse: RequestChannel.Response = null
    
    while ({currentResponse = requestChannel.receiveResponse(id); currentResponse != null}) {
      val connectionId = currentResponse.request.context.connectionId
      
      try {
        debug(s"处理新响应：连接={}, API={}", 
             connectionId, currentResponse.request.header.apiKey())
        
        // 检查连接是否仍然有效
        if (selector.channel(connectionId) != null) {
          // 序列化响应
          val responseBuffer = serializeResponse(currentResponse)
          
          // 创建NetworkSend
          val responseSend = new NetworkSend(connectionId, responseBuffer)
          
          // 发送响应
          selector.send(responseSend)
          
          // 添加到in-flight响应跟踪
          inflightResponses.put(connectionId, currentResponse)
          
        } else {
          warn(s"尝试向已断开的连接发送响应：{}", connectionId)
        }
        
      } catch {
        case e: Throwable =>
          error(s"处理响应时发生异常，连接: $connectionId", e)
          closeConnection(connectionId)
      }
    }
  }

  /**
   * 序列化响应
   * 将响应对象转换为ByteBuffer
   */
  private def serializeResponse(response: RequestChannel.Response): ByteBuffer = {
    val responseHeader = response.responseHeader()
    val responseBody = response.responseBody()
    
    // 计算响应大小
    val headerSize = responseHeader.sizeOf()
    val bodySize = responseBody.sizeOf(response.request.header.apiVersion())
    val totalSize = headerSize + bodySize
    
    // 分配缓冲区
    val buffer = ByteBuffer.allocate(4 + totalSize) // 4字节长度前缀
    
    // 写入长度前缀
    buffer.putInt(totalSize)
    
    // 序列化响应头
    responseHeader.writeTo(buffer)
    
    // 序列化响应体
    responseBody.writeTo(buffer, response.request.header.apiVersion())
    
    buffer.flip()
    
    trace(s"序列化响应完成：总大小={}字节", buffer.remaining())
    return buffer
  }
}

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188

/**
 * RequestChannel - 网络层和应用层之间的通信桥梁
 * 实现请求和响应的异步传递
 */
class RequestChannel(val queueSize: Int,
                    val numNetworkThreads: Int,
                    val metricNamePrefix: String,
                    val time: Time,
                    val apiVersionManager: ApiVersionManager) extends KafkaMetricsGroup with Closeable {

  import RequestChannel._
  
  // 请求队列 - 网络线程向应用线程传递请求
  private val requestQueue = new ArrayBlockingQueue[Request](queueSize)
  
  // 响应队列 - 每个Processor对应一个响应队列
  private val responseQueues = (0 until numNetworkThreads).map { i =>
    i -> new ArrayBlockingQueue[Response](queueSize)
  }.toMap
  
  // 监控指标
  private val requestQueueSizeHist = newHistogram(metricNamePrefix + "RequestQueueSize")
  private val responseQueueSizeHist = newHistogram(metricNamePrefix + "ResponseQueueSize")
  private val requestQueueTimeHist = newHistogram(metricNamePrefix + "RequestQueueTimeMs")

  /**
   * 发送请求到处理队列
   * 由Processor线程调用，将请求提交给应用线程处理
   */
  def sendRequest(request: Request): Unit = {
    requestQueueSizeHist.update(requestQueue.size())
    
    val requestEnqueueTimeMs = time.milliseconds()
    request.requestEnqueueTimeMs = requestEnqueueTimeMs
    
    try {
      // 尝试将请求加入队列，最多等待300ms
      val success = requestQueue.offer(request, 300, TimeUnit.MILLISECONDS)
      
      if (!success) {
        // 队列已满，发送服务器繁忙响应
        warn("请求队列已满，拒绝新请求：{}", request.header.apiKey())
        
        val throttledResponse = request.buildErrorResponse(Errors.REQUEST_TIMED_OUT)
        sendResponse(throttledResponse)
      }
      
    } catch {
      case e: InterruptedException =>
        warn("发送请求被中断")
        Thread.currentThread().interrupt()
        
      case e: Throwable =>
        error("发送请求时发生异常", e)
        val errorResponse = request.buildErrorResponse(Errors.UNKNOWN_SERVER_ERROR)
        sendResponse(errorResponse)
    }
  }

  /**
   * 接收请求进行处理
   * 由RequestHandler线程调用，从队列获取待处理请求
   */
  def receiveRequest(timeoutMs: Long): Request = {
    try {
      val request = requestQueue.poll(timeoutMs, TimeUnit.MILLISECONDS)
      
      if (request != null) {
        val queueTime = time.milliseconds() - request.requestEnqueueTimeMs
        requestQueueTimeHist.update(queueTime)
        
        debug(s"接收到请求：API={}, 队列时间={}ms", request.header.apiKey(), queueTime)
      }
      
      request
    } catch {
      case e: InterruptedException =>
        debug("接收请求被中断")
        null
    }
  }

  /**
   * 发送响应到网络线程
   * 由RequestHandler线程调用，将响应发送给指定的Processor
   */
  def sendResponse(response: Response): Unit = {
    val processorId = response.processorId()
    val responseQueue = responseQueues(processorId)
    
    responseQueueSizeHist.update(responseQueue.size())
    
    try {
      // 尝试将响应加入对应Processor的响应队列
      val success = responseQueue.offer(response, 300, TimeUnit.MILLISECONDS)
      
      if (!success) {
        warn("Processor {} 的响应队列已满，丢弃响应", processorId)
        // 关闭对应的连接，避免客户端无限等待
        closeConnection(response.request.context.connectionId)
      } else {
        debug(s"发送响应到Processor {}: API={}", processorId, response.request.header.apiKey())
      }
      
    } catch {
      case e: InterruptedException =>
        warn("发送响应被中断")
        Thread.currentThread().interrupt()
        
      case e: Throwable =>
        error("发送响应时发生异常", e)
        closeConnection(response.request.context.connectionId)
    }
  }

  /**
   * 接收响应进行发送
   * 由Processor线程调用，获取需要发送给客户端的响应
   */
  def receiveResponse(processorId: Int): Response = {
    val responseQueue = responseQueues(processorId)
    responseQueue.poll()
  }
}

/**
 * Request - 请求封装类
 * 包含请求的完整信息和上下文
 */
case class Request(processor: Int,
                  context: RequestContext,
                  startTimeNanos: Long,
                  memoryPool: MemoryPool,
                  buffer: ByteBuffer,
                  metrics: RequestChannel.Metrics,
                  envelope: Option[RequestChannel.Request] = None) extends BaseRequest {

  // 请求生命周期时间戳
  @volatile var requestDequeueTimeNanos: Long = -1L
  @volatile var apiLocalCompleteTimeNanos: Long = -1L
  @volatile var responseCompleteTimeNanos: Long = -1L
  @volatile var responseDequeueTimeNanos: Long = -1L
  @volatile var requestEnqueueTimeMs: Long = -1L

  /**
   * 构建成功响应
   */
  def buildResponse(responseBody: AbstractResponse): Response = {
    Response(
      processor = processor,
      request = this,
      responseBody = responseBody,
      onComplete = None
    )
  }

  /**
   * 构建错误响应
   */
  def buildErrorResponse(error: Errors): Response = {
    val errorResponse = body.getErrorResponse(error)
    buildResponse(errorResponse)
  }

  /**
   * 计算请求处理各阶段的延迟
   */
  def requestLatencyMs(): Double = {
    (responseCompleteTimeNanos - startTimeNanos) / 1000000.0
  }

  def requestQueueTimeMs(): Double = {
    (requestDequeueTimeNanos - startTimeNanos) / 1000000.0
  }

  def apiLocalCompleteTimeMs(): Double = {
    (apiLocalCompleteTimeNanos - requestDequeueTimeNanos) / 1000000.0
  }

  def responseQueueTimeMs(): Double = {
    (responseDequeueTimeNanos - responseCompleteTimeNanos) / 1000000.0
  }

  def responseSendTimeMs(): Double = {
    val endTimeNanos = if (responseDequeueTimeNanos == -1L) time.nanoseconds() else responseDequeueTimeNanos
    (endTimeNanos - responseCompleteTimeNanos) / 1000000.0
  }
}

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210

/**
 * Kafka协议处理器
 * 实现Kafka二进制协议的解析和生成
 */
public class KafkaProtocolHandler {
    
    /**
     * 协议消息格式
     * Kafka使用自定义的二进制协议格式
     * 
     * 消息结构：
     * [length:4][request_header][request_body]
     * 
     * 请求头结构：
     * [api_key:2][api_version:2][correlation_id:4][client_id:string]
     * 
     * 响应头结构：
     * [length:4][correlation_id:4][response_body]
     */
    
    /**
     * 解析请求消息
     * 从ByteBuffer中解析出完整的请求对象
     */
    public static RequestChannel.Request parseRequest(ByteBuffer buffer, 
                                                     String connectionId,
                                                     SecurityProtocol securityProtocol) {
        try {
            // 读取消息长度（已经被NetworkClient读取并验证）
            buffer.rewind();
            
            // 解析请求头
            RequestHeader header = RequestHeader.parse(buffer);
            ApiKeys apiKey = header.apiKey();
            short apiVersion = header.apiVersion();
            
            debug("解析请求：API={}, 版本={}, 关联ID={}", 
                 apiKey, apiVersion, header.correlationId());
            
            // 根据API类型解析请求体
            AbstractRequest requestBody;
            switch (apiKey) {
                case PRODUCE:
                    requestBody = ProduceRequest.parse(buffer, apiVersion);
                    break;
                case FETCH:
                    requestBody = FetchRequest.parse(buffer, apiVersion);
                    break;
                case METADATA:
                    requestBody = MetadataRequest.parse(buffer, apiVersion);
                    break;
                case OFFSET_COMMIT:
                    requestBody = OffsetCommitRequest.parse(buffer, apiVersion);
                    break;
                case OFFSET_FETCH:
                    requestBody = OffsetFetchRequest.parse(buffer, apiVersion);
                    break;
                case JOIN_GROUP:
                    requestBody = JoinGroupRequest.parse(buffer, apiVersion);
                    break;
                case HEARTBEAT:
                    requestBody = HeartbeatRequest.parse(buffer, apiVersion);
                    break;
                default:
                    // 使用反射机制解析未明确处理的API
                    requestBody = parseUsingReflection(apiKey, buffer, apiVersion);
            }
            
            // 创建请求上下文
            RequestContext context = new RequestContext(
                header,
                connectionId,
                InetAddress.getByName("unknown"), // 实际实现中会获取真实地址
                KafkaPrincipal.ANONYMOUS,         // 实际实现中会获取认证主体
                ListenerName.forSecurityProtocol(securityProtocol),
                securityProtocol
            );
            
            return new RequestChannel.Request(
                processor = 0, // 实际值由调用者设置
                context = context,
                startTimeNanos = System.nanoTime(),
                memoryPool = MemoryPool.NONE,
                buffer = buffer,
                metrics = null
            );
            
        } catch (Exception e) {
            error("解析请求时发生异常", e);
            throw new InvalidRequestException("无法解析请求", e);
        }
    }
    
    /**
     * 序列化响应消息
     * 将响应对象序列化为ByteBuffer
     */
    public static ByteBuffer serializeResponse(RequestChannel.Response response) {
        try {
            RequestHeader requestHeader = response.request.header;
            AbstractResponse responseBody = response.responseBody();
            
            // 创建响应头
            ResponseHeader responseHeader = new ResponseHeader(
                requestHeader.correlationId()
            );
            
            // 计算响应大小
            int headerSize = responseHeader.sizeOf();
            int bodySize = responseBody.sizeOf(requestHeader.apiVersion());
            int totalSize = headerSize + bodySize;
            
            // 分配缓冲区（包含4字节长度前缀）
            ByteBuffer buffer = ByteBuffer.allocate(4 + totalSize);
            
            // 写入消息长度
            buffer.putInt(totalSize);
            
            // 序列化响应头
            responseHeader.writeTo(buffer);
            
            // 序列化响应体
            responseBody.writeTo(buffer, requestHeader.apiVersion());
            
            buffer.flip();
            
            debug("序列化响应完成：API={}, 大小={}字节", 
                 requestHeader.apiKey(), buffer.remaining());
            
            return buffer;
            
        } catch (Exception e) {
            error("序列化响应时发生异常", e);
            throw new RuntimeException("响应序列化失败", e);
        }
    }
    
    /**
     * 协议版本协商
     * 处理客户端和服务端的API版本协商
     */
    public static class ProtocolVersionNegotiation {
        
        /**
         * 处理ApiVersions请求
         * 返回服务端支持的所有API版本信息
         */
        public static ApiVersionsResponse handleApiVersionsRequest(ApiVersionsRequest request) {
            ApiVersionsResponseData responseData = new ApiVersionsResponseData();
            
            // 添加所有支持的API版本信息
            for (ApiKeys apiKey : ApiKeys.values()) {
                if (apiKey.isVersionSupported()) {
                    ApiVersionsResponseData.ApiVersion apiVersion = 
                        new ApiVersionsResponseData.ApiVersion()
                            .setApiKey(apiKey.id)
                            .setMinVersion(apiKey.oldestVersion())
                            .setMaxVersion(apiKey.latestVersion());
                    
                    responseData.apiKeys().add(apiVersion);
                }
            }
            
            // 设置服务端信息
            responseData.setThrottleTimeMs(0);
            responseData.setErrorCode(Errors.NONE.code());
            
            debug("返回API版本信息，支持 {} 个API", responseData.apiKeys().size());
            
            return new ApiVersionsResponse(responseData);
        }
        
        /**
         * 验证API版本兼容性
         * 检查客户端请求的API版本是否被服务端支持
         */
        public static boolean isVersionSupported(ApiKeys apiKey, short version) {
            if (!apiKey.isVersionSupported()) {
                return false;
            }
            
            return version >= apiKey.oldestVersion() && version <= apiKey.latestVersion();
        }
        
        /**
         * 获取兼容的API版本
         * 在客户端和服务端版本不匹配时，选择兼容的版本
         */
        public static short getCompatibleVersion(ApiKeys apiKey, 
                                               short requestedVersion, 
                                               short serverMinVersion,
                                               short serverMaxVersion) {
            
            if (requestedVersion >= serverMinVersion && requestedVersion <= serverMaxVersion) {
                // 请求版本在支持范围内
                return requestedVersion;
            } else if (requestedVersion < serverMinVersion) {
                // 请求版本过低，使用服务端最小版本
                warn("API {} 请求版本 {} 低于服务端最小版本 {}，使用最小版本", 
                    apiKey, requestedVersion, serverMinVersion);
                return serverMinVersion;
            } else {
                // 请求版本过高，使用服务端最大版本
                warn("API {} 请求版本 {} 高于服务端最大版本 {}，使用最大版本", 
                    apiKey, requestedVersion, serverMaxVersion);
                return serverMaxVersion;
            }
        }
    }
}

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369