TCP粘包/拆包--利用LineBasedFrameDecoder解决TCP粘包问题

时间 2020-02-09 标签 tcp 粘包拆包利用 linebasedframedecoder 解决 tcp 粘包问题

节选自 Netty权威指南第二版java

TCP是个“流”协议，所谓流，就是没有界面的一串数据。你们能够想象河里的流水，bootstrap

它们是连成一片的，其间并无分界线。TCP底层并不了解上层业务数据的具体含义，它安全

会根据TCP缓冲区的实际状况进行包的划分，因此在业务上认为，一个完整的包可能会服务器

被TCP拆分红多个包进行发送，也有可能把多个小小的包封装成一个大的数据包发送，这框架

就是所谓的TCP粘包和拆包问题。socket

TCP粘包/拆包问题说明

咱们能够经过图解对TCP粘包和拆包问题进行说明，以下图ide

假设客户端分别发送了两个数据包D1和D2给服务端，因为服务端一次读取到的字节数是不肯定的，故可能存在如下4种状况。oop

1）服务端分两次取到了两个独立的数据包，分别是D1和D2，没有粘包和拆包。编码

2）服务端一次接收到了两个数据包，D1和D2粘合在一块儿，被称为TCP粘包。spa

3）服务端分两次读取到了两个数据包，第一次读取到了完整的D1包和D2包的部份内容，第二次取到了D2包的剩余内容，这被称为TCP粘包。

4）服务端分两次读取到了两个数据包，第一次读取到D1包的部份内容D1_1，第二次读取到了D1包的剩余内容和D2包的整包。

若是此时服务端TCP接受滑窗很是小，而数据包D1和D2比较大，颇有可能会发生第5种可能，即服务端分屡次才能将D1和D2包接收彻底，期间发生屡次拆包。

TCP粘包/拆包发生的缘由

问题产生的缘由有三个，分别以下。

1. 应用程序write写入的字节大小大于套接口发送缓冲区大小；

2. 进行MSS大小的TCP分段；

3. 以太网帧的payload大于MTU进行IP分片。

粘包问题的解决策略

因为底层的TCP没法理解上层的业务数据，因此在底层是没法保证数据包不被拆分和重组的，这个问题只能经过上层的应用协议栈设计来解决，根据业界的主流协议的方案，能够概括以下。

1. 消息定长，例如每一个报文的大小为固定长度200字节，若是不够，空位补空格；

2. 在包尾增长回车换行符进行分割，例如FTP协议；

3. 将消息分为消息头和消息体，消息头中包含表示消息总长度（或者消息体长度）的字段，一般涉及思路为消息头的第一个字段使用int32来表示消息的总长度；

4. 更复杂的应用层协议。

未考虑TCP粘包功能致使异常案例

1. TimeServer.java

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;

import java.net.InetSocketAddress;

public class TimeServer {
    private final static int PORT = 8080;

    public static void main(String[] args) {
        start();
    }

    private static void start() {
        final TimeServerHandler serverHandler = new TimeServerHandler();
        // 建立EventLoopGroup
        EventLoopGroup bossGroup = new NioEventLoopGroup();
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        // 建立EventLoopGroup
        ServerBootstrap b = new ServerBootstrap();
        b.group(bossGroup, workerGroup)
                //指定所使用的NIO传输Channel
                .channel(NioServerSocketChannel.class)
                //使用指定的端口设置套接字地址
                .localAddress(new InetSocketAddress(PORT))
                // 添加一个EchoServerHandler到Channle的ChannelPipeline
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel socketChannel) throws Exception {
                        //EchoServerHandler被标注为@shareable,因此咱们能够老是使用一样的案例
                        socketChannel.pipeline().addLast(serverHandler);
                    }
                });

        try {
            ChannelFuture f = b.bind().sync();
            f.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}

2. TimeServerHandler.java

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandler.Sharable;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;

import java.util.Date;

@Sharable
public class TimeServerHandler extends ChannelInboundHandlerAdapter {
    private int counter;
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        ByteBuf buf = (ByteBuf) msg;
        byte[] req = new byte[buf.readableBytes()];
        buf.readBytes(req);
        String body = new String(req, "UTF-8").substring(0, req.length - System.getProperty("line.separator").length());
        System.out.println(
                "The time server receive order: " + body+"; the counter is "+ ++counter
        );

        String currentTime = "QUERY TIME ORDER".equalsIgnoreCase(body) ? new Date(System.currentTimeMillis()).toString()
                : "BAD ORDER";

        currentTime = currentTime + System.getProperty("line.separator");
        ByteBuf resp = Unpooled.copiedBuffer(currentTime.getBytes());
        ctx.writeAndFlush(resp);
    }

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

