Netty-粘包，半包和使用Json方式传输数据

不念不忘少年蓝@ 2023-06-16 11:59 5阅读 0赞

### 文章目录 ###

*  JSON和ProtoBuf序列化
 *   *  背景
     *  详解粘包和拆包
     *   *  粘包和半包
         *   *  半包问题的实践案例
             *   *  运行结果
                 *  半包
                 *  粘包
                 *  解决
     *  JSON协议通信
     *   *  code
         *   *  Json序列化与反序列化实践案例
             *   *  JsonMsg
                 *  JsonMsgDemo
             *  JSON传输至服务器端的实践案例
             *   *  code
                 *  客户端实践案例

# JSON和ProtoBuf序列化 #

## 背景 ##

> 在开发一些远程过程调用(RPC)的程序时，通常会涉及对象的序列化/反序列化的问题， 例如一个对象从客户端通过TCP方式发送到服务器端，因为TCP协议只能发送字节流，数据接收端在反序列化成Java POJO对象即可

## 详解粘包和拆包 ##

### 粘包和半包 ###

#### 半包问题的实践案例 ####

package com.wangyg.NettyDemos.NettyDumpSend;
    
    import com.wangyg.netty.basic.NettyDemoConfig;
    import io.netty.bootstrap.Bootstrap;
    import io.netty.buffer.ByteBuf;
    import io.netty.channel.*;
    import io.netty.channel.nio.NioEventLoopGroup;
    import io.netty.channel.socket.SocketChannel;
    import io.netty.channel.socket.nio.NioSocketChannel;
    import io.netty.buffer.PooledByteBufAllocator;
    import util.Logger;
    
    import java.nio.charset.Charset;
    
    public class NettyDumpSendClient {
        private  int serverPort;
        private String serverIp;
    
        Bootstrap b = new Bootstrap();
    
        /**
         * 构造函数
         * @param serverIp
         * @param serverPort
         */
        public NettyDumpSendClient(String serverIp, int serverPort ) {
            this.serverPort = serverPort;
            this.serverIp = serverIp;
        }
        public void runCLient(){
            //创建反应器线程组
            EventLoopGroup workerLoopGroup = new NioEventLoopGroup();
            try {
                //设置reactor线程组
                b.group(workerLoopGroup);
                //设置nio类型的channel
                b.channel(NioSocketChannel.class);
                //设置监听端口
                b.remoteAddress(serverIp, serverPort);
                //设置通道的参数
                b.option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
                //装配子通道流水线
                b.handler(new ChannelInitializer<SocketChannel>() {
                    //有链接到达时会创建一个channel
                    @Override
                    protected void initChannel(SocketChannel ch) throws Exception {
                        //pipeLinde管理子通道channel中的handler
                        //向子channel 流水线添加一个Handler处理器
                        ch.pipeline().addLast(NettyEchoClientHandler.INSTANCE);
                    }
                });
                ChannelFuture f = b.connect();
                f.addListener((ChannelFuture future) ->{
                    if (future.isSuccess()) {
                        Logger.info("Echo client 客户端连接成功!");
                    } else{
                        Logger.info("Echo Client 客户端连接失败! ");
                    }
                });
    
                //阻塞，直到连接完成
                f.sync();
                Channel channel = f.channel();
    
                byte[] bytes = "疯狂创客圈：高性能学习社群!".getBytes(Charset.forName("utf-8"));
                for (int i = 0; i < 1000; i++) {
                    //发送ByteBuf
                    ByteBuf buffer = channel.alloc().buffer();
                    buffer.writeBytes(bytes);
                    channel.writeAndFlush(buffer);
                }
    
    
                // 7 等待通道关闭的异步任务结束
                // 服务监听通道会一直等待通道关闭的异步任务结束
                ChannelFuture closeFuture =channel.closeFuture();
                closeFuture.sync();
    
    
            } catch (Exception e) {
                e.printStackTrace();
            }finally {
                //优雅关闭
                workerLoopGroup.shutdownGracefully(); //进行有噶释放所有资源包括创建的线程
            }
        }
    
        /**
         * 入口程序
         * @param args
         */
        public static void main(String[] args) {
            int port  =8888;
            String ip = "127.0.0.1";
    
            new NettyDumpSendClient(ip, port).runCLient();
        }
    
    }

