目录

• TCP粘包/拆包
  • TCP粘包/拆包问题说明
  • TCP粘包/拆包发生的原因
  • 粘包问题的解决策略
• 未考虑TCP粘包导致功能异常案例 
  • TimeServer的改造
  • TimeClient的改造
• 利用LineBasedFrameDecoder解决TCP粘包问题
  • LineBasedFrameDecoder和StringDecoder的原理分析

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerAdapter;
import io.netty.channel.ChannelHandlerContext;public class TimeServerHandler extends ChannelHandlerAdapter {private int counter;@Overridepublic 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());System.out.println("The time server receive order : " + body + " ; the counter is : " + ++counter);String currentTime = "QUERY TIME ORDER".equalsIgnoreCase(body) ?new java.util.Date( System.currentTimeMillis()).toString() : "BAD ORDER";currentTime = currentTime + System.getProperty("line.separator");ByteBuf resp = Unpooled.copiedBuffer(currentTime.getBytes());ctx.writeAndFlush(resp);}@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {ctx.close();}
}

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerAdapter;
import io.netty.channel.ChannelHandlerContext;public class TimeClientHandler extends ChannelHandlerAdapter {private int counter;private byte[] req;
public TimeClientHandler() {req = ("QUERY TIME ORDER" + System.getProperty("line.separator")).getBytes();}@Overridepublic void channelActive(ChannelHandlerContext ctx) {ByteBuf message = null;for (int i = 0; i < 100; i++) {message = Unpooled.buffer(req.length);message.writeBytes(req);ctx.writeAndFlush(message);}}@Overridepublic 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");System.out.println("Now is : " + body + " ; the counter is : " + ++counter);}@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {// 释放资源
        ctx.close();}
}

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.LineBasedFrameDecoder;
import io.netty.handler.codec.string.StringDecoder;public class TimeServer {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 ChildChannelHandler());// 绑定端口，同步等待成功ChannelFuture f = b.bind(port).sync();// 等待服务端监听端口关闭
            f.channel().closeFuture().sync();} finally {// 优雅退出，释放线程池资源
            bossGroup.shutdownGracefully();workerGroup.shutdownGracefully();}}private class ChildChannelHandler extends ChannelInitializer {@Overrideprotected void initChannel(Channel arg0) throws Exception {arg0.pipeline().addLast(new LineBasedFrameDecoder(1024));arg0.pipeline().addLast(new StringDecoder());arg0.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(args[0]);} catch (NumberFormatException e) {// 采用默认值
            }}new TimeServer().bind(port);}
}import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerAdapter;
import io.netty.channel.ChannelHandlerContext;public class TimeServerHandler extends ChannelHandlerAdapter {private int counter;@Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg)throws Exception {String body = (String) msg;System.out.println("The time server receive order : " + body  + " ; the counter is : " + ++counter);String currentTime = "QUERY TIME ORDER".equalsIgnoreCase(body) ?new java.util.Date(System.currentTimeMillis()).toString() : "BAD ORDER";currentTime = currentTime + System.getProperty("line.separator");ByteBuf resp = Unpooled.copiedBuffer(currentTime.getBytes());ctx.writeAndFlush(resp);}@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {ctx.close();}
}

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.LineBasedFrameDecoder;
import io.netty.handler.codec.string.StringDecoder;public class TimeClient {public void connect(int port, String host) throws Exception {
// 配置客户端NIO线程组EventLoopGroup group = new NioEventLoopGroup();try {Bootstrap b = new Bootstrap();b.group(group).channel(NioSocketChannel.class).option(ChannelOption.TCP_NODELAY, true).handler(new ChannelInitializer() {@Overridepublic void initChannel(Channel ch)throws Exception {ch.pipeline().addLast(new LineBasedFrameDecoder(1024));ch.pipeline().addLast(new StringDecoder());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(args[0]);} catch (NumberFormatException e) {// 采用默认值
            }}new TimeClient().connect(port, "127.0.0.1");}
}import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerAdapter;
import io.netty.channel.ChannelHandlerContext;public class TimeClientHandler extends ChannelHandlerAdapter {private int counter;private byte[] req;public TimeClientHandler() {req = ("QUERY TIME ORDER" + System.getProperty("line.separator")).getBytes();}@Overridepublic void channelActive(ChannelHandlerContext ctx) {ByteBuf message = null;for (int i = 0; i < 100; i++) {message = Unpooled.buffer(req.length);message.writeBytes(req);ctx.writeAndFlush(message);}}@Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg)throws Exception {String body = (String) msg;System.out.println("Now is : " + body + " ; the counter is : "  + ++counter);}@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {// 释放资源
        ctx.close();}
}