3. TimeClient.java

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;

import java.net.InetSocketAddress;

public class TimeClient {
    private static final String HOST = "localhost";
    private static final int PORT = 8080;

    public static void main(String[] args) {
        start();
    }

    private static void start() {
        EventLoopGroup group = new NioEventLoopGroup();
        Bootstrap bootstrap = new Bootstrap();
        bootstrap.group(group)
                .channel(NioSocketChannel.class)
                .remoteAddress(new InetSocketAddress(HOST, PORT))
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel socketChannel) throws Exception {
                        socketChannel.pipeline().addLast(new TimeClientHandler());
                    }
                });
        try {
            ChannelFuture future = bootstrap.connect().sync();
            future.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }finally {
            group.shutdownGracefully();
        }
    }
}

4. TimeClientHandler.java

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;

public class TimeClientHandler extends SimpleChannelInboundHandler<ByteBuf> {
    private int counter;
    private byte[] req;
    private int req_times = 100;

    public TimeClientHandler() {
        req = ("QUERY TIME ORDER" + System.getProperty("line.separator")).getBytes();
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        ByteBuf message = null;
        for (int i=0; i<req_times; i++){
            message = Unpooled.buffer(req.length);
            message.writeBytes(req);
            ctx.writeAndFlush(message);
        }
    }

    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, ByteBuf byteBuf) throws Exception {
        byte[] req = new byte[byteBuf.readableBytes()];
        byteBuf.readBytes(req);
        String body = new String(req, "UTF-8");
        System.out.println(
                "Now is : "+ body +" ; the counter is :" + ++counter
        );
    }

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

客户端跟服务端链路创建成功以后，循环发送100条消息，每发送一条就刷新一次，保证每条消息都会被写入Channel中。按照咱们的设计，服务端应该接收到100条查询时间指令的请求消息。

分别运行服务端和客户端，运行结果以下。

服务端运行结果以下。

------------------------------------------------

The time server receive order: QUERY TIME ORDER

此处省略55行 QUERY TIME ORDER

QUERY TIME ORD; the counter is 1
The time server receive order:
此处省略42行 QUERY TIME ORDE

QUERY TIME ORD; the counter is 2
服务端运行结果代表它只接收到了两条消息，第一条包含57条“QUERY TIME ORDER”指令，第二条包含了43条“QUERY TIME ORDER”指令，总数正好是100条。而咱们期待的是收到100条消息，每条包含一条“QUERY TIME ORDER”指令，这说明发生了TCP粘包。

客户端运行结果以下。

------------------------------------------------
Now is : BAD ORDER
BAD ORDER
; the counter is :1

按照设计初衷，客户端应该收到100条当前系统时间的消息，但实际上只收到了一条。这不难理解，由于服务器端只收到了2条请求消息，因此实际服务端只发送了2条应答，因为请求消息不知足查询条件，因此返回了2条“BAD ORDER”应答消息。可是实际上客户端只收到了一条包含2条“BAD ORDER”指令的消息，说明服务端返回的应答消息也发生了粘包。

因为上面的程序没有考虑TCP的粘包/拆包，因此当发生TCP粘包时，咱们的程序就不能正常工做。

下面将演示如何经过Netty的LineBasedFrameDecoder和StringDecoder来解决TCP粘包问题。

利用LineBasedFrameDecoder解决TCP粘包问题

为了解决TCP粘包/拆包致使的半包读写为，Netty默认提供了多种编码器用于处理半包，只要能熟练掌握这些类库的使用，TCP粘包问题今后会变得很是容易，你甚至不须要关心他们，这也是其余NIO框架和JDK原生的NIO API所没法匹敌的。

改进的TimeServer

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.LineBasedFrameDecoder;
import io.netty.handler.codec.string.StringDecoder;

import java.net.InetSocketAddress;

public class TimeServer {
    private final static int PORT = 8080;

    public static void main(String[] args) {
        start();
    }

    private static void start() {
        final TimeServerHandler serverHandler = new TimeServerHandler();
        // 建立EventLoopGroup
        EventLoopGroup bossGroup = new NioEventLoopGroup();
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        // 建立EventLoopGroup
        ServerBootstrap b = new ServerBootstrap();
        b.group(bossGroup, workerGroup)
                //指定所使用的NIO传输Channel
                .channel(NioServerSocketChannel.class)
                .option(ChannelOption.SO_BACKLOG, 1024)
                //使用指定的端口设置套接字地址
                .localAddress(new InetSocketAddress(PORT))
                // 添加一个EchoServerHandler到Channle的ChannelPipeline
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel socketChannel) throws Exception {
                        //EchoServerHandler被标注为@shareable,因此咱们能够老是使用一样的案例
                        socketChannel.pipeline().addLast(new LineBasedFrameDecoder(1024));
                        socketChannel.pipeline().addLast(new StringDecoder());
                        socketChannel.pipeline().addLast(serverHandler);
                    }
                });

        try {
            ChannelFuture f = b.bind().sync();
            f.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}