##### 运行结果 #####

\[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-s5ELrFzd-1574752594196)(0F59D7E558CB4A1E91C368E1B0A2E54E)\]

##### 半包 #####

每次读取底层缓冲的数据容量是有限的，当TCP底层缓冲的数据包比较大时，会将一个底层包分成多次ByteBuf进行复制，进而保证进程缓冲区读到的是半包

##### 粘包 #####

当TCP底层缓冲的数据包比较小时，一次复制的却不知一个内核缓冲区包，进而造成进程缓冲区读到的是粘包

##### 解决 #####

两种方法

1.  可以自定义解码器分包器

> 定义自己的进程缓冲区分包器

1.  使用Netty内置的解码器

> 使用Netty内置的LengthFieldBasedFrameDecoder自定义分隔符数据包解码器，对进程缓冲区ByteBuf进行正确的分包

## JSON协议通信 ##

> Java的JSON数据由三个比较流行的开源库： FastJson, Gson Jackson

> 实际开发中，目前驻留的策略是：Google的Gson库和阿里的FastJson库两者结合使用

POJO对象转为Json字符串的应用场景

> 使用Google的 Gson

JSon字符串反序列化成POJO的应用场景

> 使用阿里的FastJson

### code ###

package com.wangyg.netty.basic.util;
    
    import com.alibaba.fastjson.JSONObject;
    import com.google.gson.Gson;
    import com.google.gson.GsonBuilder;
    
    import java.io.UnsupportedEncodingException;
    
    public class JsonUtil {
    
        //谷歌 Gson
        static Gson gson = null;
    
        static {
            //不需要html escape
            gson=new GsonBuilder()
                    .disableHtmlEscaping()
    //                .excludeFieldsWithoutExposeAnnotation()
                    .create();
    
        }
    
        //Object对象转成JSON字符串后，进一步转成字节数组
        public static byte[] Object2JsonBytes(Object obj) {
    
            //把对象转换成JSON
    
            String json = pojoToJson(obj);
            try {
                return json.getBytes("UTF-8");
            } catch (UnsupportedEncodingException e) {
                e.printStackTrace();
            }
            return null;
        }
    
        //反向：字节数组，转成JSON字符串，转成Object对象
        public static <T> T JsonBytes2Object(byte[] bytes, Class<T> tClass) {
            //字节数组，转成JSON字符串
            try {
                String json = new String(bytes, "UTF-8");
                T t = jsonToPojo(json, tClass);
                return t;
            } catch (Exception e) {
                e.printStackTrace();
            }
            return null;
        }
    
        //使用谷歌 Gson 将 POJO 转成字符串
        public static String pojoToJson(Object obj) {
            //String json = new Gson().toJson(obj);
            String json = gson.toJson(obj);
    
            return json;
        }
    
        //使用阿里 Fastjson 将字符串转成 POJO对象
        public static <T> T jsonToPojo(String json, Class<T> tClass) {
            T t = JSONObject.parseObject(json, tClass);
            return t;
        }
    }

#### Json序列化与反序列化实践案例 ####

##### JsonMsg #####

package com.wangyg.NettyDemos.MyJsonUtils;
    
    import util.JsonUtil;
    public class JsonMsg {
        //id Field(域)
        private int id;
        //content Field(域)
        private String content;
        //在通用方法中，使用阿里FastJson转成Java对象
        public static JsonMsg parseFromJson(String json) {
            return JsonUtil.jsonToPojo(json, JsonMsg.class);
        }
        //在通用方法中，使用谷歌Gson转成字符串
        public String convertToJson() {
            return JsonUtil.pojoToJson(this);
        }
    
    
        public int getId() {
            return id;
        }
        public void setId(int id) {
            this.id = id;
        }
        public String getContent() {
            return content;
        }
        public void setContent(String content) {
            this.content = content;
        }
    }

##### JsonMsgDemo #####

1.  通过谷歌的Gson框架，将POJO序列化成JSON字符串
2.  使用Stringencoder 编码器(Netty内置)将JSON字符串编码成二进制字节数组
3.  使用LengthFieldPrepender编码器(Netty内置)将二进制字节数组编码成Head-Content格式的二进制数据包

