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

  • 时间:
  • 来源:互联网
版权声明:本文为博主原创文章,遵循 CC 4.0 BY-SA 版权协议,转载请附上原文出处链接和本声明。
本文链接:https://blog.csdn.net/qq1010234991/article/details/103257013

文章目录

  • 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(); //创建对象,并执行客户端程序
    }
}

本文链接http://element-ui.cn/news/show-68.html