Netty网络编程第三卷

Posted 大忽悠爱忽悠

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了Netty网络编程第三卷相关的知识,希望对你有一定的参考价值。

Netty网络编程第三卷


三. Netty 进阶

1. 粘包与半包

1.1 粘包现象

服务端代码

/**
 * @author zdh
 */
public class HelloWorldServer 
    static final Logger log = LoggerFactory.getLogger(HelloWorldServer.class);
    void start() 
        //处理客户端连接
        NioEventLoopGroup boss = new NioEventLoopGroup(1);
        //处理客户端读写请求
        NioEventLoopGroup worker = new NioEventLoopGroup();
        try 
            ServerBootstrap serverBootstrap = new ServerBootstrap();
            serverBootstrap.channel(NioserverSocketChannel.class);
            serverBootstrap.group(boss, worker);
            //设置接收缓冲区的大小,为了让黏包和半包现象明显
            serverBootstrap.option(ChannelOption.SO_RCVBUF,10);
            serverBootstrap.childHandler(new ChannelInitializer<SocketChannel>() 
                //连接建立时触发
                @Override
                protected void initChannel(SocketChannel ch) throws Exception 
                    //Netty的日记输出级别设置
                    ch.pipeline().addLast(new LoggingHandler(LogLevel.DEBUG));
                    ch.pipeline().addLast(new ChannelInboundHandlerAdapter() 
                        //会在连接建立的时候触发,触发active事件
                        @Override
                        public void channelActive(ChannelHandlerContext ctx) throws Exception 
                            log.debug("connected ", ctx.channel());
                            super.channelActive(ctx);
                        

                        //连接断开时触发
                        @Override
                        public void channelInactive(ChannelHandlerContext ctx) throws Exception 
                            log.debug("disconnect ", ctx.channel());
                            super.channelInactive(ctx);
                        
                    );
                
            );
            ChannelFuture channelFuture = serverBootstrap.bind(8080);
            log.debug(" binding...", channelFuture.channel());
            channelFuture.sync();
            log.debug(" bound...", channelFuture.channel());
            channelFuture.channel().closeFuture().sync();
         catch (InterruptedException e) 
            log.error("server error", e);
         finally 
            boss.shutdownGracefully();
            worker.shutdownGracefully();
            log.debug("stoped");
        
    

    public static void main(String[] args) 
        new HelloWorldServer().start();
    

客户端代码希望发送 10 个消息,每个消息是 16 字节

/**
 * @author zdh
 */
public class HelloWorldClient 
    static final Logger log = LoggerFactory.getLogger(HelloWorldClient.class);
    public static void main(String[] args) 
        NioEventLoopGroup worker = new NioEventLoopGroup();
        try 
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.channel(NioSocketChannel.class);
            bootstrap.group(worker);
            bootstrap.handler(new ChannelInitializer<SocketChannel>() 
                @Override
                protected void initChannel(SocketChannel ch) throws Exception 
                    log.debug("connetted...");
                    ch.pipeline().addLast(new ChannelInboundHandlerAdapter() 
                        //连接建立成功后,会触发channelActive事件
                        @Override
                        public void channelActive(ChannelHandlerContext ctx) throws Exception 
                            log.debug("sending...");
                           //总共发送160个字节的数据给服务器端
                            for (int i = 0; i < 10; i++)
                            
                                //创建一个默认大小为256自己的直接缓冲区
                                ByteBuf buffer = ctx.alloc().buffer();
                                //向直接缓冲区中写入16个字节的数据
                                buffer.writeBytes(new byte[]0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15);
                                //发送给服务器
                                ctx.writeAndFlush(buffer);
                            
                        
                    );
                
            );
            ChannelFuture channelFuture = bootstrap.connect("127.0.0.1", 8080).sync();
            channelFuture.channel().closeFuture().sync();

         catch (InterruptedException e) 
            log.error("client error", e);
         finally 
            worker.shutdownGracefully();
        
    


