netty自定义简单解码器处理粘包、拆包

tcp连接的粘包、拆包发生在长连接中，先了解一下长、短连接的概念

短连接：请求/响应之后，关闭已经建立的tcp连接，下次请求再建立新的连接

长连接：请求/响应之后，不关闭已经建立的tcp连接，多次请求，复用同一个连接

粘包：Nagle算法，客户端累积一定量或者缓冲一段时间再传输。服务端缓冲区堆积，导致多个请求粘在一起

拆包：发送的请求大于发送缓冲区，进行分片传输。服务端缓冲区堆积，导致服务端读取的请求数据不完成

可以模拟粘包场景，新建一个socket工程如下所示

import java.io.IOException;
import java.io.OutputStream;
import java.net.InetSocketAddress;
import java.net.Socket;

public class Client {

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

        Socket socket = new Socket();
        socket.connect(new InetSocketAddress(10000));

        OutputStream outputStream = socket.getOutputStream();

        byte[] request = new byte[200];
        byte[] message = "测试测试测试测试测试测试测试测试测试测试测试测试测试测试测试测试测试".getBytes();
        System.arraycopy(message, 0, request, 0, message.length);


        //开十个线程向服务端发送消息
        for (int i = 0; i < 10; i++) {
            new Thread(() -> {
                try {
                    outputStream.write(request);
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }).start();
        }

    }
}

新建一个netty server如下所示

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

public class NettyServer {

    public static void main(String[] args) {

        ServerBootstrap serverBootstrap = new ServerBootstrap();

        EventLoopGroup eventLoopGroup = new NioEventLoopGroup();
        EventLoopGroup workerGroup = new NioEventLoopGroup();

        try {
            serverBootstrap.group(eventLoopGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .localAddress(10000)
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel socketChannel) throws Exception {
                            ChannelPipeline pipeline = socketChannel.pipeline();
                            pipeline.addLast(new Handler());
                        }
                    });
            ChannelFuture future = serverBootstrap.bind().syncUninterruptibly();
            future.channel().closeFuture().syncUninterruptibly();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            eventLoopGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}

新建处理客户端消息的Handler，这里只是简单的将消息打印出来

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;

public class Handler extends ChannelInboundHandlerAdapter {

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {

        if (msg instanceof ByteBuf) {
            byte[] bytes = new byte[((ByteBuf) msg).readableBytes()];
            ((ByteBuf) msg).readBytes(bytes);
            System.out.println(new String(bytes));
        } else {
            System.out.println(new String((byte[]) msg));
        }

        ctx.fireChannelRead(msg);
    }
}

运行程序后，服务端打印的数据如下，只打印了两条消息，这显然不是我们想要的结果

对于这种情况，如果我们需要自己处理的话，可以继承netty提供的ByteToMessageDecoder类并实现其decode方法，这其实就是所谓的编码解码过程。

代码如下所示

import java.util.List;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ByteToMessageDecoder;

public class Decoder extends ByteToMessageDecoder {

    //为了简单处理，假设协议为每次固定传输200字节
    private static final int POCKET_SIZE = 200;

    //记录上次未读完的字节
    private ByteBuf tempMessage = Unpooled.buffer();


    @Override
    protected void decode(ChannelHandlerContext channelHandlerContext, ByteBuf byteBuf, List<Object> list) throws Exception {
        int inSize = byteBuf.readableBytes();

        System.out.println("=========收到" + inSize + "字节========");
        ByteBuf inMessage;

        //加上上次未读取完成的字节
        if (tempMessage.readableBytes() == 0) {
            inMessage = byteBuf;
        } else {
            inMessage = Unpooled.buffer();
            inMessage.writeBytes(tempMessage);
            inMessage.writeBytes(byteBuf);
        }

        int counter = inMessage.readableBytes() / POCKET_SIZE;

        for (int i = 0; i < counter; i++) {
            byte[] bytes = new byte[POCKET_SIZE];
            inMessage.readBytes(bytes);
            //将处理的好的消息放入list中向下传递
            list.add(bytes);
        }

        tempMessage.clear();
        if (inMessage.readableBytes() != 0) {
            inMessage.readBytes(tempMessage, inMessage.readableBytes());
        }

    }
}

 解码器编写完成之后，还需要再server启动时加上，代码如下所示：

serverBootstrap.group(eventLoopGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .localAddress(10000)
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel socketChannel) throws Exception {
                            ChannelPipeline pipeline = socketChannel.pipeline();
                            pipeline.addLast(new Decoder());
                            pipeline.addLast(new Handler());
                        }
                    });