2.2.2 处理器使用选择器的轮询处理网络请求

港控/mmm° 2023-01-12 15:53 16阅读 0赞

#### 2.2.2 处理器使用选择器的轮询处理网络请求 ####

接收器采用Round-Robin的方式分配客户端的SocketChannel给多个处理器，每个处理器都会有多个SocketChannel，对应多个客户端连接。就像NetworkClient中用一个选择器管理多个服务端的连接，服务端的每个处理器也都使用一个选择器管理多个客户端连接。处理器接受一个新的SocketChannel通道连接时，先将其放入阻塞队列，然后唤醒选择器线程开始工作。

回顾客户端连接服务端时会创建Kafka通道，这里服务端的处理器也会为SocketChannel创建一个Kafka通道。configureNewConnections()方法会为SocketChannel注册读事件，创建Kafka通道，并将Kafka通道＃II定至UsocketChannel的选择键上。相关代码如下：

private[kafka] class Processor(val id: Int,
                                   time: Time,
                                   maxRequestSize: Int,
                                   requestChannel: RequestChannel,
                                   connectionQuotas: ConnectionQuotas,
                                   connectionsMaxIdleMs: Long,
                                   listenerName: ListenerName,
                                   securityProtocol: SecurityProtocol,
                                   config: KafkaConfig,
                                   metrics: Metrics,
                                   credentialProvider: CredentialProvider) extends AbstractServerThread(connectionQuotas) with KafkaMetricsGroup { 
    
    …………
    }

客户端的NetworkClient和服务端的处理器都使用相同的选择器类（Selector）进行轮询。发送请求和接收响应，都是通过选择器的轮询才会触发。在轮询之前，客户端需要准备待发送的请求(RequestSend）；服务端需要准备待发送的响应（ResponseSend），轮询之后才执行完成的发送和接收。如果没有轮询，发送和接收就不会被完成，因为没有轮询的话，就不会发生读写操作。

处理器使用选择器的方式不仅和INetworkClient类似，两者在轮询时的一些数据结构也类似。比如NetworkClient在调用Selectorsend()准备发送请求时，将客户端请求加入inFlightRequests队列；而处理器在准备发送响应时，将响应加入inflightResponses。相关代码如下：

public List<ClientResponse> poll(long timeout, long now) { 
            if (!abortedSends.isEmpty()) { 
                // If there are aborted sends because of unsupported version exceptions or disconnects,
                // handle them immediately without waiting for Selector#poll.
                List<ClientResponse> responses = new ArrayList<>();
                handleAbortedSends(responses);
                completeResponses(responses);
                return responses;
            }
    
            long metadataTimeout = metadataUpdater.maybeUpdate(now);
            try { 
                this.selector.poll(Utils.min(timeout, metadataTimeout, requestTimeoutMs));
            } catch (IOException e) { 
                log.error("Unexpected error during I/O", e);
            }
    
            // process completed actions
            long updatedNow = this.time.milliseconds();
            List<ClientResponse> responses = new ArrayList<>();
            handleCompletedSends(responses, updatedNow);
            handleCompletedReceives(responses, updatedNow);
            handleDisconnections(responses, updatedNow);
            handleConnections();
            handleInitiateApiVersionRequests(updatedNow);
            handleTimedOutRequests(responses, updatedNow);
            completeResponses(responses);
    
            return responses;
        }

选择器的轮询操作最后会返回collpletedReceives和collpletedSends，分别表示已经完成的接收和发送。对于客户端的NetworkClient,completedReceives表示已经完成的响应接收，collpletedSends表示已经完成的请求发送。对于服务端的处理器，collpletedReceives表示服务端成功读取客户端发送的请求，collpletedSends表示服务端成功发送给客户端的响应结果。相关代码如下：

private def processCompletedReceives() { 
        selector.completedReceives.asScala.foreach {  receive =>
          try { 
            val openChannel = selector.channel(receive.source)
            // Only methods that are safe to call on a disconnected channel should be invoked on 'openOrClosingChannel'.
            val openOrClosingChannel = if (openChannel != null) openChannel else selector.closingChannel(receive.source)
            val session = RequestChannel.Session(new KafkaPrincipal(KafkaPrincipal.USER_TYPE, openOrClosingChannel.principal.getName), openOrClosingChannel.socketAddress)
    
            val req = RequestChannel.Request(processor = id, connectionId = receive.source, session = session,
              buffer = receive.payload, startTimeNanos = time.nanoseconds,
              listenerName = listenerName, securityProtocol = securityProtocol)
            requestChannel.sendRequest(req)
            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)
          }
        }
      }

如图2-19所示，客户端和服务端的交互都是通过各向的选择器轮询所驱动。结合客户端和服务端以及选择器的轮询，把一个完整的请求和响应过程串联起来的步骤如下。  
（1）客户端完成请求的发送，服务端轮询到客户端发送的请求。  
(2）服务端接收完客户端发送的请求，进行业务处理，并准备好响应结果准备发送。  
(3）服务端完成响应的发送，客户端轮询到服务端发送的响应。  
(4）客户端接收完服务端发送的响应，整个流程结束。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70]

现在，服务端网络相关的代码已经基本上分析完毕，不过我们并没有看到在处理器中，客户端发送的请求是如何交给业务逻辑，以及业务逻辑处理完后如何调用处理器发送响应给客户端的，这中间其实还有一个重要的数据结构是服务端的请求通道（RequestChannel）。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70]: /images/20221022/9b5f417ed7f742128117c8cc320c42cb.png