1.2 半包现象

客户端代码希望发送 1 个消息,这个消息是 160 字节,代码改为

ByteBuf buffer = ctx.alloc().buffer();
for (int i = 0; i < 10; i++) 
    buffer.writeBytes(new byte[]0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15);

ctx.writeAndFlush(buffer);

为现象明显,服务端修改一下接收缓冲区,其它代码不变

serverBootstrap.option(ChannelOption.SO_RCVBUF, 10);

服务器端的某次输出,可以看到接收的消息被分为两节,第一次 20 字节,第二次 140 字节

08:43:49 [DEBUG] [main] c.i.n.HelloWorldServer - [id: 0x4d6c6a84] binding...
08:43:49 [DEBUG] [main] c.i.n.HelloWorldServer - [id: 0x4d6c6a84, L:/0:0:0:0:0:0:0:0:8080] bound...
08:44:23 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0x1719abf7, L:/127.0.0.1:8080 - R:/127.0.0.1:59221] REGISTERED
08:44:23 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0x1719abf7, L:/127.0.0.1:8080 - R:/127.0.0.1:59221] ACTIVE
08:44:23 [DEBUG] [nioEventLoopGroup-3-1] c.i.n.HelloWorldServer - connected [id: 0x1719abf7, L:/127.0.0.1:8080 - R:/127.0.0.1:59221]
08:44:24 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0x1719abf7, L:/127.0.0.1:8080 - R:/127.0.0.1:59221] READ: 20B
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 00 01 02 03 04 05 06 07 08 09 0a 0b 0c 0d 0e 0f |................|
|00000010| 00 01 02 03                                     |....            |
+--------+-------------------------------------------------+----------------+
08:44:24 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0x1719abf7, L:/127.0.0.1:8080 - R:/127.0.0.1:59221] READ COMPLETE
08:44:24 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0x1719abf7, L:/127.0.0.1:8080 - R:/127.0.0.1:59221] READ: 140B
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 04 05 06 07 08 09 0a 0b 0c 0d 0e 0f 00 01 02 03 |................|
|00000010| 04 05 06 07 08 09 0a 0b 0c 0d 0e 0f 00 01 02 03 |................|
|00000020| 04 05 06 07 08 09 0a 0b 0c 0d 0e 0f 00 01 02 03 |................|
|00000030| 04 05 06 07 08 09 0a 0b 0c 0d 0e 0f 00 01 02 03 |................|
|00000040| 04 05 06 07 08 09 0a 0b 0c 0d 0e 0f 00 01 02 03 |................|
|00000050| 04 05 06 07 08 09 0a 0b 0c 0d 0e 0f 00 01 02 03 |................|
|00000060| 04 05 06 07 08 09 0a 0b 0c 0d 0e 0f 00 01 02 03 |................|
|00000070| 04 05 06 07 08 09 0a 0b 0c 0d 0e 0f 00 01 02 03 |................|
|00000080| 04 05 06 07 08 09 0a 0b 0c 0d 0e 0f             |............    |
+--------+-------------------------------------------------+----------------+
08:44:24 [DEBUG] [nioEventLoopGroup-3-1] i.n.h.l.LoggingHandler - [id: 0x1719abf7, L:/127.0.0.1:8080 - R:/127.0.0.1:59221] READ COMPLETE

注意

serverBootstrap.option(ChannelOption.SO_RCVBUF, 10) 影响的底层接收缓冲区(即滑动窗口)大小,仅决定了 netty 读取的最小单位,netty 实际每次读取的一般是它的整数倍


1.3 现象分析

粘包

  • 现象,发送 abc def,接收 abcdef
  • 原因
    • 应用层:接收方 ByteBuf 设置太大(Netty 默认 1024)
    • 滑动窗口:假设发送方 256 bytes 表示一个完整报文,但由于接收方处理不及时且窗口大小足够大,这 256 bytes 字节就会缓冲在接收方的滑动窗口中,当滑动窗口中缓冲了多个报文就会粘包
    • Nagle 算法:会造成粘包

