Netty中TCP粘包问题代码示例
Netty中会发生TCP粘包和拆包的问题，当然，其实对于曾经的网络工程师来说，第一次看到这名词可能会有点不适应，因为在那会我们是说TCP的累计发送和分片功能，不过道理都是一样的。
代码来自于《Netty权威指南》第4章，不过我还是对其中的部分代码做了修改，原因还是Netty版本的问题，书本用的是Netty 5.x的版本，官方已经废弃了这个版本，我用的是官方4.x的版本。
需要说明的是，代码还是时间服务器的代码，只是我把很多基础性的注释都去掉了，因为到了这里的时候，其实对Netty的基本使用应该是要熟悉的。
另外我在TimeClientHandler.java代码中也加了对于TCP粘包问题的基本分析，当然也可以参考书本的说明。
服务端
TimeServer.java
package cn.xpleaf.netty;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 TimeServer { ???public void bind(int port) throws Exception { ???????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 ChildChannelHandler()); ???????????ChannelFuture f = b.bind(port).sync(); ???????????f.channel().closeFuture().sync(); ???????} finally { ???????????bossGroup.shutdownGracefully(); ???????????workerGroup.shutdownGracefully(); ???????} ???} ???private class ChildChannelHandler extends ChannelInitializer<SocketChannel> { ???????@Override ???????protected void initChannel(SocketChannel ch) throws Exception { ???????????ch.pipeline().addLast(new TimeServerHandler()); ???????} ???} ???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 TimeServer().bind(port); ???}}TimeServerHandler.java
package cn.xpleaf.netty;import java.sql.Date;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 TimeServerHandler extends ChannelInboundHandlerAdapter { ???private int counter = 0; ???@Override ???public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { ???????ByteBuf buf = (ByteBuf) msg; ???????byte[] req = new byte[buf.readableBytes()]; ???????buf.readBytes(req); ???????// 去掉换行符 ???????String body = new String(req, "utf-8").substring(0, ????????????????req.length - System.getProperty("line.separator").length()); ???????// counter的作用是标记这是第几次收到客户端的请求 ???????System.out.println("The time server receive order : " + body + " ; the counter is : " + ++counter); ???????String currentTime = "QUERY TIME ORDER".equalsIgnoreCase(body) ? ????????????????new Date(System.currentTimeMillis()).toString() : "BAD ORDER"; ???????currentTime = currentTime + System.getProperty("line.separator"); ???????ByteBuf resp = Unpooled.copiedBuffer(currentTime.getBytes()); ???????ctx.write(resp); ???} ???@Override ???public void channelReadComplete(ChannelHandlerContext ctx) throws Exception { ???????ctx.flush(); ???} ???@Override ???public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { ???????ctx.close(); ???}}客户端
TimeClient.java
package cn.xpleaf.netty;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 TimeClient { ???public void connect(int port, String host) throws Exception { ???????EventLoopGroup group = new NioEventLoopGroup(); ???????try { ???????????Bootstrap b = new Bootstrap(); ???????????b.group(group).channel(NioSocketChannel.class) ???????????????.option(ChannelOption.TCP_NODELAY, true) ???????????????.handler(new ChannelInitializer<SocketChannel>() { ???????????????????@Override ???????????????????protected void initChannel(SocketChannel ch) throws Exception { ???????????????????????ch.pipeline().addLast(new TimeClientHandler()); ???????????????????} ???????????????}); ???????????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) { ???????????????// 采用默认值 ???????????} ???????} ???????new TimeClient().connect(port, "localhost"); ???}}TimeClientHandler.java
package cn.xpleaf.netty;import java.util.logging.Logger;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 TimeClientHandler extends ChannelInboundHandlerAdapter { ???private static final Logger logger = Logger.getLogger(TimeServerHandler.class.getName()); ???private int counter; ???private byte[] req; ???public TimeClientHandler() { ???????req = ("QUERY TIME ORDER" + System.getProperty("line.separator")).getBytes(); ???} ???@Override ???public void channelActive(ChannelHandlerContext ctx) { ???????ByteBuf message = null; ???????/** ????????* 原意是向服务端发送100次消息，按照设计应该会收到服务器 ????????* 100次回应，但由于TCP的粘包问题，不会每一个消息都单独发送出去， ????????* TCP本身的机制会使得消息达到一定长度时才发送出去，所以实际测试的情况是： ????????* TCP只发送了两次数据，并将100条消息分成两部分发送出去，而不是每条消息 ????????* 单独发送，发生了所谓的粘包问题，而服务端在进行回应时，显然也很可能不会是 ????????* 回两次包，因为原因是一样的，服务端的TCP也会将netty两次发送的数据合并成一个 ????????* 包进行发送，当然，这其中涉及的TCP的相关知识，可以参考TCP/IP 卷1 ????????*/ ???????for(int i = 0; i < 100; i++) { ???????????message = Unpooled.buffer(req.length); ???????????message.writeBytes(req); ???????????ctx.writeAndFlush(message); ???????} ???} ???@Override ???public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { ???????ByteBuf buf = (ByteBuf)msg; ???????byte[] req = new byte[buf.readableBytes()]; ???????buf.readBytes(req); ???????String body = new String(req, "utf-8"); ???????// counter的作用是标记这是第几次收到客户端的请求 ???????System.out.println("Now is : " + body + " ; the counter is : " + ++counter); ???} ???@Override ???public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { ???????logger.warning("Unexpected exception from downstream : "); ???????ctx.close(); ???}}测试
服务端运行结果
The time server receive order : QUERY TIME ORDERQUERY TIME ORDERQUERY TIME ORDERQUERY TIME ORDERQUERY TIME ORDERQUERY TIME ORDERQUERY TIME ORDERQUERY TIME ORDERQUERY TIME ORDERQUERY TIME ORDERQUERY TIME ORDERQUERY TIME ORDERQUERY TIME ORDERQUERY TIME ORDERQUERY TIME ORDERQUERY TIME ORDERQUERY TIME ORDERQUERY TIME ORDERQUERY TIME ORDERQUERY TIME ORDERQUERY TIME ORDERQUERY TIME ORDERQUERY TIME ORDERQUERY TIME ORDERQUERY TIME ORDERQUERY TIME ORDERQUERY TIME ORDERQUERY TIME ORDERQUERY TIME ORDERQUERY TIME ORDERQUERY TIME ORDERQUERY TIME ORDERQUERY TIME ORDERQUERY TIME ORDERQUERY TIME ORDERQUERY TIME ORDERQUERY TIME ORDERQUERY TIME ORDERQUERY TIME ORDERQUERY TIME ORDERQUERY TIME ORDERQUERY TIME ORDERQUERY TIME ORDERQUERY TIME ORDERQUERY TIME ORDERQUERY TIME ORDERQUERY TIME ORDERQUERY TIME ORDERQUERY TIME ORDERQUERY TIME ORDERQUERY TIME ORDERQUERY TIME ORDERQUERY TIME ORDERQUERY TIME ORDERQUERY TIME ORDERQUERY TIME ORDERQUERY TIME ORDERQUERY TIME ORDERQUERY TIME ORDERQUERY TIME ORDERQUE ; the counter is : 1The time server receive order : Y TIME ORDERQUERY TIME ORDERQUERY TIME ORDERQUERY TIME ORDERQUERY TIME ORDERQUERY TIME ORDERQUERY TIME ORDERQUERY TIME ORDERQUERY TIME ORDERQUERY TIME ORDERQUERY TIME ORDERQUERY TIME ORDERQUERY TIME ORDERQUERY TIME ORDERQUERY TIME ORDERQUERY TIME ORDERQUERY TIME ORDERQUERY TIME ORDERQUERY TIME ORDERQUERY TIME ORDERQUERY TIME ORDERQUERY TIME ORDERQUERY TIME ORDERQUERY TIME ORDERQUERY TIME ORDERQUERY TIME ORDERQUERY TIME ORDERQUERY TIME ORDERQUERY TIME ORDERQUERY TIME ORDERQUERY TIME ORDERQUERY TIME ORDERQUERY TIME ORDERQUERY TIME ORDERQUERY TIME ORDERQUERY TIME ORDERQUERY TIME ORDERQUERY TIME ORDERQUERY TIME ORDERQUERY TIME ORDER ; the counter is : 2客户端运行结果
Now is : BAD ORDERBAD ORDER ; the counter is : 1从上面的运行结果，TCP粘包的问题就显而易见了。
Netty中TCP粘包问题代码示例与分析
原文地址：http://blog.51cto.com/xpleaf/2071054