2.2.1 服务端使用接收器接受客户端的连接

╰+哭是因爲堅強的太久メ 2022-11-13 04:13 134阅读 0赞

> 紧接着[上篇][Link 1]

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

### 2.2 服务端网络连接 ###

> 服务端网络连接KafkaServer是Kafka服务端的主类，KafkaServer中和网络层有关的服务组件包括SocketServer、KafkaApis和KafkaRequestHandlerPool。后两者都使用了SocketServer暴露出来的请求通道（requestChannel）来处理网络请求。KafkaServer中和服务端相关的组件还有很多，不过由于本章主要关注网络层，所以暂时不关注其他功能组件。相关代码如下：

def startup() { 
        try { 
          info("starting")
    
          if(isShuttingDown.get)
            throw new IllegalStateException("Kafka server is still shutting down, cannot re-start!")
    
          if(startupComplete.get)
            return
    
          val canStartup = isStartingUp.compareAndSet(false, true)
          if (canStartup) { 
            brokerState.newState(Starting)
    
            /* start scheduler */
            kafkaScheduler.startup()
    
            /* setup zookeeper */
            zkUtils = initZk()
    
            /* Get or create cluster_id */
            _clusterId = getOrGenerateClusterId(zkUtils)
            info(s"Cluster ID = $clusterId")
    
            /* generate brokerId */
            config.brokerId =  getBrokerId
            this.logIdent = "[Kafka Server " + config.brokerId + "], "
    
            /* create and configure metrics */
            val reporters = config.getConfiguredInstances(KafkaConfig.MetricReporterClassesProp, classOf[MetricsReporter],
                Map[String, AnyRef](KafkaConfig.BrokerIdProp -> (config.brokerId.toString)).asJava)
            reporters.add(new JmxReporter(jmxPrefix))
            val metricConfig = KafkaServer.metricConfig(config)
            metrics = new Metrics(metricConfig, reporters, time, true)
    
            /* register broker metrics */
            _brokerTopicStats = new BrokerTopicStats
    
            quotaManagers = QuotaFactory.instantiate(config, metrics, time)
            notifyClusterListeners(kafkaMetricsReporters ++ reporters.asScala)
    
            /* start log manager */
            logManager = LogManager(config, zkUtils, brokerState, kafkaScheduler, time, brokerTopicStats)
            logManager.startup()
    
            metadataCache = new MetadataCache(config.brokerId)
            credentialProvider = new CredentialProvider(config.saslEnabledMechanisms)
    
            socketServer = new SocketServer(config, metrics, time, credentialProvider)
            socketServer.startup()
    
            /* start replica manager */
            replicaManager = createReplicaManager(isShuttingDown)
            replicaManager.startup()
    
            /* start kafka controller */
            kafkaController = new KafkaController(config, zkUtils, time, metrics, threadNamePrefix)
            kafkaController.startup()
    
            adminManager = new AdminManager(config, metrics, metadataCache, zkUtils)
    
            /* start group coordinator */
            // Hardcode Time.SYSTEM for now as some Streams tests fail otherwise, it would be good to fix the underlying issue
            groupCoordinator = GroupCoordinator(config, zkUtils, replicaManager, Time.SYSTEM)
            groupCoordinator.startup()
    
            /* start transaction coordinator, with a separate background thread scheduler for transaction expiration and log loading */
            // Hardcode Time.SYSTEM for now as some Streams tests fail otherwise, it would be good to fix the underlying issue
            transactionCoordinator = TransactionCoordinator(config, replicaManager, new KafkaScheduler(threads = 1, threadNamePrefix = "transaction-log-manager-"), zkUtils, metrics, metadataCache, Time.SYSTEM)
            transactionCoordinator.startup()
    
            /* Get the authorizer and initialize it if one is specified.*/
            authorizer = Option(config.authorizerClassName).filter(_.nonEmpty).map {  authorizerClassName =>
              val authZ = CoreUtils.createObject[Authorizer](authorizerClassName)
              authZ.configure(config.originals())
              authZ
            }
    
            /* start processing requests */
            apis = new KafkaApis(socketServer.requestChannel, replicaManager, adminManager, groupCoordinator, transactionCoordinator,
              kafkaController, zkUtils, config.brokerId, config, metadataCache, metrics, authorizer, quotaManagers,
              brokerTopicStats, clusterId, time)
    
            requestHandlerPool = new KafkaRequestHandlerPool(config.brokerId, socketServer.requestChannel, apis, time,
              config.numIoThreads)
    
            Mx4jLoader.maybeLoad()
    
            /* start dynamic config manager */
            dynamicConfigHandlers = Map[String, ConfigHandler](ConfigType.Topic -> new TopicConfigHandler(logManager, config, quotaManagers),
                                                               ConfigType.Client -> new ClientIdConfigHandler(quotaManagers),
                                                               ConfigType.User -> new UserConfigHandler(quotaManagers, credentialProvider),
                                                               ConfigType.Broker -> new BrokerConfigHandler(config, quotaManagers))
    
            // Create the config manager. start listening to notifications
            dynamicConfigManager = new DynamicConfigManager(zkUtils, dynamicConfigHandlers)
            dynamicConfigManager.startup()
    
            /* tell everyone we are alive */
            val listeners = config.advertisedListeners.map {  endpoint =>
              if (endpoint.port == 0)
                endpoint.copy(port = socketServer.boundPort(endpoint.listenerName))
              else
                endpoint
            }
            kafkaHealthcheck = new KafkaHealthcheck(config.brokerId, listeners, zkUtils, config.rack,
              config.interBrokerProtocolVersion)
            kafkaHealthcheck.startup()
    
            // Now that the broker id is successfully registered via KafkaHealthcheck, checkpoint it
            checkpointBrokerId(config.brokerId)
    
            brokerState.newState(RunningAsBroker)
            shutdownLatch = new CountDownLatch(1)
            startupComplete.set(true)
            isStartingUp.set(false)
            AppInfoParser.registerAppInfo(jmxPrefix, config.brokerId.toString)
            info("started")
          }
        }
        catch { 
          case e: Throwable =>
            fatal("Fatal error during KafkaServer startup. Prepare to shutdown", e)
            isStartingUp.set(false)
            shutdown()
            throw e
        }
      }

