Netty（七）：流数据的传输处理

Socket Buffer的缺陷

对于例如TCP/IP这种基于流的传输协议实现，接收到的数据会被存储在socket的接受缓冲区内。不幸的是，这种基于流的传输缓冲区并不是一个包队列，而是一个字节队列。这意味着，即使你以两个数据包的形式发送了两条消息，操作系统却不会把它们看成是两条消息，而仅仅是一个批次的字节序列。因此，在这种情况下我们就无法保证收到的数据恰好就是远程节点所发送的数据。例如，让我们假设一个操作系统的TCP/IP堆栈收到了三个数据包：

由于这种流传输协议的普遍性质，在你的应用中有较高的可能会把这些数据读取为另外一种形式：

因此对于数据的接收方，不管是服务端还是客户端，应当重构这些接收到的数据，让其变成一种可让你的应用逻辑易于理解的更有意义的数据结构。在上面所述的这个例子中，接收到的数据应当重构为下面的形式：

第一种解决方案（使用特殊字符分割）

Netty提供了一个分隔符类DelimiterBasedFrameDecoder（自定义分隔符）

下面的开发我是居于我的Netty第一个开发程序来讲的，没看过我的这篇文章可以先看看，想信你在Netty第一个开发程序会捕获很多你想不到的知识。

服务端

1public class Server {
2
3    public static void main(String[] args) throws Exception{
4        //1 创建2个线程，一个是负责接收客户端的连接。一个是负责进行数据传输的
5        EventLoopGroup pGroup = new NioEventLoopGroup();
6        EventLoopGroup cGroup = new NioEventLoopGroup();
7
8        //2 创建服务器辅助类
9        ServerBootstrap b = new ServerBootstrap();
10        b.group(pGroup, cGroup)
11         .channel(NioServerSocketChannel.class)
12         .option(ChannelOption.SO_BACKLOG, 1024)
13         .option(ChannelOption.SO_SNDBUF, 32*1024)
14         .option(ChannelOption.SO_RCVBUF, 32*1024)
15         .childHandler(new ChannelInitializer<SocketChannel>() {
16            @Override
17            protected void initChannel(SocketChannel sc) throws Exception {
18                //1 设置特殊分隔符  
19                ByteBuf buf = Unpooled.copiedBuffer("$_".getBytes());
20                //2
21                sc.pipeline().addLast(new DelimiterBasedFrameDecoder(1024, buf));
22                //3 设置字符串形式的解码
23                sc.pipeline().addLast(new StringDecoder());
24                sc.pipeline().addLast(new ServerHandler());
25            }
26        });
27
28        //4 绑定连接
29        ChannelFuture cf = b.bind(8765).sync();
30
31        //等待服务器监听端口关闭
32        cf.channel().closeFuture().sync();
33        pGroup.shutdownGracefully();
34        cGroup.shutdownGracefully();
35
36    }
37
38}

关于EventLoopGroup、ServerBootstrap等等之类的我都在Netty的第一个程序都讲得很清楚了，需要了解的可以参考我的第一篇文章。

代码说明：

1、 Unpooled.copiedBuffer(“$_”.getBytes()) 这个是设置特殊分隔符返回的是Netty中的ByteBuf类型这里我设置的是 $_

2、DelimiterBasedFrameDecoder（）是处理分隔符的类

3、StringDecoder() 设置字符串形式的解码

服务端业务处理

1public class ServerHandler extends ChannelHandlerAdapter {
2
3    @Override
4    public void channelActive(ChannelHandlerContext ctx) throws Exception {
5        System.out.println(" server channel active... ");
6    }
7
8    @Override
9    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
10        String request = (String)msg;
11        System.out.println("Server :" + msg);
12        String response = "服务器响应：" + msg + "$_";
13        ctx.writeAndFlush(Unpooled.copiedBuffer(response.getBytes()));
14    }
15
16}

由于在服务端就使用了StringDecoder()解码成字符串形式，这里不需要用ByteBuf去转换成字符串。

客户端

1public class Client {
2
3    public static void main(String[] args) throws Exception {
4
5        EventLoopGroup group = new NioEventLoopGroup();
6
7        Bootstrap b = new Bootstrap();
8        b.group(group)
9         .channel(NioSocketChannel.class)
10         .handler(new ChannelInitializer<SocketChannel>() {
11            @Override
12            protected void initChannel(SocketChannel sc) throws Exception {
13                //1
14                ByteBuf buf = Unpooled.copiedBuffer("$_".getBytes());
15                //2
16                sc.pipeline().addLast(new DelimiterBasedFrameDecoder(1024, buf));
17                //3
18                sc.pipeline().addLast(new StringDecoder());
19                sc.pipeline().addLast(new ClientHandler());
20            }
21        });
22
23        ChannelFuture cf = b.connect("127.0.0.1", 8765).sync();
24
25        cf.channel().writeAndFlush(Unpooled.wrappedBuffer("777$_".getBytes()));
26        cf.channel().writeAndFlush(Unpooled.wrappedBuffer("666$_".getBytes()));
27        cf.channel().writeAndFlush(Unpooled.wrappedBuffer("888$_".getBytes()));
28
29
30        //等待客户端端口关闭
31        cf.channel().closeFuture().sync();
32        group.shutdownGracefully();
33
34    }
35}

