2.2.3 请求通道的请求队列和响应队列

以你之姓@ 2023-01-12 15:53 105阅读 0赞

--------------------

#### 2.2.3 请求通道的请求队列和响应队列 ####

创建SocketServer也会创建一个请求通道（RequestChannel），在KafkaServer中，会将SocketServer的请求通道传给Kafka请求处理线程（KafkaRequestHandler，下文简称“请求处理线程”）和KafkaApis。在上一节中客户端的请求已经到达服务端的处理器（processor），那么请求通道就是处理器与请求处理线程和KafkaApis交换数据的地方：如果处理器往请求通道添加请求，请求处理线程和KafkaApis都可以获取到请求通道中的请束；如果请求处理线程和KafkaApis往请求通道添加响应，处理器也可以从请求通道获取晌应。

处理器会将客户端发送的请求放到全局的请求队列（requestQueue）中，供请求处理线程获取，请求处理线程会将请求转发给KafkaApis处理。最后KafkaApis会将处理完成的响应结果放到响应队列（responseQueue）中，供处理器获取后发送给客户端。相关代码如下：

requestChannel.addResponseListener(id => processors(id).wakeup())
    
    class RequestChannel(val numProcessors: Int, val queueSize: Int) extends KafkaMetricsGroup { 
      private var responseListeners: List[(Int) => Unit] = Nil
      private val requestQueue = new ArrayBlockingQueue[RequestChannel.Request](queueSize)
      private val responseQueues = new Array[BlockingQueue[RequestChannel.Response]](numProcessors)
      for(i <- 0 until numProcessors)
        responseQueues(i) = new LinkedBlockingQueue[RequestChannel.Response]()
    
      newGauge(
        "RequestQueueSize",
        new Gauge[Int] { 
          def value = requestQueue.size
        }
      )
    
      newGauge("ResponseQueueSize", new Gauge[Int]{ 
        def value = responseQueues.foldLeft(0) { (total, q) => total + q.size()}
      })
    
      for (i <- 0 until numProcessors) { 
        newGauge("ResponseQueueSize",
          new Gauge[Int] { 
            def value = responseQueues(i).size()
          },
          Map("processor" -> i.toString)
        )
      }
    
      /** 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)
      }
    
      /** Send a response back to the socket server to be sent over the network */
      def sendResponse(response: RequestChannel.Response) { 
        responseQueues(response.processor).put(response)
        for(onResponse <- responseListeners)
          onResponse(response.processor)
      }
    
      /** Get the next request or block until specified time has elapsed */
      def receiveRequest(timeout: Long): RequestChannel.Request =
        requestQueue.poll(timeout, TimeUnit.MILLISECONDS)
    
      /** Get the next request or block until there is one */
      def receiveRequest(): RequestChannel.Request =
        requestQueue.take()
    
      /** Get a response for the given processor if there is one */
      def receiveResponse(processor: Int): RequestChannel.Response = { 
        val response = responseQueues(processor).poll()
        if (response != null)
          response.request.responseDequeueTimeNanos = Time.SYSTEM.nanoseconds
        response
      }
    
      def addResponseListener(onResponse: Int => Unit) { 
        responseListeners ::= onResponse
      }
    
      def shutdown() { 
        requestQueue.clear()
      }
    }

如图2-20所示，因为请求通道保存了请求和响应两种类型的队列，它的各个方法中关于请求和响应的接收和发送是有顺序的：发送请求→接收请求→发送n向应→接收响应。

(1)sendRequest()：处理器接收到客户端请求后，将请求放入请求队列。  
(2)receiveRequest()：请求处理线程从队列中获取请求，并交给KafkaApis处理。  
(3)sendResponse():KafkaApis处理完，将响应结果放入响应队列。  
(4)receiveResponse()：处理器从响应队列中获取响应结果发送给客户端。

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

上面只是一个请求和响应在请求通道上的调用顺序，下面以服务端同时处理多个客户端请求为例，并结合其他相关的组件，来说明处理器将请求放入请求通道，一直到从请求通道获取响应的过程(图中的编号和图2-20编号的含义相同)。如图2-21所示，由于一个SocketServer有多个处理器，每个处理器都负责一部分客户端的请求。如果请求l发送给处理器l，那么请求l对应的响应也只能发送给处理器l，所以每个处理器都有一个响应队列。虽然请求队列是所有处理器全局共享的，不过会有多个请求处理线程同时处理请求队列中的客户端请求。假设处理器3有两个客户端请求，这两个请求进入全局的请求队列后可能被不同的请求处理线程处理，最后KafkaApis会将这两个请求的响应都放入处理器3对应的响应队列中。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 1]  
从图2-21处理器使用请求通道的方式也可以看到，处理器的processCollpletedReceives()会往请求通道的请求队列添加请求，ProcessNewResponses()则从请求通道的响应队列获取响应。与之相对应的获取请求和添加响应的操作，则属于请求处理线程(KafkaRequestHandler)和KafkaApis的功能。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70]: /images/20221022/1241fa0271004052b95104a0bc6d7597.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 1]: /images/20221022/aa1de411e0644203bf1ca12efec71ce6.png