/**
     * 使用POJO 类JsonMsg  实现从POJO对象到Json的序列化，反序列化的实践案例
     */
    public class JsonMsgDemo {
    
        //构建Json对象
        public JsonMsg buildMsg() {
            JsonMsg user = new JsonMsg();
            user.setId(1000);
            user.setContent("疯狂创客圈:高性能学习社群");
            return user;
        }
    
        //序列化 serialization & 反序列化 Deserialization
        @Test
        public void serAndDesr() throws IOException {
            JsonMsg message = buildMsg(); //创建POJO对象
            //将POJO对象，序列化成字符串
            String json = message.convertToJson(); //转成json字符串
            //可以用于网络传输,保存到内存或外存
            Logger.info("json:=" + json);
    
            //JSON 字符串,反序列化成对象POJO
            JsonMsg inMsg = JsonMsg.parseFromJson(json); //然后再讲json字符串转成pojo对象
            Logger.info("id:=" + inMsg.getId());  //输出对象中的属性值
            Logger.info("content:=" + inMsg.getContent());
        }
    }

#### JSON传输至服务器端的实践案例 ####

> 为了清晰的演示Json传输，设计一个简单的客户端/服务器传输程序： 服务器接收客户端的数据，并解码成Json, 在转为POJO， 客户端将POJO转成JSON字符串，编码后在传送到服务器端

##### code #####

package com.wangyg.NettyDemos.MyJsonUtils.JsonServer;
    
    import com.wangyg.NettyDemos.MyJsonUtils.JsonMsg;
    import com.wangyg.channel.NioDemoConfig;
    import io.netty.bootstrap.ServerBootstrap;
    import io.netty.buffer.PooledByteBufAllocator;
    import io.netty.channel.*;
    import io.netty.channel.nio.NioEventLoopGroup;
    import io.netty.channel.socket.SocketChannel;
    import io.netty.channel.socket.nio.NioServerSocketChannel;
    import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
    import io.netty.handler.codec.string.StringDecoder;
    import io.netty.util.CharsetUtil;
    import util.Logger;
    
    
    /**
     * 服务器端的程序仅仅读取客户端数据包并完成解码，服务器端的程序没有写出任何的输出数据包对端（客户端）
     *  */
    
    public class JsonServer {
        private final int serverPort;
    
        ServerBootstrap b = new ServerBootstrap();
    
        /**
         * 构造函数
         * @param port
         */
        public JsonServer(int port) {
            this.serverPort = port;
        }
    
        public void runServer(){
            //创建reactor线程组
            EventLoopGroup bossLoopGroup = new NioEventLoopGroup(1);
            EventLoopGroup workderLoopGroup = new NioEventLoopGroup(); //使用默认cpu个数*2 的线程个数
    
            try {
                //设置reacotor线程组
                b.group(bossLoopGroup, workderLoopGroup);
                //设置Nio类型的channel
                b.channel(NioServerSocketChannel.class); //服务器端的serverSocket
                //设置监听端口
                b.localAddress(serverPort); //设置监听本地地址，并且传入端口
                //设置通道的参数
                b.option(ChannelOption.SO_KEEPALIVE, true);
                b.option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
                b.childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
    
                //装配子通道流水线
                b.childHandler(new ChannelInitializer<SocketChannel>() {
                    //有连接到达时会创建一个channel
                    @Override
                    protected void initChannel(SocketChannel ch) throws Exception {
                        //pipeline管离子通道channel中的handler
                        //向子channel流水线添加3个handler处理器
                        ch.pipeline().addLast(new LengthFieldBasedFrameDecoder(1024, 0, 4, 0, 4));
                        ch.pipeline().addLast(new StringDecoder(CharsetUtil.UTF_8));
                        ch.pipeline().addLast(new JsonMsgDecoder());
                    }
                });
    
                //开始绑定server
                //通过sync同步方法阻塞直到绑定陈宫
                ChannelFuture channelFuture = b.bind().sync();
                Logger.info("服务器启动陈宫，监听端口： " + channelFuture.channel().localAddress());
    
                //等待通道关闭的异步任务结束
                //服务器监听通道一致等待通道关闭的异步任务结束
                ChannelFuture closeFuture = channelFuture.channel().closeFuture();
                closeFuture.sync();
    
            } catch (Exception e) {
                e.printStackTrace();
            }finally {
                //优雅关闭eventLoopGroup 线程
                workderLoopGroup.shutdownGracefully();
                bossLoopGroup.shutdownGracefully();
            }
        }
    
    
        //服务器端业务处理器
        static class JsonMsgDecoder extends ChannelInboundHandlerAdapter {
            @Override
            public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                String json = (String) msg;
                JsonMsg jsonMsg = JsonMsg.parseFromJson(json);
                Logger.info("收到一个 Json 数据包 =》" + jsonMsg);
            }
        }
    
        /**
         * 程序执行入口
         * @param args
         */
        public static void main(String[] args) {
            int port = 8888;
            new JsonServer(port).runServer();
        }
    
    }

