netty 詳解（七）netty 自定義協議解決 TCP 粘包和拆包

目錄：

1、TCP 粘包和拆包基本介紹
2、TCP 粘包和拆包 實例演示
3、netty 自定義協議解決 TCP 粘包和拆包

 

1、TCP 粘包和拆包基本介紹    <--返回目錄

　　TCP 是面向連接的，面向流的，提供高可靠性服務，收發兩端（客戶端和服務器端）都要一 一成對的 socket。因此發送端為了將多個發給接收端的包更有效的發給對方，使用了優化方法（Nagle 算法），將多次間隔較小且數據量小的數據，合並成一個大的數據塊，然后進行封包。這樣雖然提高了效率，但是接收端就難於分辨出完整的數據包了，因為面向流的通信是無消息保護邊界的。

　　由TCP 無消息保護邊界，需要在接收端處理消息邊界問題，也就是我們所說的粘包、拆包問題。

 

2、TCP 粘包和拆包 實例演示    <--返回目錄

　　在編寫 netty 程序時，如果沒有做處理，就會發生粘包和拆包的問題。看一個具體的實例：

 　　Server 端程序

package com.oy.tcp;

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

public class Server {
    public static void main(String[] args) {
        NioEventLoopGroup boss = new NioEventLoopGroup(1);
        NioEventLoopGroup work = new NioEventLoopGroup();

        try {
            ServerBootstrap serverBootstrap = new ServerBootstrap();
            serverBootstrap
                    .group(boss, work)
                    .channel(NioServerSocketChannel.class)
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            ChannelPipeline pipeline = ch.pipeline();
                            pipeline.addLast(new MyServerHandler());
                        }
                    });
            ChannelFuture future = serverBootstrap.bind(8005).sync();
            System.out.println("server started and listen " + 8005);
            future.channel().closeFuture().sync();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            boss.shutdownGracefully();
            work.shutdownGracefully();
        }
    }
}


package com.oy.tcp;

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

import java.util.UUID;

public class MyServerHandler extends SimpleChannelInboundHandler<ByteBuf> {
    private int count;

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) throws Exception {
        //System.out.println("服務器收到的數據：" + msg.toString(CharsetUtil.UTF_8));
        byte[] buffer = new byte[msg.readableBytes()];
        msg.readBytes(buffer);

        // 將 buffer 轉成字符串
        String  message = new String(buffer, CharsetUtil.UTF_8);
        System.out.println("服務器接收數據 " + message);
        System.out.println("服務器接收到消息量= " + (++this.count));

        // 服務器會送數據
        ByteBuf response = Unpooled.copiedBuffer(UUID.randomUUID().toString(), CharsetUtil.UTF_8);
        ctx.writeAndFlush(response);
    }

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

　　Client 端程序

package com.oy.tcp;

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

public class Client {
    public static void main(String[] args) {
        NioEventLoopGroup group = new NioEventLoopGroup();

        try {
            Bootstrap bootstrap = new Bootstrap();
            bootstrap
                    .group(group)
                    .channel(NioSocketChannel.class)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            ChannelPipeline pipeline = ch.pipeline();
                            pipeline.addLast(new MyClientHandler());
                        }
                    });
            ChannelFuture future = bootstrap.connect("127.0.0.1", 8005).sync();
            future.channel().closeFuture().sync();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            group.shutdownGracefully();
        }
    }
}



package com.oy.tcp;

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

public class MyClientHandler extends SimpleChannelInboundHandler<ByteBuf> {
    private int count;

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) throws Exception {
        byte[] buffer = new byte[msg.readableBytes()];
        msg.readBytes(buffer);

        // 將 buffer 轉成字符串
        String  message = new String(buffer, CharsetUtil.UTF_8);
        System.out.println("客戶端接收數據 " + message);
        System.out.println("客戶端接收到消息量= " + (++this.count));
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        // 客戶端發送 10 條數據
        for (int i = 0; i < 10; i++) {
            ByteBuf buffer = Unpooled.copiedBuffer("hello, server " + i, CharsetUtil.UTF_8);
            ctx.writeAndFlush(buffer);
        }
    }

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

 

3、netty 自定義協議解決 TCP 粘包和拆包    <--返回目錄

　　使用 自定義協議 + 編解碼器 來解決。關鍵是要解決服務器每次讀取數據長度的問題，這個問題解決，就不會出現服務器多讀或少讀數據問題，從而避免 TCP 粘包和拆包。

