netty 解决TCP粘包与拆包问题(二)

 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 }