##### 客户端实践案例 #####

package com.wangyg.NettyDemos.MyJsonUtils.JsonServer;
    
    
    
    import com.wangyg.NettyDemos.MyJsonUtils.JsonMsg;
    import com.wangyg.netty.basic.NettyDemoConfig;
    import io.netty.bootstrap.Bootstrap;
    import io.netty.buffer.PooledByteBufAllocator;
    import io.netty.channel.*;
    import io.netty.channel.nio.NioEventLoopGroup;
    import io.netty.channel.socket.SocketChannel;
    import io.netty.channel.socket.nio.NioSocketChannel;
    import io.netty.handler.codec.LengthFieldPrepender;
    import io.netty.handler.codec.string.StringEncoder;
    import io.netty.util.CharsetUtil;
    import util.Logger;
    
    public class JsonSendClient {
        static String content = "疯狂创客圈：高性能学习社群!";
    
        private int serverPort; //端口号
        private String serverIp; //服务器端ip地址
    
        Bootstrap b = new Bootstrap();
    
        /**
         * 构造函数
         * @param serverIp
         * @param serverPort
         */
        public JsonSendClient( String serverIp, int serverPort) {
            this.serverPort = serverPort;
            this.serverIp = serverIp;
        }
    
    
        public void runClient() {
            //创建reactor 线程组
            EventLoopGroup workerLoopGroup = new NioEventLoopGroup();
    
            try {
                //1 设置reactor 线程组
                b.group(workerLoopGroup);
                //2 设置nio类型的channel
                b.channel(NioSocketChannel.class);
                //3 设置监听端口
                b.remoteAddress(serverIp, serverPort);
                //4 设置通道的参数
                b.option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
    
                //5 装配通道流水线
                b.handler(new ChannelInitializer<SocketChannel>() {
                    //初始化客户端channel
                    protected void initChannel(SocketChannel ch) throws Exception {
                        // 客户端channel流水线添加2个handler处理器
                        ch.pipeline().addLast(new LengthFieldPrepender(4));
                        ch.pipeline().addLast(new StringEncoder(CharsetUtil.UTF_8));
                    }
                });
                ChannelFuture f = b.connect();
                f.addListener((ChannelFuture futureListener) ->
                {
                    if (futureListener.isSuccess()) {
                        Logger.info("EchoClient客户端连接成功!");
                    } else {
                        Logger.info("EchoClient客户端连接失败!");
                    }
                });
    
                // 阻塞,直到连接完成
                f.sync();
                Channel channel = f.channel();
    
                //发送 Json 字符串对象
                for (int i = 0; i < 1000; i++) {
                    JsonMsg user = build(i, i + "->" + content);
                    channel.writeAndFlush(user.convertToJson());
                    Logger.info("发送报文：" + user.convertToJson());
                }
                channel.flush();
    
    
                // 7 等待通道关闭的异步任务结束
                // 服务监听通道会一直等待通道关闭的异步任务结束
                ChannelFuture closeFuture = channel.closeFuture();
                closeFuture.sync();
    
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                // 优雅关闭EventLoopGroup，
                // 释放掉所有资源包括创建的线程
                workerLoopGroup.shutdownGracefully();
            }
    
        }
    
    
        public JsonMsg build(int id, String content) {
            JsonMsg user = new JsonMsg();
            user.setId(id);
            user.setContent(content);
            return user;
        }
    
        /**
         * 客户端程序入口
         * @param args
         */
        public static void main(String[] args) {
            int port = 8888;
            String ip = "127.0.0.1";
            new JsonSendClient(ip, port).runClient(); //创建对象，并执行客户端程序
        }
    }