[toc]
Netty中使用MessagePack时的TCP粘包问题与解决方案
通过下面的实例代码来演示在Netty中使用MessagPack时会出现的TCP粘包问题，为了学习的连贯性，参考了《Netty权威指南》第7章中的代码，但是需要注意的是，书中并没有提供完整代码，提供的代码都是片段性的，所以我根据自己的理解把服务端的代码和客户端的代码写了出来，可以作为参考。
仍然需要注意的是，我使用的是Netty 4.x的版本。
另外我在程序代码中写了非常详细的注释，所以这里不再进行更多的说明。
在使用MessagePack时的TCP粘包问题
编码器与解码器
MsgpackEncoder.java
package cn.xpleaf.msgpack;import org.msgpack.MessagePack;import io.netty.buffer.ByteBuf;import io.netty.channel.ChannelHandlerContext;import io.netty.handler.codec.MessageToByteEncoder;/** * MsgpackEncoder继承自Netty中的MessageToByteEncoder类， * 并重写抽象方法encode(ChannelHandlerContext ctx, Object msg, ByteBuf out) * 它负责将Object类型的POJO对象编码为byte数组，然后写入到ByteBuf中 * @author yeyonghao * */public class MsgpackEncoder extends MessageToByteEncoder<Object> { @Override protected void encode(ChannelHandlerContext ctx, Object msg, ByteBuf out) throws Exception { // 创建MessagePack对象 MessagePack msgpack = new MessagePack(); // 将对象编码为MessagePack格式的字节数组 byte[] raw = msgpack.write(msg); // 将字节数组写入到ByteBuf中 out.writeBytes(raw); }}MsgpackDecoder.java
package cn.xpleaf.msgpack;import java.util.List;import org.msgpack.MessagePack;import io.netty.buffer.ByteBuf;import io.netty.channel.ChannelHandlerContext;import io.netty.handler.codec.ByteToMessageDecoder;import io.netty.handler.codec.MessageToMessageDecoder;/** * MsgpackDecoder继承自Netty中的MessageToMessageDecoder类， * 并重写抽象方法decode(ChannelHandlerContext ctx, ByteBuf msg, List<Object> out) * 首先从数据报msg（数据类型取决于继承MessageToMessageDecoder时填写的泛型类型）中获取需要解码的byte数组 * 然后调用MessagePack的read方法将其反序列化（解码）为Object对象 * 将解码后的对象加入到解码列表out中，这样就完成了MessagePack的解码操作 * @author yeyonghao * */public class MsgpackDecoder extends MessageToMessageDecoder<ByteBuf> { @Override protected void decode(ChannelHandlerContext ctx, ByteBuf msg, List<Object> out) throws Exception { // 从数据报msg中（这里的数据类型为ByteBuf，因为Netty的通信基于ByteBuf对象） final byte[] array; final int length = msg.readableBytes(); array = new byte[length]; /** * 这里使用的是ByteBuf的getBytes方法来将ByteBuf对象转换为字节数组，前面是使用readBytes，直接传入一个接收的字节数组参数即可 * 这里的参数比较多，第一个参数是index，关于readerIndex，说明如下： * ByteBuf是通过readerIndex跟writerIndex两个位置指针来协助缓冲区的读写操作的，具体原理等到Netty源码分析时再详细学习一下 * 第二个参数是接收的字节数组 * 第三个参数是dstIndex the first index of the destination * 第四个参数是length the number of bytes to transfer */ msg.getBytes(msg.readerIndex(), array, 0, length); // 创建一个MessagePack对象 MessagePack msgpack = new MessagePack(); // 解码并添加到解码列表out中 out.add(msgpack.read(array)); }}服务端
EchoServer.java
package cn.xpleaf.echo;import cn.demo.simple.MsgPackDecode;import cn.xpleaf.msgpack.MsgpackDecoder;import cn.xpleaf.msgpack.MsgpackEncoder;import io.netty.bootstrap.ServerBootstrap;import io.netty.channel.ChannelFuture;import io.netty.channel.ChannelInitializer;import io.netty.channel.ChannelOption;import io.netty.channel.EventLoopGroup;import io.netty.channel.nio.NioEventLoopGroup;import io.netty.channel.socket.SocketChannel;import io.netty.channel.socket.nio.NioServerSocketChannel;public class EchoServer { public void bind(int port) throws Exception { // 配置服务端的NIO线程组 EventLoopGroup bossGroup = new NioEventLoopGroup(); EventLoopGroup workerGroup = new NioEventLoopGroup(); try { ServerBootstrap b = new ServerBootstrap(); b.group(bossGroup, workerGroup) .channel(NioServerSocketChannel.class) .option(ChannelOption.SO_BACKLOG, 1024) .childHandler(new ChannelInitializer<SocketChannel>() { @Override protected void initChannel(SocketChannel ch) throws Exception { // 添加MesspagePack解码器 ch.pipeline().addLast("msgpack decoder", new MsgPackDecode()); // 添加MessagePack编码器 ch.pipeline().addLast("msgpack encoder", new MsgpackEncoder()); // 添加业务处理handler ch.pipeline().addLast(new EchoServerHandler()); } }); // 绑定端口，同步等待成功 ChannelFuture f = b.bind(port).sync(); // 等待服务端监听端口关闭 f.channel().closeFuture().sync(); } finally { // 优雅退出，释放线程池资源 bossGroup.shutdownGracefully(); workerGroup.shutdownGracefully(); } } public static void main(String[] args) throws Exception { int port = 8080; if(args != null && args.length > 0) { try { port = Integer.valueOf(port); } catch (NumberFormatException e) { // TODO: handle exception } } new EchoServer().bind(port); }}EchoServerHandler.java
package cn.xpleaf.echo;import io.netty.channel.ChannelHandlerContext;import io.netty.channel.ChannelInboundHandlerAdapter;public class EchoServerHandler extends ChannelInboundHandlerAdapter { @Override public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { System.out.println("Server receive the msgpack message : " + msg); ctx.write(msg); } @Override public void channelReadComplete(ChannelHandlerContext ctx) throws Exception { ctx.flush(); } @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { // 发生异常，关闭链路 ctx.close(); }}客户端
EchoClient.java
package cn.xpleaf.echo;import cn.demo.simple.MsgPackDecode;import cn.xpleaf.msgpack.MsgpackDecoder;import cn.xpleaf.msgpack.MsgpackEncoder;import io.netty.bootstrap.Bootstrap;import io.netty.channel.ChannelFuture;import io.netty.channel.ChannelInitializer;import io.netty.channel.ChannelOption;import io.netty.channel.EventLoopGroup;import io.netty.channel.nio.NioEventLoopGroup;import io.netty.channel.socket.SocketChannel;import io.netty.channel.socket.nio.NioSocketChannel;public class EchoClient { public void connect(String host, int port, int sendNumber) throws Exception { // 配置客户端NIO线程组 EventLoopGroup group = new NioEventLoopGroup(); try { Bootstrap b = new Bootstrap(); b.group(group).channel(NioSocketChannel.class) .option(ChannelOption.TCP_NODELAY, true) // 设置TCP连接超时时间 .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 3000) .handler(new ChannelInitializer<SocketChannel>() { @Override protected void initChannel(SocketChannel ch) throws Exception { // 添加MesspagePack解码器 ch.pipeline().addLast("msgpack decoder", new MsgPackDecode()); // 添加MessagePack编码器 ch.pipeline().addLast("msgpack encoder", new MsgpackEncoder()); // 添加业务处理handler ch.pipeline().addLast(new EchoClientHandler(sendNumber)); } }); // 发起异步连接操作 ChannelFuture f = b.connect(host, port).sync(); // 等待客户端链路关闭 f.channel().closeFuture().sync(); } finally { // 优雅退出，释放NIO线程组 group.shutdownGracefully(); } } public static void main(String[] args) throws Exception { int port = 8080; if(args != null && args.length > 0) { try { port = Integer.valueOf(port); } catch (NumberFormatException e) { // 采用默认值 } } int sendNumber = 1000; new EchoClient().connect("localhost", port, sendNumber); }}EchoClientHander.java
package cn.xpleaf.echo;import cn.xpleaf.pojo.User;import io.netty.buffer.ByteBuf;import io.netty.buffer.Unpooled;import io.netty.channel.ChannelHandlerAdapter;import io.netty.channel.ChannelHandlerContext;import io.netty.channel.ChannelInboundHandlerAdapter;public class EchoClientHandler extends ChannelInboundHandlerAdapter { // sendNumber为写入发送缓冲区的对象数量 private int sendNumber; public EchoClientHandler(int sendNumber) { this.sendNumber = sendNumber; } /** * 构建长度为userNum的User对象数组 * @param userNum * @return */ private User[] getUserArray(int userNum) { User[] users = new User[userNum]; User user = null; for(int i = 0; i < userNum; i++) { user = new User(); user.setName("ABCDEFG --->" + i); user.setAge(i); users[i] = user; } return users; } @Override public void channelActive(ChannelHandlerContext ctx) { User[] users = getUserArray(sendNumber); for (User user : users) { ctx.writeAndFlush(user); } } @Override public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { System.out.println("Client receive the msgpack message : " + msg); } @Override public void channelReadComplete(ChannelHandlerContext ctx) throws Exception { ctx.flush(); } @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { ctx.close(); }}POJO
User.java
package cn.xpleaf.pojo;import org.msgpack.annotation.Message;@Messagepublic class User { private String name; private int age; public String getName() { return name; } public void setName(String name) { this.name = name; } public int getAge() { return age; } public void setAge(int age) { this.age = age; } @Override public String toString() { return "User [name=" + name + ", age=" + age + "]"; }}测试
当EchoClient.java中的sendNumber为1时，服务端和客户端都是正常工作的，此时，服务端和客户端的输出分别如下：
服务端：
Server receive the msgpack message : ["ABCDEFG --->0",0]
客户端：
Client receive the msgpack message : ["ABCDEFG --->0",0]
但是当sendNumber数字很大时，就不能正常工作了，比如可以设置为1000，此时输出结果如下：
服务端：
Server receive the msgpack message : ["ABCDEFG --->0",0]Server receive the msgpack message : ["ABCDEFG --->1",1]Server receive the msgpack message : ["ABCDEFG --->3",3]...省略输出...Server receive the msgpack message : ["ABCDEFG --->146",146]Server receive the msgpack message : 70Server receive the msgpack message : ["ABCDEFG --->156",156]Server receive the msgpack message : ["ABCDEFG --->157",157]...省略输出...
客户端：
Client receive the msgpack message : ["ABCDEFG --->0",0]Client receive the msgpack message : 62Client receive the msgpack message : 68
显然运行结果跟预期的不太一样，这是因为出现了TCP粘包问题。
粘包问题解决方案
在前面代码的基础上，只需要对EchoServer.java和EchoClient.java中的代码进行修改即可。
EchoServer.java
package cn.xpleaf.echo02;import cn.demo.simple.MsgPackDecode;import cn.xpleaf.msgpack.MsgpackDecoder;import cn.xpleaf.msgpack.MsgpackEncoder;import io.netty.bootstrap.ServerBootstrap;import io.netty.channel.ChannelFuture;import io.netty.channel.ChannelInitializer;import io.netty.channel.ChannelOption;import io.netty.channel.EventLoopGroup;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.LengthFieldPrepender;public class EchoServer { public void bind(int port) throws Exception { // 配置服务端的NIO线程组 EventLoopGroup bossGroup = new NioEventLoopGroup(); EventLoopGroup workerGroup = new NioEventLoopGroup(); try { ServerBootstrap b = new ServerBootstrap(); b.group(bossGroup, workerGroup) .channel(NioServerSocketChannel.class) .option(ChannelOption.SO_BACKLOG, 1024) .childHandler(new ChannelInitializer<SocketChannel>() { @Override protected void initChannel(SocketChannel ch) throws Exception { // 添加长度字段解码器 // 在MessagePack解码器之前增加LengthFieldBasedFrameDecoder，用于处理半包消息 // 它会解析消息头部的长度字段信息，这样后面的MsgpackDecoder接收到的永远是整包消息 ch.pipeline().addLast("frameDecoder", new LengthFieldBasedFrameDecoder(65535, 0, 2, 0, 2)); // 添加MesspagePack解码器 ch.pipeline().addLast("msgpack decoder", new MsgPackDecode()); // 添加长度字段编码器 // 在MessagePack编码器之前增加LengthFieldPrepender，它将在ByteBuf之前增加2个字节的消息长度字段 ch.pipeline().addLast("frameEncoder", new LengthFieldPrepender(2)); // 添加MessagePack编码器 ch.pipeline().addLast("msgpack encoder", new MsgpackEncoder()); // 添加业务处理handler ch.pipeline().addLast(new EchoServerHandler()); } }); // 绑定端口，同步等待成功 ChannelFuture f = b.bind(port).sync(); // 等待服务端监听端口关闭 f.channel().closeFuture().sync(); } finally { // 优雅退出，释放线程池资源 bossGroup.shutdownGracefully(); workerGroup.shutdownGracefully(); } } public static void main(String[] args) throws Exception { int port = 8080; if(args != null && args.length > 0) { try { port = Integer.valueOf(port); } catch (NumberFormatException e) { // TODO: handle exception } } new EchoServer().bind(port); }}EchoClient.java
package cn.xpleaf.echo02;import cn.demo.simple.MsgPackDecode;import cn.xpleaf.msgpack.MsgpackDecoder;import cn.xpleaf.msgpack.MsgpackEncoder;import io.netty.bootstrap.Bootstrap;import io.netty.channel.ChannelFuture;import io.netty.channel.ChannelInitializer;import io.netty.channel.ChannelOption;import io.netty.channel.EventLoopGroup;import io.netty.channel.nio.NioEventLoopGroup;import io.netty.channel.socket.SocketChannel;import io.netty.channel.socket.nio.NioSocketChannel;import io.netty.handler.codec.LengthFieldBasedFrameDecoder;import io.netty.handler.codec.LengthFieldPrepender;public class EchoClient { public void connect(String host, int port, int sendNumber) throws Exception { // 配置客户端NIO线程组 EventLoopGroup group = new NioEventLoopGroup(); try { Bootstrap b = new Bootstrap(); b.group(group).channel(NioSocketChannel.class) .option(ChannelOption.TCP_NODELAY, true) // 设置TCP连接超时时间 .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 3000) .handler(new ChannelInitializer<SocketChannel>() { @Override protected void initChannel(SocketChannel ch) throws Exception { // 添加长度字段解码器 // 在MessagePack解码器之前增加LengthFieldBasedFrameDecoder，用于处理半包消息 // 它会解析消息头部的长度字段信息，这样后面的MsgpackDecoder接收到的永远是整包消息 ch.pipeline().addLast("frameDecoder", new LengthFieldBasedFrameDecoder(65535, 0, 2, 0, 2)); // 添加MesspagePack解码器 ch.pipeline().addLast("msgpack decoder", new MsgPackDecode()); // 添加长度字段编码器 // 在MessagePack编码器之前增加LengthFieldPrepender，它将在ByteBuf之前增加2个字节的消息长度字段 ch.pipeline().addLast("frameEncoder", new LengthFieldPrepender(2)); // 添加MessagePack编码器 ch.pipeline().addLast("msgpack encoder", new MsgpackEncoder()); // 添加业务处理handler ch.pipeline().addLast(new EchoClientHandler(sendNumber)); } }); // 发起异步连接操作 ChannelFuture f = b.connect(host, port).sync(); // 等待客户端链路关闭 f.channel().closeFuture().sync(); } finally { // 优雅退出，释放NIO线程组 group.shutdownGracefully(); } } public static void main(String[] args) throws Exception { int port = 8080; if(args != null && args.length > 0) { try { port = Integer.valueOf(port); } catch (NumberFormatException e) { // 采用默认值 } } int sendNumber = 1000; new EchoClient().connect("localhost", port, sendNumber); }}测试
可以将EchoClient.java中sendNumber设置为1000或更大，此时服务端和客户端的输出结果跟预期的都是一样的。
测试结果为，服务端和客户端都会打印1000行的信息（假设sendNumber为1000），这里不再给出运行结果。