kafka之网络模型总结

2019-06-28  本文已影响0人  外星人rsz

弄清楚kafka的网络模型原理，能很好的帮助理解和优化kafka服务。kafka底层的网络通信，没有使用第三方rpc实现，如netty等，而是使用了java的NIO实现的一套自己的通信框架协议。本文主要描述kafka基础网络通信的相关实现原理，版本为1.1.0。

java NIO具体细节不再描述，主要包含3个部分：

kafka实现网络模型主要涉及到3个类：

通信概况

kafka网络模型_精简

kafka接收消息、处理并返回，主要有以下步骤：

  1. Acceptor：监听OP_ACCEPT，将连接的channel以round-robin的方式选择processor进行处理

  2. Processor：监听连接的OP_READ和OP_WRITE，主要负责读写和解析数据

    • 读取客户端消息解析后放入requestQueue，解析头部信息
    • 将responseQueue中的消息发送回客户端

  3. KafkaRequestHandler：从requestQueue中获取连接，根据头部信息获取对应的KafkaApi协议进行相关处理，并通过回调，将处理后的结果通过RequestChannel写到对应Processor的responseQueue中，等待Processor线程处理

通信处理详细步骤

kafka网络模型

如图所示，其中，蓝色线表示请求处理及流向，绿色组件表示消息缓存。

  1. SocketServer中，主要功能有：

    • 通过配置的listeners可以监听多个interface/port，构建对应的Acceptor
    • 每个Acceptor构建num.network.threads个Processor，用于处理连接请求
    • 构建RequestChannel，保存解析好的连接请求
    # 主要逻辑
private def addProcessors(acceptor: Acceptor, endpoint: EndPoint, newProcessorsPerListener: Int): Unit = synchronized {
    ...
    for (_ <- 0 until newProcessorsPerListener) {
      val processor = newProcessor(nextProcessorId, connectionQuotas, listenerName, securityProtocol, memoryPool)
      listenerProcessors += processor
      requestChannel.addProcessor(processor)
      nextProcessorId += 1
    }
    listenerProcessors.foreach(p => processors.put(p.id, p))
    acceptor.addProcessors(listenerProcessors)
  }

  2. Acceptor为线程，主要功能有：

    • 监听OP_ACCEPT事件，不断循环获取已经accept的连接
    • 判断每个ip的连接数是否超过quota限制
    • 通过round-robin的方式，选择Processor，放入Processor对应的newConnections缓存中
    • 代码逻辑主要看run()方法

  3. Processor为线程，主要功能为缓存新建连接、接收并缓存数据、缓存返回信息、处理断开连接等

    • configureNewConnections(): 处理新建的连接，监听OP_READ事件，等待读取数据
    • poll()
      • 真正读取数据，并放入接收缓存队列stagedReceives，缓存所有channel的请求
      • 拿出每个channel的第一个请求，解析协议头部，放入completedReceives缓存中
      • 如果channel写出ready，则进行write，将response返回给客户端
    • processCompletedReceives()：将请求解析为Request并放入requestQueue缓存
    # 主要代码逻辑
override def run() {
    startupComplete()
    try {
      while (isRunning) {
        try {
          // setup any new connections that have been queued up
          configureNewConnections()
          // register any new responses for writing
          processNewResponses()
          poll()
          processCompletedReceives()
          processCompletedSends()
          processDisconnected()
          ...

  4. KafkaRequestHandlerPool初始化num.io.threads个KafkaRequestHandler

    • 从requestQueue中拿去request，并根据协议头，选择对应的KafkaApi进行处理
    • 使用回调，将处理完成的response通过KafkaChannel放入当前channel对应的Processor的responseQueue中

由此可见，kafka内部通信使用了NIO+缓存+异步，从而极大提升了kafka的单机并发能力。

至此，整个网络处理大概步骤完成，其中很多细节内容，这里不再赘述，感兴趣的同学可以看源码或一起交流。

通信协议

kafka网络模型_协议规范

kafka接收到的数据内容格式如上图所示：

KafkaApi支持各种协议，具体可参见源码，这里只大概描述Produce协议的主要内容。基于此，kafka各版本之间可以保证很好的协议兼容性。

优化

基于以上分析，我们通过分析相关监控指标，进行了相关调优工作，主要调优内容为(仅供参考)：

zookeeper.session.timeout.ms=30000
num.network.threads=12
num.io.threads=16
num.replica.fetchers=3
replica.fetch.max.bytes=2097152
replica.fetch.response.max.bytes=20971520

调整后，cpu的负载明显提升，更加有效的利用机器资源，processor和handler的idle比例明显提高，isr expand/shrink出现频次和梳理也明显降低，大大提升的集群的稳定性。

结尾

本文主要介绍kafka内部的通信模型，kafka组件还有很多模块，后续会不断深入学习和理解，欢迎大家一起交流和学习。最近Pulsar也有兴起之势，后续也需要学习，真是学无止境啊~~

参考

matt博客

上一篇下一篇

猜你喜欢

热点阅读