半包

  • 现象,发送 abcdef,接收 abc def
  • 原因
    • 应用层:接收方 ByteBuf 小于实际发送数据量
    • 滑动窗口:假设接收方的窗口只剩了 128 bytes,发送方的报文大小是 256 bytes,这时放不下了,只能先发送前 128 bytes,等待 ack 后才能发送剩余部分,这就造成了半包
    • MSS 限制:当发送的数据超过 MSS 限制后,会将数据切分发送,就会造成半包

本质是因为 TCP 是流式协议,消息无边界

滑动窗口

  • TCP 以一个段(segment)为单位,每发送一个段就需要进行一次确认应答(ack)处理,但如果这么做,缺点是包的往返时间越长性能就越差

  • 为了解决此问题,引入了窗口概念,窗口大小即决定了无需等待应答而可以继续发送的数据最大值

  • 窗口实际就起到一个缓冲区的作用,同时也能起到流量控制的作用

    • 图中深色的部分即要发送的数据,高亮的部分即窗口
    • 窗口内的数据才允许被发送,当应答未到达前,窗口必须停止滑动
    • 如果 1001~2000 这个段的数据 ack 回来了,窗口就可以向前滑动
    • 接收方也会维护一个窗口,只有落在窗口内的数据才能允许接收

MSS 限制

  • 链路层对一次能够发送的最大数据有限制,这个限制称之为 MTU(maximum transmission unit),不同的链路设备的 MTU 值也有所不同,例如

  • 以太网的 MTU 是 1500

  • FDDI(光纤分布式数据接口)的 MTU 是 4352

  • 本地回环地址的 MTU 是 65535 - 本地测试不走网卡

  • MSS (传输层)是最大段长度(maximum segment size),它是 MTU 刨去 tcp 头和 ip 头后剩余能够作为数据传输的字节数

  • ipv4 tcp 头占用 20 bytes,ip 头占用 20 bytes,因此以太网 MSS 的值为 1500 - 40 = 1460

  • TCP 在传递大量数据时,会按照 MSS 大小将数据进行分割发送

  • MSS 的值在三次握手时通知对方自己 MSS 的值,然后在两者之间选择一个小值作为 MSS


Nagle 算法

  • 即使发送一个字节,也需要加入 tcp 头和 ip 头,也就是总字节数会使用 41 bytes,非常不经济。因此为了提高网络利用率,tcp 希望尽可能发送足够大的数据,这就是 Nagle 算法产生的缘由
  • 该算法是指发送端即使还有应该发送的数据,但如果这部分数据很少的话,则进行延迟发送
    • 如果 SO_SNDBUF 的数据达到 MSS,则需要发送
    • 如果 SO_SNDBUF 中含有 FIN(表示需要连接关闭)这时将剩余数据发送,再关闭
    • 如果 TCP_NODELAY = true,则需要发送
    • 已发送的数据都收到 ack 时,则需要发送
    • 上述条件不满足,但发生超时(一般为 200ms)则需要发送
    • 除上述情况,延迟发送

1.4 解决方案

  1. 短链接,发一个包建立一次连接,这样连接建立到连接断开之间就是消息的边界,缺点效率太低
  2. 每一条消息采用固定长度,缺点浪费空间
  3. 每一条消息采用分隔符,例如 \\n,缺点需要转义
  4. 每一条消息分为 head 和 body,head 中包含 body 的长度

方法1,短链接

以解决粘包为例

