netty 解决TCP粘包与拆包问题

Posted 2020-07-18 全力以赴001

TCP以流的方式进行数据传输，上层应用协议为了对消息的区分，采用了以下几种方法。

1.消息固定长度

2.第一篇讲的回车换行符形式

3.以特殊字符作为消息结束符的形式

4.通过消息头中定义长度字段来标识消息的总长度

 

一、采用指定分割符解决粘包与拆包问题

服务端

 1 package com.ming.netty.nio.stickpack;
 2 
 3 
 4 
 5 import java.net.InetSocketAddress;
 6 
 7 import io.netty.bootstrap.ServerBootstrap;
 8 import io.netty.buffer.ByteBuf;
 9 import io.netty.buffer.Unpooled;
10 import io.netty.channel.ChannelFuture;
11 import io.netty.channel.ChannelInitializer;
12 import io.netty.channel.ChannelOption;
13 import io.netty.channel.EventLoopGroup;
14 import io.netty.channel.nio.NioEventLoopGroup;
15 import io.netty.channel.socket.SocketChannel;
16 import io.netty.channel.socket.nio.NioserverSocketChannel;
17 import io.netty.handler.codec.DelimiterBasedFrameDecoder;
18 import io.netty.handler.codec.string.StringDecoder;
19 import io.netty.handler.logging.LogLevel;
20 import io.netty.handler.logging.LoggingHandler;
21 
22 public class EchoServer {
23 
24     public void bind(String addr,int port) throws Exception{
25         EventLoopGroup bossGroup=new NioEventLoopGroup();
26         EventLoopGroup workGroup=new NioEventLoopGroup();
27         try {
28             ServerBootstrap server=new ServerBootstrap();
29             server.group(bossGroup,workGroup)
30                   .channel(NioServerSocketChannel.class)
31                   .option(ChannelOption.SO_BACKLOG, 100)
32                   .handler(new LoggingHandler(LogLevel.INFO))
33                   .childHandler(new ChannelInitializer<SocketChannel>() {
34 
35                     @Override
36                     protected void initChannel(SocketChannel sc) throws Exception {
37                         ByteBuf delimiter=Unpooled.copiedBuffer("$_".getBytes());//指定消息分割符处理数据
38                         sc.pipeline().addLast(new DelimiterBasedFrameDecoder(1024, delimiter));//如果取消了分割符解码，就会出现TCP粘包之类的问题了
39                         sc.pipeline().addLast(new StringDecoder());
40                         sc.pipeline().addLast(new EchoServerHandler());
41                         
42                     }
43                       
44                 });
45             ChannelFuture f=server.bind(new InetSocketAddress(addr, port)).sync();
46             System.out.println("启动服务器:"+f.channel().localAddress());
47             //等等服务器端监听端口关闭
48             f.channel().closeFuture().sync();
49         } catch (Exception e) {
50             e.printStackTrace();
51         }finally{
52             bossGroup.shutdownGracefully();
53             workGroup.shutdownGracefully();
54         }
55     }
56     
57     
58     public static void main(String[] args) throws Exception{
59         new EchoServer().bind("192.168.1.108", 8500);
60     }
61     
62 }

 1 package com.ming.netty.nio.stickpack;
 2 
 3 import io.netty.buffer.ByteBuf;
 4 import io.netty.buffer.Unpooled;
 5 import io.netty.channel.ChannelHandlerAdapter;
 6 import io.netty.channel.ChannelHandlerContext;
 7 
 8 public class EchoServerHandler extends ChannelHandlerAdapter{
 9 
10     int count=0;
11     
12     @Override
13     public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
14         
15         String body=(String)msg;
16         System.out.println("服务器收到"+(++count)+"次客户端消息，消息是:"+body);
17         body+="$_";
18         ByteBuf rep=Unpooled.copiedBuffer(body.getBytes());
19         ctx.writeAndFlush(rep);
20     }
21 
22     @Override
23     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
24         cause.printStackTrace();
25         ctx.close();
26     }
27 
28     
29 }

 

