概述

前面笔者写了一篇文章一文讲清粘包拆包全面的讲解了TCP粘包拆包相关的问题。下面进行一个简单的总结：

TCP粘包拆包产生的原因是，应用层有意义的数据包，传输层的协议并不了解其含义，不会去根据你的业务内容去分包和发送，只会按照自己的协议格式传送数据。

知道问题的本质后，解决问题就简单了。就需要在应用层收到数据后根据标识判断一下，数据是否完整，如果完整了我们再进行数据包解析，最后交给业务代码处理。

解决粘包拆包问题的方法：

（1）消息定长；
（2）增加特殊字符进行分割，比如每条数据末尾都添加一个换行符；
（3）自定义协议，例如 len + data，其中len是代表data的字节长度；

kakfa是如何解决粘包拆包问题的呢

首先看粘包，也就是接收到了多余的数据，该如何拆分数据包，读取到正确完整的数据包？

kafka使用到是上面的第三种解决方法，自定义协议格式。

kafka接收到数据包后，会进行这些操作：

先读取前4字节，转换为一个int，即长度；
根据长度申请内存buffer；
最后读取指定大小的数据到申请好的buffer中；

具体代码实现在：KafkaChannel.read()

public NetworkReceive read() throws IOException {NetworkReceive result = null;if (receive == null) {receive = new NetworkReceive(maxReceiveSize, id);}// 读取数据receive(receive);// 判断是否读取完数据if (receive.complete()) {// 数据读取完后，rewind一下，准备读receive.payload().rewind();result = receive;receive = null;}// return result;
}private long receive(NetworkReceive receive) throws IOException {return receive.readFrom(transportLayer);
}public long readFrom(ScatteringByteChannel channel) throws IOException {return readFromReadableChannel(channel);
}public long readFromReadableChannel(ReadableByteChannel channel) throws IOException {int read = 0;// size就是存放len的缓冲区，大小4个字节，定义：size = ByteBuffer.allocate(4);if (size.hasRemaining()) {int bytesRead = channel.read(size);if (bytesRead < 0)throw new EOFException();read += bytesRead;// 读完了长度if (!size.hasRemaining()) {// rewind一下，准备获取具体的len值size.rewind();// size里的值就是接下来要读取的数据的长度int receiveSize = size.getInt();if (receiveSize < 0)throw new InvalidReceiveException("Invalid receive (size = " + receiveSize + ")");if (maxSize != UNLIMITED && receiveSize > maxSize)throw new InvalidReceiveException("Invalid receive (size = " + receiveSize + " larger than " + maxSize + ")");// 分配读取数据的缓冲区this.buffer = ByteBuffer.allocate(receiveSize);}}if (buffer != null) {int bytesRead = channel.read(buffer);if (bytesRead < 0)throw new EOFException();read += bytesRead;}// 返回读取的总长度return read;
}

接下来，再看看拆包代码。拆包也就是接收到数据不够组成一条完整的数据，该如何等待完整的数据包？

最主要的代码，在上面的receive.complete()方法中的判断逻辑。

public boolean complete() {return !size.hasRemaining() && !buffer.hasRemaining();
}

!size.hasRemaining()：接收到的len数据已经读取完成；
!buffer.hasRemaining()：接收到的data数据已经读取完成；

两个条件同时成立，也就是说既要读取完len，也要读取完data，才算读取了完整的一条数据。

只要一条数据没读完整，那么receive.complete()函数返回值就是false，那么最终返回的结果就是null，等待下一次OP_READ事件的时候再接着上次没读完的数据读取，直到读取一条完整的数据为止。

那么这次读取的数据就会暂存起来，存入stageReceives这个数据结构中等待下一次读取。

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

本文小结

本文详细介绍了kafka是如何解决粘包拆包问题的，关键是要掌握粘包拆包产生的根本原因。

kafka是如何解决粘包拆包的相关推荐

什么是粘包和拆包，Netty如何解决粘包拆包？
Netty粘包拆包 TCP 粘包拆包是指发送方发送的若干包数据到接收方接收时粘成一包或某个数据包被拆开接收. 如下图所示,client 发送了两个数据包 D1 和 D2,但是 server 端可能会收 ...
Netty如何解决粘包拆包？（二）
前言 TCP是个流协议,所谓流,就是没有界限的一串数据.大家可以想想河里的流水,是连成一片的,其间并没有分界线.TCP底层并不了解上层业务数据的具体含义,它会根据TCP缓冲区的实际情况进行包的划分,所 ...
http协议解决粘包拆包半包的编码解码过程、以及netty 使用http协议的原理
本文主要介绍netty对http协议解析原理,着重讲解keep-alive,gzip,truncked等机制,详细描述了netty如何实现对http解析的高性能. 1 http协议 1.1 描述标示 ...
Netty工作笔记0083---通过自定义协议解决粘包拆包问题1
技术交流QQ群[JAVA,C++,Python,.NET,BigData,AI]:170933152 咱们写一个案例,来解决拆包,粘包的问题自己定义一个协议,先去,这里只有协议内容和长度然后用以前 ...
Netty工作笔记0084---通过自定义协议解决粘包拆包问题2
技术交流QQ群[JAVA,C++,Python,.NET,BigData,AI]:170933152 继续写服务器回复给客户端的内容这里写入回复消息代码加入编码器,编码后发给客户端客户端加入解码 ...
《精通并发与Netty》学习笔记（13 - 解决TCP粘包拆包（一）概念及实例演示）
一.粘包/拆包概念 TCP是一个"流"协议,所谓流,就是没有界限的一长串二进制数据.TCP作为传输层协议并不不了解上层业务数据的具体含义,它会根据TCP缓冲区的实际情况进行数据包的 ...
Netty 粘包拆包编码解码序列化介绍
目录: 粘包 & 拆包及解决方案 ByteToMessageDecoder 基于长度编解码器基于分割符的编解码器 google 的 Protobuf 序列化介绍其他的前言 Netty 作 ...
粘包拆包，Netty及远洋通信中的解决方案！超实用
在进行Java NIO学习时,发现,如果客户端连续不断的向服务端发送数据包时,服务端接收的数据会出现两个数据包粘在一起的情况,这就是TCP协议中经常会遇到的粘包以及拆包的问题. 一.什么粘包和拆包? ...
netty的编解码、粘包拆包问题、心跳检测机制原理
文章目录 1. 编码解码器 2. 编解码序列化机制的性能优化 3. Netty粘包拆包 4. Netty心跳检测机制 5. Netty断线自动重连实现 1. 编码解码器当你通过netty发送或者接受 ...

kafka是如何解决粘包拆包的

文章目录

概述

kakfa是如何解决粘包拆包问题的呢

本文小结

kafka是如何解决粘包拆包的相关推荐

最新文章

热门文章