即使关闭nagle算法,也不能解决粘包问题
https://waylau.com/netty-4-user-guide/Getting%20Started/Dealing%20with%20a%20Stream%20based%20Transport.html
Dealing with a Stream-based Transport 处理一个基于流的传输
One Small Caveat of Socket Buffer 关于 Socket Buffer的一个小警告
基于流的传输比如 TCP/IP, 接收到数据是存在 socket 接收的 buffer 中。不幸的是,基于流的传输并不是一个数据包队列,而是一个字节队列。意味着,即使你发送了2个独立的数据包,操作系统也不会作为2个消息处理而仅仅是作为一连串的字节而言。因此这是不能保证你远程写入的数据就会准确地读取。举个例子,让我们假设操作系统的 TCP/TP 协议栈已经接收了3个数据包:
由于基于流传输的协议的这种普通的性质,在你的应用程序里读取数据的时候会有很高的可能性被分成下面的片段
因此,一个接收方不管他是客户端还是服务端,都应该把接收到的数据整理成一个或者多个更有意思并且能够让程序的业务逻辑更好理解的数据。在上面的例子中,接收到的数据应该被构造成下面的格式:
https://netty.io/wiki/user-guide-for-4.x.html
Dealing with a Stream-based Transport
One Small Caveat of Socket Buffer
In a stream-based transport such as TCP/IP, received data is stored into a socket receive buffer. Unfortunately, the buffer of a stream-based transport is not a queue of packets but a queue of bytes. It means, even if you sent two messages as two independent packets, an operating system will not treat them as two messages but as just a bunch of bytes. Therefore, there is no guarantee that what you read is exactly what your remote peer wrote. For example, let us assume that the TCP/IP stack of an operating system has received three packets:
Because of this general property of a stream-based protocol, there's high chance of reading them in the following fragmented form in your application:
Therefore, a receiving part, regardless it is server-side or client-side, should defrag the received data into one or more meaningful frames that could be easily understood by the application logic. In case of the example above, the received data should be framed like the following:
The First Solution
Now let us get back to the TIME
client example. We have the same problem here. A 32-bit integer is a very small amount of data, and it is not likely to be fragmented often. However, the problem is that it can be fragmented, and the possibility of fragmentation will increase as the traffic increases.
The simplistic solution is to create an internal cumulative buffer and wait until all 4 bytes are received into the internal buffer. The following is the modified TimeClientHandler
implementation that fixes the problem:
package io.netty.example.time;
import java.util.Date;
public class TimeClientHandler extends ChannelInboundHandlerAdapter {
private ByteBuf buf;
@Override
public void handlerAdded(ChannelHandlerContext ctx) {
buf = ctx.alloc().buffer(4); // (1)
}
@Override
public void handlerRemoved(ChannelHandlerContext ctx) {
buf.release(); // (1)
buf = null;
}
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) {
ByteBuf m = (ByteBuf) msg;
buf.writeBytes(m); // (2)
m.release();
if (buf.readableBytes() >= 4) { // (3)
long currentTimeMillis = (buf.readUnsignedInt() - 2208988800L) * 1000L;
System.out.println(new Date(currentTimeMillis));
ctx.close();
}
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
cause.printStackTrace();
ctx.close();
}
}
- A
ChannelHandler
has two life cycle listener methods:handlerAdded()
andhandlerRemoved()
. You can perform an arbitrary (de)initialization task as long as it does not block for a long time. - First, all received data should be cumulated into
buf
. - And then, the handler must check if
buf
has enough data, 4 bytes in this example, and proceed to the actual business logic. Otherwise, Netty will call thechannelRead()
method again when more data arrives, and eventually all 4 bytes will be cumulated.
The Second Solution
Although the first solution has resolved the problem with the TIME
client, the modified handler does not look that clean. Imagine a more complicated protocol which is composed of multiple fields such as a variable length field. Your ChannelInboundHandler
implementation will become unmaintainable very quickly.
As you may have noticed, you can add more than one ChannelHandler
to a ChannelPipeline
, and therefore, you can split one monolithic ChannelHandler
into multiple modular ones to reduce the complexity of your application. For example, you could split TimeClientHandler
into two handlers:
TimeDecoder
which deals with the fragmentation issue, and- the initial simple version of
TimeClientHandler
.
Fortunately, Netty provides an extensible class which helps you write the first one out of the box:
package io.netty.example.time;
public class TimeDecoder extends ByteToMessageDecoder { // (1)
@Override
protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) { // (2)
if (in.readableBytes() < 4) {
return; // (3)
}
out.add(in.readBytes(4)); // (4)
}
}
ByteToMessageDecoder
is an implementation ofChannelInboundHandler
which makes it easy to deal with the fragmentation issue.ByteToMessageDecoder
calls thedecode()
method with an internally maintained cumulative buffer whenever new data is received.decode()
can decide to add nothing toout
where there is not enough data in the cumulative buffer.ByteToMessageDecoder
will calldecode()
again when there is more data received.- If
decode()
adds an object toout
, it means the decoder decoded a message successfully.ByteToMessageDecoder
will discard the read part of the cumulative buffer. Please remember that you don't need to decode multiple messages.ByteToMessageDecoder
will keep calling thedecode()
method until it adds nothing toout
.
Now that we have another handler to insert into the ChannelPipeline
, we should modify the ChannelInitializer
implementation in the TimeClient
:
b.handler(new ChannelInitializer<SocketChannel>() {
@Override
public void initChannel(SocketChannel ch) throws Exception {
ch.pipeline().addLast(new TimeDecoder(), new TimeClientHandler());
}
});
If you are an adventurous person, you might want to try the ReplayingDecoder
which simplifies the decoder even more. You will need to consult the API reference for more information though.
public class TimeDecoder extends ReplayingDecoder<Void> {
@Override
protected void decode(
ChannelHandlerContext ctx, ByteBuf in, List<Object> out) {
out.add(in.readBytes(4));
}
}
Additionally, Netty provides out-of-the-box decoders which enables you to implement most protocols very easily and helps you avoid from ending up with a monolithic unmaintainable handler implementation. Please refer to the following packages for more detailed examples:
io.netty.example.factorial
for a binary protocol, andio.netty.example.telnet
for a text line-based protocol.