改进的TimeServerHandler

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandler.Sharable;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;

import java.util.Date;

@Sharable
public class TimeServerHandler extends ChannelInboundHandlerAdapter {
    private int counter;
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        String body = (String) msg;
        System.out.println(
                "The time server receive order: " + body+"; the counter is "+ ++counter
        );

        String currentTime = "QUERY TIME ORDER".equalsIgnoreCase(body) ? new Date(System.currentTimeMillis()).toString()
                : "BAD ORDER";

        currentTime = currentTime + System.getProperty("line.separator");
        ByteBuf resp = Unpooled.copiedBuffer(currentTime.getBytes());
        ctx.writeAndFlush(resp);
    }

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

改进的 TimeClient

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.LineBasedFrameDecoder;
import io.netty.handler.codec.string.StringDecoder;

import java.net.InetSocketAddress;

public class TimeClient {
    private static final String HOST = "localhost";
    private static final int PORT = 8080;

    public static void main(String[] args) {
        start();
    }

    private static void start() {
        EventLoopGroup group = new NioEventLoopGroup();
        Bootstrap bootstrap = new Bootstrap();
        bootstrap.group(group)
                .channel(NioSocketChannel.class)
                .remoteAddress(new InetSocketAddress(HOST, PORT))
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel socketChannel) throws Exception {
                        socketChannel.pipeline().addLast(new LineBasedFrameDecoder(1024));
                        socketChannel.pipeline().addLast(new StringDecoder());
                        socketChannel.pipeline().addLast(new TimeClientHandler());
                    }
                });
        try {
            ChannelFuture future = bootstrap.connect().sync();
            future.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }finally {
            group.shutdownGracefully();
        }
    }
}

改进的TimeClientHandler

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.util.CharsetUtil;

public class TimeClientHandler extends ChannelInboundHandlerAdapter {
    private int counter;
    private byte[] req;
    private int req_times = 100;

    public TimeClientHandler() {
        req = ("QUERY TIME ORDER" + System.getProperty("line.separator")).getBytes();
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        ByteBuf message = null;
        for (int i=0; i<req_times; i++){
            message = Unpooled.buffer(req.length);
            message.writeBytes(req);
            ctx.writeAndFlush(message);
        }
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        String body = (String) msg;
        System.out.println(
                "Now is " + body + " ; the counter is "+ ++counter
        );
    }


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

分别运行TimeServer和TimeClient，执行结果以下。

服务端执行结果以下。

------------------------------------------------

The time server receive order: QUERY TIME ORDER; the counter is 1
// 此处省略2-99行

The time server receive order: QUERY TIME ORDER; the counter is 100

客户端运行结果以下。

------------------------------------------------
Now is Mon Oct 23 17:30:07 CST 2017 ; the counter is 1

// 此处省略2-99行

Now is Mon Oct 23 17:30:07 CST 2017 ; the counter is 100

程序的运行结果安全符合预期，说明经过使用LineBasedFrameDecoder和StringDecoder成功解决了TCP粘包致使的读半包问题。对于使用者来讲，只要将支持半包解码的Handler添加到ChannelPipeline中便可，不须要写额外的代码，用户使用起来很是简单。

LineBasedFrameDecoder和StringDecoder的原理分析

LineBasedFrameDecoder的工做原理是它一次遍历ByteBuf中的可读字节，判断看是否有“\n”或者“\r\n”，若是有，就以此位置为结束位置，从可读索引到结束位置区间的字节就组成了一行。它是以换行符为结束标志的解码器，支持携带结束符或者不携带结束符两种解码方式，同时支持配置单行的最大长度。若是连续读取到最大长度后仍然没有出现换行符，就会抛出异常，同时忽略掉以前读到的异常码流。

StringDecoder的功能很是简单，就是将接收到的对象转换成字符串，而后继续调用后面的Handler。LineBasedFrameDecoder+StringDecoder组合就是按行切换的文本解码器，它被设计用来支持TCP的粘包和拆包。

固然，若是发送的消息不是以换行符结束的，该怎么办呢？或者没有回车换行符，靠消息头中的长度字段来分包怎么办？是否是须要本身写半包解码器？

答案是否认的，Netty提供了多种支持TCP粘包/拆包的解码器用来知足用户的不一样诉求。