一、是什么？

客户端通过socket给服务端发送数据，为了传输更有效率，会将多次间隔较小的且数据量小的数据，通过nagle算法，合并成一个大的数据块，然后进行封包。这样做提高了效率，缺点就是你发送到服务端的数据，服务端不知道是不是完整的，不知道哪几小块数据拼起来才是原来的数据。举个例子：客户端要发送原信息是A和B两个数据包，服务端接收到之后，可能出现如下情况：

• 正常情况：读取到了A和B两个数据包；
• 粘包：A和B两个数据包一起读取了；
• 拆包：读取了A数据包的一部分，A的另一部分和B数据包一起读取了；

由于TCP是没有消息保护边界的，也就是上面的消息，没有边界，服务端并不知道hello的o是一个边界，hello是一个单词，所以我们就得中服务端处理边界问题。这也就是粘包拆包问题。

二、Netty中的粘包拆包如何解决

• 使用自定义协议 +　编解码器来解决。说人话就是：服务端你不是不知道消息的长度吗？那我就让客户端发送的消息封装成一个对象，对象包括消息长度和消息内容，服务端读取的时候通过对象就可以拿到每次读取的长度了。

下面看具体案例：

• 封装消息对象 MessageProtocol.java：

@Data
public class MessageProtocol {
    private int len; // 长度
    private byte[] content; // 发送的内容
}

• 解码器 MessageDecoder.java：

public class MessageDecoder extends ReplayingDecoder<Void> {
    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
        System.out.println("MessageDecoder.decode 被调用");
        // 将byte转成MessageProtocol对象
        MessageProtocol msg = new MessageProtocol();
        int len = in.readInt();
        byte[] content = new byte[len];
        in.readBytes(content);
        msg.setContent(content);
        msg.setLen(len);
        // 放入到out中传递给下一个handler处理
        out.add(msg);
    }
}

• 编码器 MessageEncoder.java：

public class MessageEncoder extends MessageToByteEncoder<MessageProtocol> {
    @Override
    protected void encode(ChannelHandlerContext ctx, MessageProtocol msg, ByteBuf out) throws Exception {
        System.out.println("MessageEncoder.encode被调用");
        out.writeInt(msg.getLen());
        out.writeBytes(msg.getContent());
    }
}

• 客户端 --- NettyClient.java：

public class NettyClient {
    public static void main(String[] args) throws Exception {
        // 1. 创建事件循环组
        EventLoopGroup eventLoopGroup = new NioEventLoopGroup();
        try {
            // 2. 创建启动对象
            Bootstrap bootstrap = new Bootstrap();
            // 3. 设置相关参数
            bootstrap.group(eventLoopGroup) // 设置线程组
                     .channel(NioSocketChannel.class) // 设置通道
                     .handler(new NettyClientInitializer());
            // 4. 连接服务端
            ChannelFuture channelFuture = bootstrap.connect("127.0.0.1", 6666).sync();
            // 5. 监听通道关闭
            channelFuture.channel().closeFuture().sync();
        } finally {
            eventLoopGroup.shutdownGracefully();
        }   
    }
}

• 客户端 --- NettyClientInitializer.java：

public class NettyClientInitializer extends ChannelInitializer<SocketChannel>{
    @Override
    protected void initChannel(SocketChannel sc) throws Exception {
        ChannelPipeline pipeline = sc.pipeline();
        pipeline.addLast(new MessageEncoder());
        pipeline.addLast(new MessageDecoder());
        pipeline.addLast(new NettyClientHandler());
    }
}

• 客户端 --- NettyClientHandler.java：

public class NettyClientHandler extends SimpleChannelInboundHandler<MessageProtocol>{

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        // 发送10条数据
        for (int i=0; i<5; i++) {
            String msg = "hello " + i;
            byte[] bys = msg.getBytes("utf-8");
            int len = msg.getBytes("utf-8").length;
            // 创建协议包
            MessageProtocol message = new MessageProtocol();
            message.setLen(len);
            message.setContent(bys);
            // 发送
            ctx.writeAndFlush(message);
        }
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ch, MessageProtocol msg) throws Exception {
        int len = msg.getLen();
        byte[] bys = msg.getContent();
        System.out.println("客户端收到消息：长度 = " + len + ", 内容 = " + new String(bys, Charset.forName("utf-8")));
    }
}

• 服务端 NettyServer.java：

public class NettyServer {
    public static void main(String[] args) throws Exception {
        // 1. 创建boss group (boss group和work group含有的子线程数默认是cpu数 * 2)
        EventLoopGroup bossGroup = new NioEventLoopGroup();
        // 2. 创建work group
        EventLoopGroup workGroup = new NioEventLoopGroup();
        try {
            // 3. 创建服务端启动对象
            ServerBootstrap bootstrap = new ServerBootstrap();
            // 4. 配置启动参数
            bootstrap.group(bossGroup, workGroup) // 设置两个线程组
                     .channel(NioServerSocketChannel.class) // 使用NioSocketChannel 作为服务器的通道
                     .childHandler(new NettyServerInitializer());
            // 5. 启动服务器并绑定端口
            ChannelFuture cf = bootstrap.bind(6666).sync();
            // 6. 对关闭通道进行监听
            cf.channel().closeFuture().sync();
        } finally {
            bossGroup.shutdownGracefully();
            workGroup.shutdownGracefully();
        }
    }
}

• 服务端 NettyServerInitializer.java：

public class NettyServerInitializer extends ChannelInitializer<SocketChannel>{
    @Override
    protected void initChannel(SocketChannel sc) throws Exception {
        sc.pipeline().addLast(new MessageDecoder());
        sc.pipeline().addLast(new MessageEncoder());
        sc.pipeline().addLast(new NettyServerHandler());
    }
}

• 服务端 NettyServerHandler.java：

public class NettyServerHandler extends SimpleChannelInboundHandler<MessageProtocol> {
    private int count;
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, MessageProtocol msg) throws Exception {
        // 接收数据并处理
        int len = msg.getLen();
        byte[] bys = msg.getContent();
        System.out.println("服务端第" + (++count) + "次收到消息：长度 = " + len + ", 内容 = " + new String(bys, Charset.forName("utf-8")));
        
        // 给客户端回复消息
        String responseContent = UUID.randomUUID().toString();
        byte[] rbys = responseContent.getBytes("utf-8");
        int rlen = responseContent.getBytes("utf-8").length;
        MessageProtocol rmsg = new MessageProtocol();
        rmsg.setContent(rbys);
        rmsg.setLen(rlen);
        ctx.writeAndFlush(rmsg);
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        System.out.println(cause.getMessage());
        ctx.close();
    }
}

怎么样，是不是很简单呢！