public class HelloWorldClient 
    static final Logger log = LoggerFactory.getLogger(HelloWorldClient.class);

    public static void main(String[] args) 
        // 分 10 次发送
        for (int i = 0; i < 10; i++) 
            send();
        
    

    private static void send() 
        NioEventLoopGroup worker = new NioEventLoopGroup();
        try 
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.channel(NioSocketChannel.class);
            bootstrap.group(worker);
            bootstrap.handler(new ChannelInitializer<SocketChannel>() 
                @Override
                protected void initChannel(SocketChannel ch) throws Exception 
                    log.debug("conneted...");
                    ch.pipeline().addLast(new LoggingHandler(LogLevel.DEBUG));
                    ch.pipeline().addLast(new ChannelInboundHandlerAdapter() 
                        @Override
                        public void channelActive(ChannelHandlerContext ctx) throws Exception 
                            log.debug("sending...");
                            ByteBuf buffer = ctx.alloc().buffer();
                            buffer.writeBytes(new byte[]0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15);
                            ctx.writeAndFlush(buffer);
                            // 发完即关
                            ctx.close();
                        
                    );
                
            );
            ChannelFuture channelFuture = bootstrap.connect("localhost", 8080).sync();
            channelFuture.channel().closeFuture().sync();

         catch (InterruptedException e) 
            log.error("client error", e);
         finally 
            worker.shutdownGracefully();
        
    

服务端:

            //设置系统的接收缓冲区(滑动窗口)
      //      serverBootstrap.option(ChannelOption.SO_RCVBUF,10);
        //调整netty的接收缓冲区(byteBuf)
            serverBootstrap.childOption(ChannelOption.RCVBUF_ALLOCATOR,new AdaptiveRecvByteBufAllocator(16,16,16));

SO_RCVBUF和SO_SNDBUF每个套接口都有一个发送缓冲区和一个接收缓冲区,使用这两个套接口选项可以改变缺省缓冲区大小。

当设置TCP套接口接收缓冲区的大小时,函数调用顺序是很重要的,因为TCP的窗口规模选项是在建立连接时用SYN与对方互换得到的。对于客户,SO_RCVBUF选项必须在connect之前设置;对于服务器,SO_RCVBUF选项必须在listen前设置。

设置netty接收缓冲区大小为16字节的情况下:

半包用这种办法还是不好解决,因为接收方的缓冲区大小是有限的


方法2,固定长度

让所有数据包长度固定(假设长度为 8 字节),服务器端加入

//第一个入站处理器
ch.pipeline().addLast(new FixedLengthFrameDecoder(8));

客户端测试代码,注意, 采用这种方法后,客户端什么时候 flush 都可以

public class HelloWorldClient 
    static final Logger log = LoggerFactory.getLogger(HelloWorldClient.class);

    public static void main(String[] args) 
        NioEventLoopGroup worker = new NioEventLoopGroup();
        try 
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.channel(NioSocketChannel.class);
            bootstrap.group(worker);
            bootstrap.handler(new ChannelInitializer<SocketChannel>() 
                @Override
                protected void initChannel(SocketChannel ch) throws Exception 
                    log.debug("connetted...");
                    ch.pipeline().addLast(new LoggingHandler(LogLevel.DEBUG));
                    ch.pipeline().addLast(new ChannelInboundHandlerAdapter() 
                        @Override
                        public void channelActive(ChannelHandlerContext ctx) throws Exception 
                            log.debug("sending...");
                            // 发送内容随机的数据包
                            Random r = new Random();
                            char c = 'a';
                            ByteBuf buffer = ctx.alloc().buffer();
                            for (int i = 0; i < 10; i++) 
                                byte[] bytes = new byte[8];
                                for (int j = 0; j < r.nextInt(8); j++) 
                                    bytes[j] = (byte) c;
                                
                                c++;
                                buffer.writeBytes(bytes);
                            
                            ctx.writeAndFlush(buffer);
                        

以上是关于Netty网络编程第三卷的主要内容,如果未能解决你的问题,请参考以下文章

普林斯顿数学指南(第三卷)

python3多线程应用详解(第三卷:图解多线程中join,守护线程应用)

数学 它的内容,方法和意义 第三卷

计算机程序设计艺术(第三卷)pdf

纪念那个中国最后的私塾老先生——朱执中转载

中间件兴趣圈第三本电子书重磅来袭