array(2) { ["docs"]=> array(10) { [0]=> array(10) { ["id"]=> string(3) "428" ["text"]=> string(77) "Visual Studio 2017 单独启动MSDN帮助（Microsoft Help Viewer）的方法" ["intro"]=> string(288) "目录 ECharts 异步加载 ECharts 数据可视化在过去几年中取得了巨大进展。开发人员对可视化产品的期望不再是简单的图表创建工具，而是在交互、性能、数据处理等方面有更高的要求。 chart.setOption({ color: [ " ["username"]=> string(8) "DonetRen" ["tagsname"]=> string(55) "Visual Studio 2017|MSDN帮助|C#程序|.NET|Help Viewer" ["tagsid"]=> string(23) "[401,402,403,"300",404]" ["catesname"]=> string(0) "" ["catesid"]=> string(2) "[]" ["createtime"]=> string(10) "1511400964" ["_id"]=> string(3) "428" } [1]=> array(10) { ["id"]=> string(3) "427" ["text"]=> string(42) "npm -v；报错 cannot find module "wrapp"" ["intro"]=> string(288) "目录 ECharts 异步加载 ECharts 数据可视化在过去几年中取得了巨大进展。开发人员对可视化产品的期望不再是简单的图表创建工具，而是在交互、性能、数据处理等方面有更高的要求。 chart.setOption({ color: [ " ["username"]=> string(4) "zzty" ["tagsname"]=> string(50) "node.js|npm|cannot find module "wrapp“|node" ["tagsid"]=> string(19) "[398,"239",399,400]" ["catesname"]=> string(0) "" ["catesid"]=> string(2) "[]" ["createtime"]=> string(10) "1511400760" ["_id"]=> string(3) "427" } [2]=> array(10) { ["id"]=> string(3) "426" ["text"]=> string(54) "说说css中pt、px、em、rem都扮演了什么角色" ["intro"]=> string(288) "目录 ECharts 异步加载 ECharts 数据可视化在过去几年中取得了巨大进展。开发人员对可视化产品的期望不再是简单的图表创建工具，而是在交互、性能、数据处理等方面有更高的要求。 chart.setOption({ color: [ " ["username"]=> string(12) "zhengqiaoyin" ["tagsname"]=> string(0) "" ["tagsid"]=> string(2) "[]" ["catesname"]=> string(0) "" ["catesid"]=> string(2) "[]" ["createtime"]=> string(10) "1511400640" ["_id"]=> string(3) "426" } [3]=> array(10) { ["id"]=> string(3) "425" ["text"]=> string(83) "深入学习JS执行--创建执行上下文（变量对象，作用域链，this）" ["intro"]=> string(288) "目录 ECharts 异步加载 ECharts 数据可视化在过去几年中取得了巨大进展。开发人员对可视化产品的期望不再是简单的图表创建工具，而是在交互、性能、数据处理等方面有更高的要求。 chart.setOption({ color: [ " ["username"]=> string(7) "Ry-yuan" ["tagsname"]=> string(33) "Javascript|Javascript执行过程" ["tagsid"]=> string(13) "["169","191"]" ["catesname"]=> string(0) "" ["catesid"]=> string(2) "[]" ["createtime"]=> string(10) "1511399901" ["_id"]=> string(3) "425" } [4]=> array(10) { ["id"]=> string(3) "424" ["text"]=> string(30) "C# 排序技术研究与对比" ["intro"]=> string(288) "目录 ECharts 异步加载 ECharts 数据可视化在过去几年中取得了巨大进展。开发人员对可视化产品的期望不再是简单的图表创建工具，而是在交互、性能、数据处理等方面有更高的要求。 chart.setOption({ color: [ " ["username"]=> string(9) "vveiliang" ["tagsname"]=> string(0) "" ["tagsid"]=> string(2) "[]" ["catesname"]=> string(8) ".Net Dev" ["catesid"]=> string(5) "[199]" ["createtime"]=> string(10) "1511399150" ["_id"]=> string(3) "424" } [5]=> array(10) { ["id"]=> string(3) "423" ["text"]=> string(72) "【算法】小白的算法笔记：快速排序算法的编码和优化" ["intro"]=> string(288) "目录 ECharts 异步加载 ECharts 数据可视化在过去几年中取得了巨大进展。开发人员对可视化产品的期望不再是简单的图表创建工具，而是在交互、性能、数据处理等方面有更高的要求。 chart.setOption({ color: [ " ["username"]=> string(9) "penghuwan" ["tagsname"]=> string(6) "算法" ["tagsid"]=> string(7) "["344"]" ["catesname"]=> string(0) "" ["catesid"]=> string(2) "[]" ["createtime"]=> string(10) "1511398109" ["_id"]=> string(3) "423" } [6]=> array(10) { ["id"]=> string(3) "422" ["text"]=> string(64) "JavaScript数据可视化编程学习（二）Flotr2，雷达图" ["intro"]=> string(288) "目录 ECharts 异步加载 ECharts 数据可视化在过去几年中取得了巨大进展。开发人员对可视化产品的期望不再是简单的图表创建工具，而是在交互、性能、数据处理等方面有更高的要求。 chart.setOption({ color: [ " ["username"]=> string(7) "chengxs" ["tagsname"]=> string(28) "数据可视化|前端学习" ["tagsid"]=> string(9) "[396,397]" ["catesname"]=> string(18) "前端基本知识" ["catesid"]=> string(5) "[198]" ["createtime"]=> string(10) "1511397800" ["_id"]=> string(3) "422" } [7]=> array(10) { ["id"]=> string(3) "421" ["text"]=> string(36) "C#表达式目录树（Expression）" ["intro"]=> string(288) "目录 ECharts 异步加载 ECharts 数据可视化在过去几年中取得了巨大进展。开发人员对可视化产品的期望不再是简单的图表创建工具，而是在交互、性能、数据处理等方面有更高的要求。 chart.setOption({ color: [ " ["username"]=> string(4) "wwym" ["tagsname"]=> string(0) "" ["tagsid"]=> string(2) "[]" ["catesname"]=> string(4) ".NET" ["catesid"]=> string(7) "["119"]" ["createtime"]=> string(10) "1511397474" ["_id"]=> string(3) "421" } [8]=> array(10) { ["id"]=> string(3) "420" ["text"]=> string(47) "数据结构 队列_队列实例：事件处理" ["intro"]=> string(288) "目录 ECharts 异步加载 ECharts 数据可视化在过去几年中取得了巨大进展。开发人员对可视化产品的期望不再是简单的图表创建工具，而是在交互、性能、数据处理等方面有更高的要求。 chart.setOption({ color: [ " ["username"]=> string(7) "idreamo" ["tagsname"]=> string(40) "C语言|数据结构|队列|事件处理" ["tagsid"]=> string(23) "["246","247","248",395]" ["catesname"]=> string(12) "数据结构" ["catesid"]=> string(7) "["133"]" ["createtime"]=> string(10) "1511397279" ["_id"]=> string(3) "420" } [9]=> array(10) { ["id"]=> string(3) "419" ["text"]=> string(47) "久等了,博客园官方Android客户端发布" ["intro"]=> string(288) "目录 ECharts 异步加载 ECharts 数据可视化在过去几年中取得了巨大进展。开发人员对可视化产品的期望不再是简单的图表创建工具，而是在交互、性能、数据处理等方面有更高的要求。 chart.setOption({ color: [ " ["username"]=> string(3) "cmt" ["tagsname"]=> string(0) "" ["tagsid"]=> string(2) "[]" ["catesname"]=> string(0) "" ["catesid"]=> string(2) "[]" ["createtime"]=> string(10) "1511396549" ["_id"]=> string(3) "419" } } ["count"]=> int(200) } 222 Kafka 源码分析之网络层（二） - 爱码网

Kafka 源码分析之网络层（二）

Kafka 源码分析之网络层（二）


上一篇介绍了概述和网络层模型实现《Kafka 源码分析之网络层（一）》，本编主要介绍在Processor中使用的nio selector的又一封装，负责具体数据的接收和发送。

PS：丰富的一线技术、多元化的表现形式，尽在“HULK一线技术杂谈”，点关注哦！


对nio的封装:Selector类

  • 所在文件: clients/src/main/java/org/apache/kafka/commmon/network/Selector.java

  • 源码中的注释:

A nioSelector interface for doing non-blocking multi-connection network I/O. This class works with NetworkSend} and NetworkReceive to transmit size-delimited network requests and responses.

重要函数解析:

(1) register(String id, SocketChannel socketChannel): 注册这个socketChannel到一个nio selector, 将其读事件添加到selector的监听队列; 这个socketChannel通常是服务器接收到的客户端的连接:



  1. SelectionKey key = socketChannel.register(nioSelector, SelectionKey.OP_READ);



同时创建KafkaChannel, 负责实际的数据接收和发送:




    

  1. KafkaChannel channel = channelBuilder.buildChannel(id, key, maxReceiveSize);
    2. 

  2. key.attach(channel);
    3. 

  3. this.channels.put(id, channel);
    4. 




上面的id即为我们在上篇介绍的非常重要的ConnectionId。



(2) connect: 使用nio的SocketChannel连接到给定的地址,并且注册到nio selector,同时也创建了KafkaChannel,负责实际的数据接收和发送;




    

  1. SocketChannel socketChannel = SocketChannel.open();
    2. 

  2. socketChannel.configureBlocking(false);
    3. 

  3. Socket socket = socketChannel.socket();
    4. 

  4. socket.setKeepAlive(true);
    5. 

  5. socketChannel.connect(address);
    6. 

  6. SelectionKey key = socketChannel.register(nioSelector, SelectionKey.OP_CONNECT);
    7. 

  7. KafkaChannel channel = channelBuilder.buildChannel(id, key, maxReceiveSize);
    8. 

  8. key.attach(channel);
    9. 

  9. this.channels.put(id, channel);
    10. 




(3) poll: 核心函数:




Do whatever I/O can be done on each connection without blocking. This includes completing connections, completing disconnections, initiating new sends, or making progress on in-progress
sends or receives.





处理作为客户端的主动连接事件:




    

  1. if (key.isConnectable()) {
    2. 

  2.         channel.finishConnect();
    3. 

  3.          this.connected.add(channel.id());
    4. 

  4.          this.sensors.connectionCreated.record();
    5. 

  5. }
    6. 




处理连接建立或接收后的ssl握手或sasl签权操作:




    

  1. if (channel.isConnected() && !channel.ready())
    2. 

  2.       channel.prepare();
    3. 




处理触发的读事件:




    

  1. if (channel.ready() && key.isReadable() && !hasStagedReceive(channel)) {
    2. 

  2.           NetworkReceive networkReceive;
    3. 

  3.           while ((networkReceive = channel.read()) != null)
    4. 

  4.                      addToStagedReceives(channel, networkReceive);
    5. 

  5. }
    6. 




使用一个while循环力求每次读事件触发时都读尽可能多的数据; channel.read()里会作拆包处理(后面会讲到),返回非null表示当前返回的NetworkReceive里包含了完整的应用层协议数据;



处理触发的写事件:




    

  1. if (channel.ready() && key.isWritable()) {
    2. 

  2.          Send send = channel.write();
    3. 

  3.           if (send != null) {
    4. 

  4.                  this.completedSends.add(send);
    5. 

  5.                   this.sensors.recordBytesSent(channel.id(), send.size());
    6. 

  6.           }
    7. 

  7. }
    8. 




需要发送数据通过调用Selector::send方法,设置封装了写数据的NetworkSend,再将这个NetworkSend通过KafkaChannel::setSend接口设置到KafkaChannel,同时将写事件添加到selector的监听队列中,等待写事件被触发后,通过KafkaChannel::write将数据发送出去;



addToCompletedReceives()



将当前接收到的完整的的request到添加到completedReceives中,上一篇中介绍的SocketServer会作completedReceives中取出这些request作处理;



封装对单个连接的读写操作:KafkaChannel类


    

  • 

    所在文件: clients/src/main/java/org/apache/kafka/common/network/KafkaChannel.java
    

    • 

  • 

    包括transportLayer和authenticator, 完成ssh握手,sasl签权,数据的接收和发送;
    

    • 




传输层:TransportLayer类


    

  • 

    所在文件 clients/src/main/java/org/apache/kafka/common/network/TransportLayer.java
    

    • 

  • 

    两个子类: PlanintextTransportLayer和SslTransportLayer
    

    • 

  • 

    PlanintextTransportLayer的实现主要是通过NetworkReceive和NetworkSend;
    

    • 

  • 

    SslTransportLayer的实现主要是通过SocketChannel,ByteBuffers和SSLEngine实际了加密数据的接收和发送(看到ssl就头大啊,这部分先忽略~~~);
    

    • 




Kafka协议的包结构:


    

  • 

    前4个字节固定, 值是后面的实际数据的长度;
    

    • 

  • 

    NetworkReceive: 接收时先接收4个字节, 获取到长度,然后再接收实际的数据;
    

    • 

  • 

    NetworkSend: 发送时实际数据前先加上4个字节的数据长度再发送;
    

    • 




上图:



Kafka 源码分析之网络层（二）







希望这两篇网络层相关的源码分析文章能够帮助大家更好的理解和使用kafka，后续我们会带来更多的源码分析文章，敬请期待~，大家有想了解的技术，也可以留言给我们哦。

扫描下方二维码了解更多内容


Kafka 源码分析之网络层（二）





 
                    
            
                

                  
 
        

	 
        

            

                                            

        

        
相关文章：

        

      
            
 
                    
    
                                            

               
   
         

    


  






 
          

    
猜你喜欢 

        
    
                        

   
 



    
相关资源 

        
    
              
                            

   

   
    

     
相似解决方案 

         
    
                          

    

  
   

    

 
   

    
热门标签  

    

                    Java
                        Python
                        linux
                        javascript
                        Mysql
                        C#
                        Docker
                        算法
                        前端
                        SpringBoot
                        Redis
                        Vue
                        spring
                        设计模式
                        .net core
                        .net
                        kubernetes
                        c++
                        数据库
                        数据结构
                        大数据
                        js
                        机器学习
                        微服务
                        Android
                        Go
                        程序员
                        面试
                        JVM
                        ASP.net core
                        云原生
                        人工智能
                        后端
                        PHP
                        git
                        CSS
                        golang
                        k8s
                        Nginx
                        Django
                        mybatis
                        深度学习
                        多线程
                        React
                        架构
                        devops
                        爬虫
                        云计算
                        Spring Boot
                        LeetCode