非法网站开发是什么意思,做网站租服务器,息烽县抖音seo推广,企业邮箱在哪里查整体架构如下所示#xff1a; responseQueue不在RequestChannel中#xff0c;在Processor中#xff0c;每个Processor内部有一个responseQueue
客户端发送的请求被Acceptor转发给Processor处理处理器将请求放到RequestChannel的requestQueue中KafkaRequestHandler取出reque…整体架构如下所示 responseQueue不在RequestChannel中在Processor中每个Processor内部有一个responseQueue
客户端发送的请求被Acceptor转发给Processor处理处理器将请求放到RequestChannel的requestQueue中KafkaRequestHandler取出requestQueue中的请求调用KafkaApis进行业务逻辑处理KafkaApis将响应结果放到对应的Processor的responseQueue中processor从responseQueue中取出响应结果processor将响应结果返回给客户端
KafkaServer是Kafka服务端的主类KafkaServer中和网络成相关的服务组件包括SocketServer、KafkaApis和KafkaRequestHandlerPool。SocketServer主要关注网络层的通信协议具体的业务处理逻辑则交给KafkaRequestHandler和KafkaApis来完成。
class KafkaServer(val config: KafkaConfig) {def startup() {socketServer new SocketServer(config, metrics, time, credentialProvider)socketServer.startup(startupProcessors false)/* start processing requests */apis new KafkaApis(socketServer.requestChannel, ...)requestHandlerPool new KafkaRequestHandlerPool(config.brokerId, ...)}}SocketServer def startup(startupProcessors: Boolean true) {this.synchronized {...createAcceptorAndProcessors(config.numNetworkThreads, config.listeners)if (startupProcessors) {startProcessors()}}private def createAcceptorAndProcessors(processorsPerListener: Int,endpoints: Seq[EndPoint]): Unit synchronized {...endpoints.foreach { endpoint ...val acceptor new Acceptor(endpoint, ...)addProcessors(acceptor, endpoint, processorsPerListener)KafkaThread.nonDaemon(skafka-socket-acceptor-$listenerName-$securityProtocol-${endpoint.port}, acceptor).start()acceptor.awaitStartup()acceptors.put(endpoint, acceptor)}}可以看出SocketServer.startup()中会根据listener的个数创建相同个数的acceptor每个acceptor关联数个processor。这是一种典型的Reactor模式acceptor负责与客户端建立连接并将连接分发给processorprocessor负责所分连接后续的所有读写交互。
Acceptor def run() {serverChannel.register(nioSelector, SelectionKey.OP_ACCEPT)startupComplete()try {var currentProcessor 0while (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.nextiter.remove()if (key.isAcceptable) {val processor synchronized {currentProcessor currentProcessor % processors.sizeprocessors(currentProcessor)}accept(key, processor)} elsethrow new IllegalStateException(Unrecognized key state for acceptor thread.)// round robin to the next processor thread, mod(numProcessors) will be done latercurrentProcessor currentProcessor 1} 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 dont want// the broker to stop responding to requests from other clients in these scenarios.case e: ControlThrowable throw ecase e: Throwable error(Error occurred, e)}}} finally {debug(Closing server socket and selector.)CoreUtils.swallow(serverChannel.close(), this, Level.ERROR)CoreUtils.swallow(nioSelector.close(), this, Level.ERROR)shutdownComplete()}}上面是Acceptor的run()方法可以看出Acceptor在通道上注册了SelectionKey.OP_ACCEPT事件OP_READ、OP_WRITE、OP_CONNECT、OP_ACCEPT客户端监听OP_CONNECT事件负责发起连接服务端监听OP_CONNECT事件负责建立连接负责与客户端建立连接。并将建立的连接通过轮询的方式指派给processor。
Processor
每个Processor都会分到数个与客户端的连接。Processor的处理逻辑如下所示 override def run() {startupComplete()try {while (isRunning) {try {// 在新分到的客户端连接上注册OP_READ事件configureNewConnections()// 从responseQueue中取响应赋值给KafkaChannel的send等待poll时发送processNewResponses()// selector轮询各种事件读取请求或者发送响应poll()// 封装selector.completedReceives中的请求放入requestQueueprocessCompletedReceives()// 处理selector.completedSends响应移除inflightResponses中的记录执行响应的回调函数processCompletedSends()processDisconnected()} catch {...}}} finally {...}}Processor线程的名字中有kafka-network字样可以通过jstack -l pid | grep kafka-network进行筛选。
KafkaRequestHandlerPool
KafkaServer会创建请求处理线程池KafkaRequestHandlerPool在请求处理线程池中会创建并启动多个请求处理线程KafkaRequestHandler。KafkaRequestHandler会获取RequestChannel.requestQueue中的请求进行处理在内部实际处理会交给KafkaApis完成。
class KafkaRequestHandlerPool(val brokerId: Int, ...) {...for (i - 0 until numThreads) {createHandler(i)}def createHandler(id: Int): Unit synchronized {runnables new KafkaRequestHandler(..., requestChannel, apis, time)KafkaThread.daemon(kafka-request-handler- id, runnables(id)).start()}
}KafkaRequestHandler的run()方法如下
class KafkaRequestHandler(id: Int,...) extends Runnable with Logging {...def run() {while (!stopped) {val req requestChannel.receiveRequest(300)req match {case RequestChannel.ShutdownRequest shutdownComplete.countDown()returncase request: RequestChannel.Request try {request.requestDequeueTimeNanos endTimeapis.handle(request)} catch {case e: FatalExitError shutdownComplete.countDown()Exit.exit(e.statusCode)case e: Throwable error(Exception when handling request, e)} finally {request.releaseBuffer()}case null // continue}}shutdownComplete.countDown()}}
