1、短连接（粘包问题）

将 TCP 连接改成短连接，一个请求一个短连接。这样的话，建立连接到释放连接之间的消息即为传输的信息，消息也就产生了边界。

优点：方法十分简单，不需要在我们的应用中做过多修改
缺点：效率低下，TCP 连接和断开都会涉及三次握手以及四次握手，每个消息都会涉及这些过程，十分浪费性能

  1. public class HelloWorldClient {
  2.     static final Logger log = LoggerFactory.getLogger(HelloWorldClient.class);
  4.     public static void main(String[] args) {
  5.         // 分 10 次发送
  6.         for (int i = 0; i < 10; i++) {
  7.             send();
  8.         }
  9.     }
  11.     private static void send() {
  12.         NioEventLoopGroup worker = new NioEventLoopGroup();
  13.         try {
  14.             Bootstrap bootstrap = new Bootstrap();
  15.             bootstrap.channel(NioSocketChannel.class);
  16.             bootstrap.group(worker);
  17.             bootstrap.handler(new ChannelInitializer<SocketChannel>() {
  18.                 @Override
  19.                 protected void initChannel(SocketChannel ch) throws Exception {
  20.                     log.debug("conneted...");
  21.                     ch.pipeline().addLast(new LoggingHandler(LogLevel.DEBUG));
  22.                     ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
  23.                         @Override
  24.                         public void channelActive(ChannelHandlerContext ctx) throws Exception {
  25.                             log.debug("sending...");
  26.                             ByteBuf buffer = ctx.alloc().buffer();
  27.                             buffer.writeBytes(new byte[]{0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15});
  28.                             ctx.writeAndFlush(buffer);
  29.                             // 发完即关
  30.                             ctx.close();
  31.                         }
  32.                     });
  33.                 }
  34.             });
  35.             ChannelFuture channelFuture = bootstrap.connect("localhost", 8080).sync();
  36.             channelFuture.channel().closeFuture().sync();
  38.         } catch (InterruptedException e) {
  39.             log.error("client error", e);
  40.         } finally {
  41.             worker.shutdownGracefully();
  42.         }
  43.     }
  44. }

2、固定长度

让所有数据包长度固定

客户端消息不足16个字节时，进行填充

  1. public class StickAndHalfPacketClient {
  3.     static final Logger log = LoggerFactory.getLogger(StickAndHalfPacketClient.class);
  4.     public static void main(String[] args) {
  5.         NioEventLoopGroup worker = new NioEventLoopGroup();
  6.         try {
  7.             Bootstrap bootstrap = new Bootstrap();
  8.             bootstrap.channel(NioSocketChannel.class);
  9.             bootstrap.group(worker);
  10.             bootstrap.handler(new ChannelInitializer<SocketChannel>() {
  11.                 @Override
  12.                 protected void initChannel(SocketChannel ch) throws Exception {
  13.                     log.debug("connetted...");
  14.                     ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
  15.                         @Override
  16.                         public void channelActive(ChannelHandlerContext ctx) throws Exception {
  17.                             log.debug("sending...");
  18.                             ByteBuf buffer = ctx.alloc().buffer();
  19.                             for (int i = 0; i < 10; i++) {
  20.                                 buffer.writeBytes(fillByteTo16(i));
  21.                             }
  22.                             ctx.writeAndFlush(buffer);
  23.                         }
  24.                     });
  25.                 }
  26.             });
  27.             ChannelFuture channelFuture = bootstrap.connect("127.0.0.1", 8080).sync();
  28.             channelFuture.channel().closeFuture().sync();
  30.         } catch (InterruptedException e) {
  31.             log.error("client error", e);
  32.         } finally {
  33.             worker.shutdownGracefully();
  34.         }
  35.     }
  37.     private static byte[] fillByteTo16(int a){
  38.         byte[] data = new byte[16];
  39.         data[0] = (byte)a;
  40.         IntStream.range(1,16).forEach(i -> data[i] = '#');
  41.         return data;
  42.     }
  43. }

服务端使用定长解码器

  1. public class StickAndHalfPacketServer {
  2.     static final Logger log = LoggerFactory.getLogger(StickAndHalfPacketServer.class);
  3.     void start() {
  4.         NioEventLoopGroup boss = new NioEventLoopGroup(1);
  5.         NioEventLoopGroup worker = new NioEventLoopGroup();
  6.         try {
  7.             ServerBootstrap serverBootstrap = new ServerBootstrap();
  8.             serverBootstrap.channel(NioServerSocketChannel.class);
  9.             serverBootstrap.group(boss, worker);
  10.             serverBootstrap.option(ChannelOption.SO_RCVBUF, 10);
  11.             serverBootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
  12.                 @Override
  13.                 protected void initChannel(SocketChannel ch) throws Exception {
  14.                     ch.pipeline().addLast(new FixedLengthFrameDecoder(16));
  15.                     ch.pipeline().addLast(new LoggingHandler(LogLevel.DEBUG));
  16.                     ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
  17.                         @Override
  18.                         public void channelActive(ChannelHandlerContext ctx) throws Exception {
  19.                             log.debug("connected {}", ctx.channel());
  20.                             super.channelActive(ctx);
  21.                         }
  23.                         @Override
  24.                         public void channelInactive(ChannelHandlerContext ctx) throws Exception {
  25.                             log.debug("disconnect {}", ctx.channel());
  26.                             super.channelInactive(ctx);
  27.                         }
  28.                     });
  29.                 }
  30.             });
  31.             ChannelFuture channelFuture = serverBootstrap.bind(8080);
  32.             log.debug("{} binding...", channelFuture.channel());
  33.             channelFuture.sync();
  34.             log.debug("{} bound...", channelFuture.channel());
  35.             channelFuture.channel().closeFuture().sync();
  36.         } catch (InterruptedException e) {
  37.             log.error("server error", e);
  38.         } finally {
  39.             boss.shutdownGracefully();
  40.             worker.shutdownGracefully();
  41.             log.debug("stoped");
  42.         }
  43.     }
  45.     public static void main(String[] args) {
  46.         new StickAndHalfPacketServer().start();
  47.     }
  48. }