如图2-17所示，SocketServer主要关注网络层的通信协议，具体的业务处理逻辑则交给KafkaRequestHandler和KafkaApis来完成。SocketServer和这两个组件一起完成一次请求处理的具体步骤如下。

(1)客户端发送的请求被接收器（Acceptor)转发给处理器（processor)处理。  
(2)处理器将请求放到请求通道（RequestChannel)的全局请求队列中。  
(3)KafkaRequestHandler取州请求通道中的客户端请求。  
(4)调用KafkaAp1s进行业务逻辑处理。  
(5)KafkaApis将响应结果发送给请求通道中与处理器对应的响应队列。  
(6)处理器从对应的响应队列中取出响应结果。  
(7)处理器将响应结果返回给客户端，客户端请求处理完毕。

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

#### 2.2.1服务端使用接收器接受客户端的连接 ####

> SocketServer是一个NIO服务，它会启动一个接收器线程（Acceptor）和多个处理器（processor）。NIO服务用一个接收器线程负责接收所有的客户端连接请求，并将接收到的请求分发给不同的处理器处理。这是一种典型的Reactor模式，因为服务端要接收多个客户端的连接和请求。

Reactor模式的设计思想，实际上是将连接部分和请求部分用不同的线程来处理，这样请求的处理不会阻塞不断到来的连接。否则，如果服务端为每个客户端都维护一个连接，不仅会耗光服务端的资隙，而且会降低服务端的性能。使用Reactor模式并结合选择器管理多个客户端的网络连接，可以减少线程之间的上下文切换和资源的开销。相关代码如下：

class SocketServer(val config: KafkaConfig, val metrics: Metrics, val time: Time, val credentialProvider: CredentialProvider) extends Logging with KafkaMetricsGroup { 
    
      private val endpoints = config.listeners.map(l => l.listenerName -> l).toMap
      private val numProcessorThreads = config.numNetworkThreads
      private val maxQueuedRequests = config.queuedMaxRequests
      private val totalProcessorThreads = numProcessorThreads * endpoints.size
    
      private val maxConnectionsPerIp = config.maxConnectionsPerIp
      private val maxConnectionsPerIpOverrides = config.maxConnectionsPerIpOverrides
    
      this.logIdent = "[Socket Server on Broker " + config.brokerId + "], "
    
      val requestChannel = new RequestChannel(totalProcessorThreads, maxQueuedRequests)
      private val processors = new Array[Processor](totalProcessorThreads)
    
      private[network] val acceptors = mutable.Map[EndPoint, Acceptor]()
      private var connectionQuotas: ConnectionQuotas = _
      …………
      }