　　案例：

• 要求客戶端發送 5 個 Message 對象，客戶端每次發送一個 Message 對象
• 服務器端每次接收一個 Message 對象，分 5 次進行解碼，每次讀取到一個 Message，會回復一個 Message 對象給客戶端

 

　　封裝傳輸數據類 MessageProtocol

package com.oy.prototcp;

public class MessageProtocol {
    private int len;
    private byte[] content;

    public int getLen() {
        return len;
    }

    public void setLen(int len) {
        this.len = len;
    }

    public byte[] getContent() {
        return content;
    }

    public void setContent(byte[] content) {
        this.content = content;
    }
}

　　編碼器

package com.oy.prototcp;

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.MessageToByteEncoder;

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

　　解碼器

package com.oy.prototcp;

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ReplayingDecoder;

import java.util.List;

public class MessageDecoder extends ReplayingDecoder<Void> {
    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
        System.out.println("MessageDecoder#decode() 被調用");
        // 將得到的二進制字節碼 =》 MessageProtocol 對象
        int len = in.readInt();
        byte[] content = new byte[len];
        in.readBytes(content);
        // 封裝成 MessageProtoCol 對象，傳遞給下一個 handler
        MessageProtocol messageProtocol = new MessageProtocol();
        messageProtocol.setLen(len);
        messageProtocol.setContent(content);
        out.add(messageProtocol);
    }
}

　　服務器端  Server

package com.oy.prototcp;

import com.google.gson.internal.$Gson$Preconditions;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;

public class Server {
    public static void main(String[] args) {
        NioEventLoopGroup boss = new NioEventLoopGroup(1);
        NioEventLoopGroup work = new NioEventLoopGroup();

        try {
            ServerBootstrap serverBootstrap = new ServerBootstrap();
            serverBootstrap
                    .group(boss, work)
                    .channel(NioServerSocketChannel.class)
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            ChannelPipeline pipeline = ch.pipeline();
                            pipeline.addLast(new MessageDecoder()); // 解碼器
                            pipeline.addLast(new MyServerHandler());
                        }
                    });
            ChannelFuture future = serverBootstrap.bind(8005).sync();
            System.out.println("server started and listen " + 8005);
            future.channel().closeFuture().sync();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            boss.shutdownGracefully();
            work.shutdownGracefully();
        }
    }
}

　　MyServerHandler

package com.oy.prototcp;

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;

import java.nio.charset.Charset;

public class MyServerHandler extends SimpleChannelInboundHandler<MessageProtocol> {
    private int count;

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, MessageProtocol msg) throws Exception {
        int len = msg.getLen();
        byte[] content = msg.getContent();
        System.out.println("服務器接收到數據. 長度：" + len + ", 內容：" + new String(content, Charset.forName("utf-8")));

        System.out.println("服務器接收到消息包數量" + (++this.count));
    }
}

　　客戶端 Client

package com.oy.prototcp;

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

public class Client {
    public static void main(String[] args) {
        NioEventLoopGroup group = new NioEventLoopGroup();

        try {
            Bootstrap bootstrap = new Bootstrap();
            bootstrap
                    .group(group)
                    .channel(NioSocketChannel.class)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            ChannelPipeline pipeline = ch.pipeline();
                            pipeline.addLast(new MessageEncoder()); // 編碼器
                            pipeline.addLast(new MyClientHandler());
                        }
                    });
            ChannelFuture future = bootstrap.connect("127.0.0.1", 8005).sync();
            future.channel().closeFuture().sync();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            group.shutdownGracefully();
        }
    }
}

　　MyClientHandler

package com.oy.prototcp;

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;

import java.nio.charset.Charset;

public class MyClientHandler extends SimpleChannelInboundHandler<MessageProtocol> {
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, MessageProtocol msg) throws Exception {

    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        for (int i = 0; i < 5; i++) {
            String messsage = "您好，吃了沒";
            byte[] content = messsage.getBytes(Charset.forName("utf-8"));
            int len = messsage.getBytes(Charset.forName("utf-8")).length;
            // 創建協議包對象
            MessageProtocol messageProtocol = new MessageProtocol();
            messageProtocol.setContent(content);
            messageProtocol.setLen(len);
            ctx.writeAndFlush(messageProtocol);
        }
    }

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

 ---