Kafka源码分析-网络层-3

  • 这篇将是网络层源码分析的最后一篇
  • 北京的天,无力吐槽啊~
  • 快年底了, 每个团队都在旁边录制新年寄语,各种口号~

对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通常是服务器接收到的客户端的连接:
SelectionKey key = socketChannel.register(nioSelector, SelectionKey.OP_READ);

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

KafkaChannel channel = channelBuilder.buildChannel(id, key, maxReceiveSize);
key.attach(channel);
this.channels.put(id, channel);

上面的id即为我们在上篇介绍的非常重要的ConnectionId; (2) connect: 使用nio的SocketChannel连接到给定的地址,并且注册到nio selector,同时也创建了KafkaChannel,负责实际的数据接收和发送;

SocketChannel socketChannel = SocketChannel.open();
socketChannel.configureBlocking(false);
Socket socket = socketChannel.socket();
socket.setKeepAlive(true);
socketChannel.connect(address);
SelectionKey key = socketChannel.register(nioSelector, SelectionKey.OP_CONNECT);
KafkaChannel channel = channelBuilder.buildChannel(id, key, maxReceiveSize);
key.attach(channel);
this.channels.put(id, channel);

(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.

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

if (key.isConnectable()) {
        channel.finishConnect();
         this.connected.add(channel.id());
         this.sensors.connectionCreated.record();
}

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

if (channel.isConnected() && !channel.ready())
      channel.prepare();

处理触发的读事件:

if (channel.ready() && key.isReadable() && !hasStagedReceive(channel)) {
          NetworkReceive networkReceive;
          while ((networkReceive = channel.read()) != null)
                     addToStagedReceives(channel, networkReceive);
}

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

if (channel.ready() && key.isWritable()) {
         Send send = channel.write();
          if (send != null) {
                 this.completedSends.add(send);
                  this.sensors.recordBytesSent(channel.id(), send.size());
          }
}

需要发送数据通过调用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个字节的数据长度再发送;

上图:

selector.png

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏信安之路

Red Team 工具集之攻击武器库

上图是一个 Red Team 攻击的生命周期,整个生命周期包括:信息收集、攻击尝试获得权限、持久性控制、权限提升、网络信息收集、横向移动、数据分析(在这个基础上...

23400
来自专栏较真的前端

同样是客户端会话级存储,sessionStorage和session cookie有什么?

35840
来自专栏Hadoop实操

CDH集群安装YARN无法正常启动及解决办法

在安装5.11.2版本的CDH集群时,或者为5.11.2版本的CDH集群扩容时,可能会遇到YARN的NodeManager实例无法启动的问题。本文主要讲述Nod...

90240
来自专栏计算机编程

SNS项目笔记<四>--RXjs简要用法

在命令行输入ionic g provider youProviderName 在创建好后,系统会自动导入从@angular/http里导入Http这个类,方便后...

15540
来自专栏挖掘大数据

ambari安装指南

(一)准备工作 1、准备几台机器 10.1.51.100 ambariserver 本地mirrorserver及ambari server都在这一台机器 10...

31180
来自专栏技术碎碎念

Tomcat NIO

说起Tomcat的NIO,不得不提的就是Connector这个Tomcat组件。Connector是Tomcat的连接器,其主要任务是负责处理收到的请求,并创建...

68360
来自专栏Golang语言社区

gRPC服务发现&amp;负载均衡

构建高可用、高性能的通信服务,通常采用服务注册与发现、负载均衡和容错处理等机制实现。根据负载均衡实现所在的位置不同,通常可分为以下三种解决方案:

87820
来自专栏大魏分享(微信公众号:david-share)

用Ansible自动供应vmware虚拟机--构建数据中心一体化运维平台第二篇

1.1 简述 一直以来,打开邮箱被ticket糊一脸的事情时有发生。我一直在想,能不能以一种简单的方案(不花老板的钱)来供应(provisioning)虚拟机呢...

74020
来自专栏智能大石头

NewLife.Net——开始网络编程

14330
来自专栏我和未来有约会

Silverlight 3.0 中的 Local Connection

现在很多的需求中需要一个插件实例和另一个实例进行通讯。在同一个页面中调用Html、js等来通讯,而这个往往有一些限制,需要专门的去设置一些权限。在Silverl...

22170

扫码关注云+社区

领取腾讯云代金券