客服端:

 1 package com.ming.netty.nio.stickpack;
 2 
 3 import java.net.InetSocketAddress;
 4 
 5 import io.netty.bootstrap.Bootstrap;
 6 import io.netty.buffer.ByteBuf;
 7 import io.netty.buffer.Unpooled;
 8 import io.netty.channel.ChannelFuture;
 9 import io.netty.channel.ChannelInitializer;
10 import io.netty.channel.ChannelOption;
11 import io.netty.channel.EventLoopGroup;
12 import io.netty.channel.nio.NioEventLoopGroup;
13 import io.netty.channel.socket.SocketChannel;
14 import io.netty.channel.socket.nio.NioSocketChannel;
15 import io.netty.handler.codec.DelimiterBasedFrameDecoder;
16 import io.netty.handler.codec.string.StringDecoder;
17 
18 public class EchoClient {
19 
20     public void connect(String addr,int port) throws Exception{
21         EventLoopGroup workGroup=new NioEventLoopGroup();
22         try {
23             Bootstrap b=new Bootstrap();
24             b.group(workGroup)
25              .channel(NioSocketChannel.class)
26              .option(ChannelOption.TCP_NODELAY, true)
27              .handler(new ChannelInitializer<SocketChannel>() {
28 
29                 @Override
30                 protected void initChannel(SocketChannel sc) throws Exception {
31                     ByteBuf delimiter=Unpooled.copiedBuffer("$_".getBytes());//指定消息分割符
32                     sc.pipeline().addLast(new DelimiterBasedFrameDecoder(1024, delimiter));
33                     sc.pipeline().addLast(new StringDecoder());
34                     sc.pipeline().addLast(new EchoClientHandler());
35                 }
36                  
37             });
38             
39             ChannelFuture f=b.connect(new InetSocketAddress(addr, port)).sync();
40             System.out.println("连接服务器:"+f.channel().remoteAddress()+",本地地址:"+f.channel().localAddress());
41             f.channel().closeFuture().sync();//等待客户端关闭连接
42             
43         } catch (Exception e) {
44             e.printStackTrace();
45         }finally{
46             workGroup.shutdownGracefully();
47         }
48     }
49     
50     public static void main(String[] args) throws Exception{
51         new EchoClient().connect("192.168.1.108", 8500);
52     }
53 }

 1 package com.ming.netty.nio.stickpack;
 2 
 3 import io.netty.buffer.Unpooled;
 4 import io.netty.channel.ChannelHandlerAdapter;
 5 import io.netty.channel.ChannelHandlerContext;
 6 
 7 public class EchoClientHandler extends ChannelHandlerAdapter{
 8 
 9     int count=0;
10     
11     static final String REQUEST_TEST_DATA="I love you....$_";
12 
13     
14     
15     @Override
16     public void channelActive(ChannelHandlerContext ctx) throws Exception {
17         //发送消息,模拟发送向服务端发送1000条数据
18         for(int i=0,j=1000;i<j;i++){
19             ctx.writeAndFlush(Unpooled.copiedBuffer(REQUEST_TEST_DATA.getBytes()));
20         }
21     }
22 
23     @Override
24     public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
25         String sendMsg=(String)msg;
26         System.out.println("客户端发送给服务器的次数:"+(++count)+"，服务器接收数据为："+sendMsg);
27     }
28 
29     
30     
31     @Override
32     public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
33         ctx.flush();
34     }
35 
36     @Override
37     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
38         cause.printStackTrace();
39         ctx.close();
40     }
41     
42     
43 }

 

很多事情看代码解决，hello world!

 

下篇打算写定长解码了...最后写一下通过消息头中定义长度字段来标识消息的总长度来解码玩玩....

 

感觉可以点个赞吧,好自恋一把