kafka 的network
2016-07-05 11:22
357 查看
好了 , 接着上一章 , 我们回到kafka的 kafkaserver 这个重量级的类。
在初始化zk连接, 加载topic信息之后。kafka开始跟做一些io的东西。个人对这部分还是很感兴趣的。让我们点进去看一看。
注释写的很精彩啊:
他都已经说了,这是 NIO 线程模型是 单线程负责处理所以的连接。n个线程异步处理这些连接。
从这个注释入手,我们看一看 Acceptor 和 Processor 是如何实现的。
如果明白 java NIO 的相关部分,就会比较容易看懂这部分,忘了的上网搜搜。 一段标准的java server端的NIO 的操作, 绑定端口,注册事件,轮询 selector。如果有连接事件,就交给 processor 来处理。 简单而强有力的做法。下面看看Processor 是咋实现的。
好 , 咱们看看首先他有一个newConnections队列 用来存储SocketChannel 对象,实际上可以看成缓存请求的消息队列。
在 run方法中, 先清空了这个队列,同时在selector 中注册这些事件。然后又是nio的一段标准的程序。看看read 方法中都干了什么:
关键是匹配事件之后,他们都干了什么。由于对scala 语法不是那么纯熟,不知道咱们的就调用到了
也就是说 在消费消息的时候是把 内容封装到 MessageSetSend 中作为参数返回给客户端。
val handlers = new KafkaRequestHandlers(logManager) socketServer = new SocketServer(config.port, config.numThreads, config.monitoringPeriodSecs, handlers.handlerFor, config.socketSendBuffer, config.socketReceiveBuffer, config.maxSocketRequestSize)
在初始化zk连接, 加载topic信息之后。kafka开始跟做一些io的东西。个人对这部分还是很感兴趣的。让我们点进去看一看。
注释写的很精彩啊:
/** * An NIO socket server. The thread model is * 1 Acceptor thread that handles new connections * N Processor threads that each have their own selectors and handle all requests from their connections synchronously */
他都已经说了,这是 NIO 线程模型是 单线程负责处理所以的连接。n个线程异步处理这些连接。
从这个注释入手,我们看一看 Acceptor 和 Processor 是如何实现的。
/** * Thread that accepts and configures new connections. There is only need for one of these */ private[kafka] class Acceptor(val port: Int, private val processors: Array[Processor], val sendBufferSize: Int, val receiveBufferSize: Int) extends AbstractServerThread { /** * Accept loop that checks for new connection attempts */ def run() { val serverChannel = ServerSocketChannel.open() serverChannel.configureBlocking(false) serverChannel.socket.bind(new InetSocketAddress(port)) serverChannel.register(selector, SelectionKey.OP_ACCEPT); logger.info("Awaiting connections on port " + port) startupComplete() var currentProcessor = 0 while(isRunning) { val ready = selector.select(500) if(ready > 0) { val keys = selector.selectedKeys() val iter = keys.iterator() while(iter.hasNext && isRunning) { var key: SelectionKey = null try { 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 => logger.error("Error in acceptor", e) } } } } logger.debug("Closing server socket and selector.") Utils.swallow(logger.error, serverChannel.close()) Utils.swallow(logger.error, selector.close()) shutdownComplete() }
如果明白 java NIO 的相关部分,就会比较容易看懂这部分,忘了的上网搜搜。 一段标准的java server端的NIO 的操作, 绑定端口,注册事件,轮询 selector。如果有连接事件,就交给 processor 来处理。 简单而强有力的做法。下面看看Processor 是咋实现的。
private val newConnections = new ConcurrentLinkedQueue[SocketChannel](); private val requestLogger = Logger.getLogger("kafka.request.logger") override def run() { startupComplete() while(isRunning) { // setup any new connections that have been queued up configureNewConnections() val ready = selector.select(500) if(ready > 0) { val keys = selector.selectedKeys() val iter = keys.iterator() while(iter.hasNext && isRunning) { var key: SelectionKey = null try { key = iter.next iter.remove() if(key.isReadable) read(key) else if(key.isWritable) write(key) else if(!key.isValid) close(key) else throw new IllegalStateException("Unrecognized key state for processor thread.") } catch { case e: EOFException => { logger.info("Closing socket connection to %s.".format(channelFor(key).socket.getInetAddress)) close(key) } case e: InvalidRequestException => { logger.info("Closing socket connection to %s due to invalid request: %s".format(channelFor(key).socket.getInetAddress, e.getMessage)) close(key) } case e: Throwable => { logger.error("Closing socket for " + channelFor(key).socket.getInetAddress + " because of error", e) close(key) } } } } } logger.debug("Closing selector.") Utils.swallow(logger.info, selector.close()) shutdownComplete() }
好 , 咱们看看首先他有一个newConnections队列 用来存储SocketChannel 对象,实际上可以看成缓存请求的消息队列。
在 run方法中, 先清空了这个队列,同时在selector 中注册这些事件。然后又是nio的一段标准的程序。看看read 方法中都干了什么:
/** * Handle a completed request producing an optional response */ private def handle(key: SelectionKey, request: Receive): Option[Send] = { val requestTypeId = request.buffer.getShort() if(requestLogger.isTraceEnabled) { requestTypeId match { case RequestKeys.Produce => requestLogger.trace("Handling produce request from " + channelFor(key).socket.getRemoteSocketAddress()) case RequestKeys.Fetch => requestLogger.trace("Handling fetch request from " + channelFor(key).socket.getRemoteSocketAddress()) case RequestKeys.MultiFetch => requestLogger.trace("Handling multi-fetch request from " + channelFor(key).socket.getRemoteSocketAddress()) case RequestKeys.MultiProduce => requestLogger.trace("Handling multi-produce request from " + channelFor(key).socket.getRemoteSocketAddress()) case RequestKeys.Offsets => requestLogger.trace("Handling offset request from " + channelFor(key).socket.getRemoteSocketAddress()) case _ => throw new InvalidRequestException("No mapping found for handler id " + requestTypeId) } } val handler = handlerMapping(requestTypeId, request) if(handler == null) throw new InvalidRequestException("No handler found for request") val start = time.nanoseconds val maybeSend = handler(request) stats.recordRequest(requestTypeId, time.nanoseconds - start) maybeSend }
关键是匹配事件之后,他们都干了什么。由于对scala 语法不是那么纯熟,不知道咱们的就调用到了
var response: MessageSetSend = null try { trace("Fetching log segment for topic, partition, offset, maxSize = " + fetchRequest) val log = logManager.getLog(fetchRequest.topic, fetchRequest.partition) if (log != null) { response = new MessageSetSend(log.read(fetchRequest.offset, fetchRequest.maxSize)) BrokerTopicStat.getBrokerTopicStat(fetchRequest.topic).recordBytesOut(response.messages.sizeInBytes) BrokerTopicStat.getBrokerAllTopicStat.recordBytesOut(response.messages.sizeInBytes) } else response = new MessageSetSend() }
也就是说 在消费消息的时候是把 内容封装到 MessageSetSend 中作为参数返回给客户端。
相关文章推荐
- 拨开kafka 的羊毛衫
- ProgressDialog使用
- shell 判断 404
- AES对称加密
- 101. Symmetric Tree
- Route Filters
- Routes
- Android中的布局和控件的隐藏和触发显示
- java 内存模型与线程 正文
- AndroidStudio如何打包生成realease版本的arr包,并上传到Nexus搭建的maven仓库,供项目远程依赖(二)
- [转载]DynaActionForm (动态ActionForm)
- Utuntu安装Bugzilla
- 主键约束,唯一约束与默认约束
- JAVA学习笔记two:关键字
- java 内存模型与线程 前传
- 大于、等于、小于等符号转换
- 使用tmodjs预编译模板实例
- C语言的常量
- Redis 3.0配置
- awk 中简单的去重方法