SocketServer的启动方法中先创建了处理器，然后才创建接收器，并把事先创建好的处理器作为接收器类的参数。当服务端接收到一个客户端连接请求时，接收器就可以决定要将连接转发给其中的一个处理器处理。由于服务端资源有限，处理器数量不会很多，而客户端连接则成千上万，所以一个处理器会管理多个客户端的连接。

接收器的工作职责很简单只管接受客户端的连接请求，并创建和客户端通信的SocketChannel。具体在这个SocketChannel上发生的读写操作者I~和接收器无关，因为它已经把创建好的SocketChannel转交给了处理器，那么处理器就应该全权负责这个通道上的操作了。相关代码如下：

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 { 
    
      private val nioSelector = NSelector.open()
      val serverChannel = openServerSocket(endPoint.host, endPoint.port)
    
      this.synchronized { 
        processors.foreach {  processor =>
          Utils.newThread(s"kafka-network-thread-$brokerId-${endPoint.listenerName}-${endPoint.securityProtocol}-${processor.id}",
            processor, false).start()
        }
      }
    
      /** * Accept loop that checks for new connection attempts */
      def run() { 
        serverChannel.register(nioSelector, SelectionKey.OP_ACCEPT)
        startupComplete()
        try { 
          var currentProcessor = 0
          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()
                    if (key.isAcceptable)
                      accept(key, processors(currentProcessor))
                    else
                      throw new IllegalStateException("Unrecognized key state for acceptor thread.")
    
                    // round robin to the next processor thread
                    currentProcessor = (currentProcessor + 1) % processors.length
                  } catch { 
                    case e: Throwable => error("Error while accepting connection", e)
                  }
                }
              }
            }
            catch { 
              // We catch all the throwables to prevent the acceptor thread from exiting on exceptions due
              // to a select operation on a specific channel or a bad request. We don't want
              // the broker to stop responding to requests from other clients in these scenarios.
              case e: ControlThrowable => throw e
              case e: Throwable => error("Error occurred", e)
            }
          }
        } finally { 
          debug("Closing server socket and selector.")
          swallowError(serverChannel.close())
          swallowError(nioSelector.close())
          shutdownComplete()
        }
      }

服务端的接收器负责接受客户端的连接，就必须要将它和客户端结合起来分析才比较清楚。接收器线程启动时就注册了OP\_ACCEPT事件，Java版本的生产者客户端调用connect()方法连接服务端时，接收器线程的选择器就会监听到OP\_ACCEPT事件。服务端对OP\_ACCEPT事件的处理是：获取绑定到选择键上的ServerSocketChannel，调用它的accept()方法，在服务端就会生成一个和客户端连接的网络通道。

如图2-18所示，服务端接受客户端连接后，会创建对应的SocketChannel来和客户端通信。而客户端和l服务端建立连接的Kafka通道实际上也是一个SocketChannel用来和l服务端通信。客户端和l服务端对于各向的通道都使用了选择器模式，从客户端建立连接到服务端接受连接的步骤如下。  
(l)服务端的ServerSocketChannel向选择器注册OP\_ACCEPT事件。  
(2)客户端向选择器注册OP\_CONNECT事件，并调用SocketChannel.connect()连接服务端。  
(3)服务端的选择器监听到客户端的连接事件，接受客户端的连接。  
(4)服务端使用ServerSocketChannel.accept()创建和客户端通信的SocketChannel。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 1]  
客户端和服务端的其他事件(比如读写)也都是类似的，都要先注册相应的事件，然后选择器才有可能监听到某种类型的事件。只不过，客户端的选择器会处理连接／读写事件，而接收器只处理连接事件，读写事件交给了处理器。不过总的来说，客户端和服务端的事件都是对应的，客户端连接OP\_CONNECT对应服务端接受OP\_ACCEPT，客户端写入OP\_WRITE对应服务端读取OP\_READ，服务端写入OP\_WRITE对应客户端读取OP\_READ。这是因为通信必须是双方一起参与，任何一方只发送不接收，或者只接收不发送，都是不可行的。

[Link 1]: https://waterwang.blog.csdn.net/article/details/115188594
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70]: /images/20221022/1b911e945e9f4456bc816e7e2b150476.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1dBTlRBV0FZMzE0_size_16_color_FFFFFF_t_70 1]: /images/20221022/f023728ba8864cc6a27d374f222631ed.png