Netty 中 TCP 消息边界问题及按行分割消息

Posted 2021-08-02 马士兵

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了Netty 中 TCP 消息边界问题及按行分割消息相关的知识，希望对你有一定的参考价值。

在TCP连接开始到结束连接，之间可能会多次传输数据，也就是服务器和客户端之间可能会在连接过程中互相传输多条消息。理想状况是一方每发送一条消息，另一方就立即接收到一条，也就是一次write对应一次read。但是，现实不总是按照剧本来走。

Netty官方文档节选：

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.

TCP是基于字节流的协议，它只能保证一方发送和另一方接收到的数据的字节顺序一致，但是，并不能保证一方每发送一条消息，另一方就能完整的接收到一条信息。有可能发送了两条对方将其合并成一条，也有可能发送了一条对方将其拆分成两条。所以在上一篇文章中的Demo，可以说是一个错误的示范。不过服务器和客户端在同一台机器上或者在局域网等网速很好的情况下，这种问题还是很难测试出来。

举个简单了例子（这个例子来源于Netty官方文档）：消息发送方发送了三个字符串：但是接收方收到的可能是这样的： Netty 中 TCP 消息边界问题及按行分割消息那么问题就很严重了，接收方没法分开这三条信息了，也就没法解析了。

怎么办呢？我给大家写一段代码：

public class TcpServer {

 public static void main(String[] args) throws InterruptedException {
   EventLoopGroup bossGroup = new NioEventLoopGroup();
   EventLoopGroup workerGroup = new NioEventLoopGroup();
   try {
     ServerBootstrap b = new ServerBootstrap();
     b.group(bossGroup, workerGroup)
         .channel(NioserverSocketChannel.class)
         .childHandler(new ChannelInitializer<SocketChannel>() {
           @Override
           public void initChannel(SocketChannel ch)
               throws Exception {
             ChannelPipeline pipeline = ch.pipeline();

             // LineBasedFrameDecoder按行分割消息
             pipeline.addLast(new LineBasedFrameDecoder(80));
             // 再按UTF-8编码转成字符串
             pipeline.addLast(new StringDecoder(CharsetUtil.UTF_8));

             pipeline.addLast(new TcpServerHandler());
           }
         });
     ChannelFuture f = b.bind(8080).sync();
     f.channel().closeFuture().sync();
   } finally {
     workerGroup.shutdownGracefully();
     bossGroup.shutdownGracefully();
   }
 }

}

class TcpServerHandler extends ChannelInboundHandlerAdapter {

 // 接收到新的数据
 @Override
 public void channelRead(ChannelHandlerContext ctx, Object msg) {

   // msg经过StringDecoder后类型不再是ByteBuf而是String
   String line = (String) msg;
   System.out.println("channelRead:" + line);
 }

 @Override
 public void channelActive(ChannelHandlerContext ctx) {
   System.out.println("channelActive");
 }

 @Override
 public void channelInactive(ChannelHandlerContext ctx) {
   System.out.println("channelInactive");
 }

 @Override
 public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
   cause.printStackTrace();
   ctx.close();
 }
}

需要在ChannelPipeline加上一些ChannelHandler用来对原始数据进行处理。这里用LineBasedFrameDecoder将接收到的数据按行分割，StringDecoder再将数据由字节码转成字符串。同样，接收到的数据进过加工后，在channelRead事件函数中，msg参数不再是ByteBuf而是String。

下面我们写一个测试一下：

public class TcpClient {

 public static void main(String[] args) throws IOException {

   Socket socket = null;
   OutputStream out = null;

   try {

     socket = new Socket("localhost", 8080);
     out = socket.getOutputStream();

     // 请求服务器
     String lines = "床前明月光\r\n疑是地上霜\r\n举头望明月\r\n低头思故乡\r\n";
     byte[] outputBytes = lines.getBytes("UTF-8");
     out.write(outputBytes);
     out.flush();

   } finally {
     // 关闭连接
     out.close();
     socket.close();
   }

 }

}