由于这里客户端也接收服务端返回的数据所以也采用了与服务端一样的处理方式。

客户端业务处理

1public class ClientHandler extends ChannelHandlerAdapter{
2
3    @Override
4    public void channelActive(ChannelHandlerContext ctx) throws Exception {
5        System.out.println("client channel active... ");
6    }
7
8    @Override
9    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
10        try {
11            String response = (String)msg;
12            System.out.println("Client: " + response);
13        } finally {
14            ReferenceCountUtil.release(msg);
15        }
16    }
17}

好！到这第一种解决方案就编写结束了，先启动服务端，再启动客户端

客户端打印如下：

客户端签到后服务端的打印如下：

第二种解决方案（定长）

Netty提供了一个定长类FixdeLengthFraneDecoder。

使用这个定长的有个弊端：如果由多个字段比如可变长度的字段组成时这个时候并解决不了什么问题，建议使用第一个解决方案。

FixdeLengthFraneDecoder的使用跟DelimiterBasedFrameDecoder差不多，由于代码都差不多一样这里我不做太多的说明。

服务端

1public class Server {
2
3    public static void main(String[] args) throws Exception{
4        //创建2个线程，一个是负责接收客户端的连接。一个是负责进行数据传输的
5        EventLoopGroup pGroup = new NioEventLoopGroup();
6        EventLoopGroup cGroup = new NioEventLoopGroup();
7
8        //创建服务器辅助类
9        ServerBootstrap b = new ServerBootstrap();
10        b.group(pGroup, cGroup)
11         .channel(NioServerSocketChannel.class)
12         .option(ChannelOption.SO_BACKLOG, 1024)
13         .option(ChannelOption.SO_SNDBUF, 32*1024)
14         .option(ChannelOption.SO_RCVBUF, 32*1024)
15         .childHandler(new ChannelInitializer<SocketChannel>() {
16            @Override
17            protected void initChannel(SocketChannel sc) throws Exception {
18                //1  设置定长字符串接收  
19                sc.pipeline().addLast(new FixedLengthFrameDecoder(3));
20                //2  设置字符串形式的解码
21                sc.pipeline().addLast(new StringDecoder());
22                sc.pipeline().addLast(new ServerHandler());
23            }
24        });
25
26        //4 绑定连接
27        ChannelFuture cf = b.bind(8765).sync();
28
29        //等待服务器监听端口关闭
30        cf.channel().closeFuture().sync();
31        pGroup.shutdownGracefully();
32        cGroup.shutdownGracefully();
33
34    }
35
36}

1、FixedLengthFrameDecoder(3) 这里设置定长字符串接收具体设置多长自己定。

2、StringDecoder() 设置字符串形式的解码。

服务端业务处理

1public class ServerHandler extends ChannelHandlerAdapter {
2
3
4    @Override
5    public void channelActive(ChannelHandlerContext ctx) throws Exception {
6        System.out.println(" server channel active... ");
7    }
8
9    @Override
10    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
11        String request = (String)msg;
12        System.out.println("Server :" + msg);
13        String response =  request ;
14        ctx.writeAndFlush(Unpooled.copiedBuffer(response.getBytes()));
15    }
16}

服务端

1public class Client {
2
3    public static void main(String[] args) throws Exception {
4
5        EventLoopGroup group = new NioEventLoopGroup();
6
7        Bootstrap b = new Bootstrap();
8        b.group(group)
9         .channel(NioSocketChannel.class)
10         .handler(new ChannelInitializer<SocketChannel>() {
11            @Override
12            protected void initChannel(SocketChannel sc) throws Exception {
13                sc.pipeline().addLast(new FixedLengthFrameDecoder(3));
14                sc.pipeline().addLast(new StringDecoder());
15                sc.pipeline().addLast(new ClientHandler());
16            }
17        });
18
19        ChannelFuture cf = b.connect("127.0.0.1", 8765).sync();
20
21        cf.channel().writeAndFlush(Unpooled.wrappedBuffer("777".getBytes()));
22        cf.channel().writeAndFlush(Unpooled.wrappedBuffer("666".getBytes()));
23        cf.channel().writeAndFlush(Unpooled.wrappedBuffer("888".getBytes()));
24
25        //等待客户端端口关闭
26        cf.channel().closeFuture().sync();
27        group.shutdownGracefully();
28
29    }
30}

客户端业务处理

1public class ClientHandler extends ChannelHandlerAdapter{
2
3    @Override
4    public void channelActive(ChannelHandlerContext ctx) throws Exception {
5        System.out.println("client channel active... ");
6    }
7
8    @Override
9    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
10        try {
11            String response = (String)msg;
12            System.out.println("Client: " + response);
13        } finally {
14            ReferenceCountUtil.release(msg);
15        }
16    }
17}

好！到这第二种解决方案就编写结束了，先启动服务端，再启动客户端

客户端打印如下：

客户端签到后服务端的打印如下：