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

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类型的channelb.channel(NioSocketChannel.class);//设置监听端口b.remoteAddress(serverIp, serverPort);//设置通道的参数b.option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);//装配子通道流水线b.handler(new ChannelInitializer<SocketChannel>() {//有链接到达时会创建一个channel@Overrideprotected 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++) {//发送ByteBufByteBuf 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. 可以自定义解码器分包器

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

2. 使用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 {//谷歌 Gsonstatic Gson gson = null;static {//不需要html escapegson=new GsonBuilder().disableHtmlEscaping()
//                .excludeFieldsWithoutExposeAnnotation().create();}//Object对象转成JSON字符串后，进一步转成字节数组public static byte[] Object2JsonBytes(Object obj) {//把对象转换成JSONString 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@Testpublic void serAndDesr() throws IOException {JsonMsg message = buildMsg(); //创建POJO对象//将POJO对象，序列化成字符串String json = message.convertToJson(); //转成json字符串//可以用于网络传输,保存到内存或外存Logger.info("json:=" + json);//JSON 字符串,反序列化成对象POJOJsonMsg 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类型的channelb.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@Overrideprotected 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 {@Overridepublic 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类型的channelb.channel(NioSocketChannel.class);//3 设置监听端口b.remoteAddress(serverIp, serverPort);//4 设置通道的参数b.option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);//5 装配通道流水线b.handler(new ChannelInitializer<SocketChannel>() {//初始化客户端channelprotected 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(); //创建对象，并执行客户端程序}
}