3、固定分隔符

  1. public class StickAndHalfPacketClient {
  3.     static final Logger log = LoggerFactory.getLogger(StickAndHalfPacketClient.class);
  4.     public static void main(String[] args) {
  5.         NioEventLoopGroup worker = new NioEventLoopGroup();
  6.         try {
  7.             Bootstrap bootstrap = new Bootstrap();
  8.             bootstrap.channel(NioSocketChannel.class);
  9.             bootstrap.group(worker);
  10.             bootstrap.handler(new ChannelInitializer<SocketChannel>() {
  11.                 @Override
  12.                 protected void initChannel(SocketChannel ch) throws Exception {
  13.                     log.debug("connetted...");
  14.                     ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
  15.                         @Override
  16.                         public void channelActive(ChannelHandlerContext ctx) throws Exception {
  17.                             log.debug("sending...");
  18.                             ByteBuf buffer = ctx.alloc().buffer();
  19.                             buffer.writeBytes("hello netty\nhello coder\nhahahah\n ddadaodahfdohqwopygfqficgdkuyqrofghoqvofgfq\n".getBytes(StandardCharsets.UTF_8));
  20.                             ctx.writeAndFlush(buffer);
  21.                         }
  22.                     });
  23.                 }
  24.             });
  25.             ChannelFuture channelFuture = bootstrap.connect("127.0.0.1", 8080).sync();
  26.             channelFuture.channel().closeFuture().sync();
  28.         } catch (InterruptedException e) {
  29.             log.error("client error", e);
  30.         } finally {
  31.             worker.shutdownGracefully();
  32.         }
  33.     }
  36. }
  1. public class StickAndHalfPacketServer {
  2.     static final Logger log = LoggerFactory.getLogger(StickAndHalfPacketServer.class);
  3.     void start() {
  4.         NioEventLoopGroup boss = new NioEventLoopGroup(1);
  5.         NioEventLoopGroup worker = new NioEventLoopGroup();
  6.         try {
  7.             ServerBootstrap serverBootstrap = new ServerBootstrap();
  8.             serverBootstrap.channel(NioServerSocketChannel.class);
  9.             serverBootstrap.group(boss, worker);
  10.             serverBootstrap.option(ChannelOption.SO_RCVBUF, 10);
  11.             serverBootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
  12.                 @Override
  13.                 protected void initChannel(SocketChannel ch) throws Exception {
  14.                     ch.pipeline().addLast(new LineBasedFrameDecoder(1024));
  15.                     ch.pipeline().addLast(new LoggingHandler(LogLevel.DEBUG));
  16.                     ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
  17.                         @Override
  18.                         public void channelActive(ChannelHandlerContext ctx) throws Exception {
  19.                             log.debug("connected {}", ctx.channel());
  20.                             super.channelActive(ctx);
  21.                         }
  23.                         @Override
  24.                         public void channelInactive(ChannelHandlerContext ctx) throws Exception {
  25.                             log.debug("disconnect {}", ctx.channel());
  26.                             super.channelInactive(ctx);
  27.                         }
  28.                     });
  29.                 }
  30.             });
  31.             ChannelFuture channelFuture = serverBootstrap.bind(8080);
  32.             log.debug("{} binding...", channelFuture.channel());
  33.             channelFuture.sync();
  34.             log.debug("{} bound...", channelFuture.channel());
  35.             channelFuture.channel().closeFuture().sync();
  36.         } catch (InterruptedException e) {
  37.             log.error("server error", e);
  38.         } finally {
  39.             boss.shutdownGracefully();
  40.             worker.shutdownGracefully();
  41.             log.debug("stoped");
  42.         }
  43.     }
  45.     public static void main(String[] args) {
  46.         new StickAndHalfPacketServer().start();
  47.     }
  48. }

4、预设长度

在发送消息前，先约定用定长字节表示接下来数据的长度

  1.     private final int maxFrameLength;  //发送数据包的最大长度
  2.     private final int lengthFieldOffset; //长度字段偏移量
  3.     private final int lengthFieldLength; //长度字段本身占用的字节数
  4.     private final int lengthFieldEndOffset; 
  5.     private final int lengthAdjustment; //长度字段的偏移量矫正
  6.     private final int initialBytesToStrip; //丢弃的起始字节数
  1. public class HeadContent {
  3.     public static void main(String[] args) {
  4.         EmbeddedChannel channel = new EmbeddedChannel(new LengthFieldBasedFrameDecoder(1024,0,4,3,4),
  5.                 new LoggingHandler(LogLevel.DEBUG));
  6.         final ByteBuf buffer = ByteBufAllocator.DEFAULT.buffer();
  7.         addMsg("hello,world",buffer);
  8.         addMsg("hello,java",buffer);
  9.         addMsg("hello,netty",buffer);
  10.         channel.writeInbound(buffer);
  12.     }
  14.     private static void addMsg(String msg,ByteBuf buffer){
  15.         final byte[] bytes = msg.getBytes(StandardCharsets.UTF_8);
  16.         final int length = bytes.length;
  17.         buffer.writeInt(length);
  18.         buffer.writeBytes(bytes);
  19.         buffer.writeBytes("-v1".getBytes(StandardCharsets.UTF_8));
  20.     }
  21. }