Netty基础—5.Netty的使用简介

大纲

1.Netty服务端的启动流程

2.服务端IO事件的处理类

3.Netty客户端的启动流程

4.客户端IO事件的处理类

5.启动Netty服务端和客户端的方法说明

6.Netty服务端和客户端使用总结

7.什么是TCP粘包拆包

8.TCP粘包拆包的几种情况

9.TCP粘包拆包的原因

10.粘包问题的解决策略

11.拆包的原理

12.粘包问题演示

13.换行符解码器LineBasedFrameDecoder

14.分隔符解码器DelimiterBasedFrameDecoder

15.固定长度解码器FixedLengthFrameDecoder

16.基于长度域解码器LengthFieldBasedDecoder

17.Java序列化的缺点

18.Netty基本组件与BIO的对应

1.Netty服务端的启动流程

步骤一：

首先创建两个NioEventLoopGroup实例，bossGroup实例用于接收客户端的连接，workerGroup实例用于处理每个连接的读写。NioEventLoopGroup是个线程组，它包含了一组NIO线程，专门用于处理网络事件。

步骤二：

然后创建ServerBootstrap实例，ServerBootstrap是Netty用于启动NIO服务端的启动引导类。

步骤三：

接着调用ServerBootstrap的group()方法指定线程模型，也就是将两个NIO线程组传递到ServerBootstrap中。然后调用ServerBootstrap的channel()方法指定IO模型为NIO，也就是指定创建的Channel为NioServerSocketChannel。然后调用ServerBootstrap的option()方法指定TCP参数，也就是配置NioServerSocketChannel的TCP参数。然后调用ServerBootstrap的childHandler()方法指定业务处理逻辑，也就是绑定IO事件的处理类ChildHandler等。

步骤四：

完成服务端的辅助启动类的配置后，就调用它的bind()方法来异步绑定监听端口。然后继续调用它的sync()方法进行同步阻塞来等待绑定操作完成，绑定操作完成后会返回一个ChannelFuture。

步骤五：

接着使用ChannelFuture的方法进行阻塞，直到服务端链路关闭。

步骤六：

最后使用NIO线程组(NioEventLoopGroup)的shutdownGracefully()方法进行优雅退出。

public class NettyServer {
    public static void main(String[] args) {
        NioEventLoopGroup bossGroup = new NioEventLoopGroup();
        NioEventLoopGroup workerGroup = new NioEventLoopGroup();
      
        try {
            ServerBootstrap b = new ServerBootstrap();//相当于Netty的服务器
            b.group(bossGroup, workerGroup)//指定线程模型
            .channel(NioServerSocketChannel.class)//指定IO模型为NIO
            .option(ChannelOption.SO_BACKLOG, 1024)//指定TCP参数
            .childHandler(new ChannelInitializer<NioSocketChannel>() {//指定IO处理逻辑 
                @Override
                protected void initChannel(NioSocketChannel ch) throws Exception {
                    ch.pipeline().addLast(new NettyServerHandler());//针对网络请求的处理逻辑
                }
            });
            ChannelFuture f = b.bind(50070).sync();//绑定端口，同步等待成功
            f.channel().closeFuture().sync();//等待服务端监听端口关闭
        } catch (Exception e) {
            e.printStackTrace();  
        } finally {
            //优雅退出，释放线程资源
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}

2.服务端IO事件的处理类

说明一：

IO事件的处理类继承自ChannelInboundHandlerAdapter，服务端IO事件的处理类主要需要关注三个方法：channelRead()、channelReadComplete()、exceptionCaught()。

说明二：

ByteBuf类似于NIO的ByteBuffer，但功能更强大和灵活。通过ByteBuf的readableBytes()方法可以获得缓冲区可读的字节数，然后就可以根据缓冲区可读的字节数创建byte数组，接着通过ByteBuf的readBytes()方法便可以将缓冲区的字节数组复制到新创建的byte数组中。

说明三：

通过ChannelHandlerContext的write()方法会把待发送的消息放到发送缓冲区中，通过ChannelHandlerContext的flush()方法会将发送缓冲区中的消息写入到SocketChannel中发送出去。

为了防止频繁唤醒Selector进行消息发送，ChannelHandlerContext的write()方法并不直接将消息写入SocketChannel中，而只是把消息放到发送缓冲区中。当调用ChannelHandlerContext的flush()方法时，才会将发送缓冲区中的消息写入到SocketChannel中发送出去。

public class NettyServerHandler extends ChannelInboundHandlerAdapter {
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        ByteBuf requestBuffer = (ByteBuf) msg;
        byte[] requestBytes = new byte[requestBuffer.readableBytes()];
        requestBuffer.readBytes(requestBytes);
        
        String request = new String(requestBytes, "UTF-8");
        System.out.println("接收到的请求：" + request); 
        
        String response = "你好，我收到你的消息了";
        ByteBuf responseBuffer = Unpooled.copiedBuffer(response.getBytes());
        ctx.write(responseBuffer);
        
        //Netty底层就有类似Processor的东西，负责从网络连接中读取请求
        //然后把读取出来的请求交给这里的Handler来处理，处理完以后把响应返回回去
    }
   
    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        ctx.flush();
    }
   
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
        ctx.close();
    }
}

3.Netty客户端的启动流程

步骤一：

首先创建客户端处理IO读写的NioEventLoopGroup实例。NioEventLoopGroup是个线程组，它包含了一组NIO线程，专门用于处理网络事件。

步骤二：

然后创建Bootstrap实例，Bootstrap是Netty用于启动NIO客户端的启动引导类。

步骤三：

接着调用Bootstrap的group()、channel()、option()、handler()方法配置好：线程模型、IO模型、TCP参数、业务处理逻辑。

步骤四：

完成客户端启动辅助类的配置后，就调用它的connect()方法来异步发起连接，然后调用sync()方法进行同步阻塞来等待连接成功。

步骤五：

接着使用ChannelFuture的方法进行阻塞，直到客户端连接关闭。

步骤六：

最后使用NIO线程组(NioEventLoopGroup)的shutdownGracefully()方法进行优雅退出。

public class NettyClient {
    public static void main(String[] args) {
        NioEventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group(group)//指定线程模型
            .channel(NioSocketChannel.class)//指定IO模型为NIO
            .option(ChannelOption.TCP_NODELAY, true)//指定TCP参数
            .handler(new ChannelInitializer<Channel>() {//指定IO处理逻辑
                @Override
                protected void initChannel(Channel channel) throws Exception {
                    channel.pipeline().addLast(new NettyClientHandler());  
                }
            });
         
            ChannelFuture f = bootstrap.connect("127.0.0.1", 50070).sync();//建立连接
            f.channel().closeFuture().sync();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            group.shutdownGracefully();
        }
    }
}

4.客户端IO事件的处理类

说明一：

IO事件的处理类继承自ChannelInboundHandlerAdapter，客户端IO事件的处理类主要需要关注三个方法：channelActive()、channelRead()、exceptionCaught()。

说明二：

当客户端和服务端建立好TCP连接后，Netty的NIO线程会调用channelActive()方法，而调用ChannelHandlerContext的writeAndFlush()方法会将请求消息发送给服务端。

说明三：

当服务端返回应答消息时，channelRead()方法会被调用。当发生异常时，exceptionCaught()方法会被调用。

注意：Netty里的数据是以ByteBuf为单位的，所有需要读和写的数据都必须放到一个ByteBuf中。其中通过ctx.alloc().buffer()可以分配一个ByteBuf，通过Unpooled.buffer()也可以分配一个ByteBuf。

public class NettyClientHandler extends ChannelInboundHandlerAdapter {
    private ByteBuf requestBuffer;
   
    public NettyClientHandler() {
        byte[] requestBytes = "你好，我发送第一条消息".getBytes();
        requestBuffer = Unpooled.buffer(requestBytes.length);
        requestBuffer.writeBytes(requestBytes);
    }
   
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        ctx.writeAndFlush(requestBuffer);
    }
    
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        ByteBuf responseBuffer = (ByteBuf) msg;
        byte[] responseBytes = new byte[responseBuffer.readableBytes()];
        responseBuffer.readBytes(responseBytes);
      
        String response = new String(responseBytes, "UTF-8"); 
        System.out.println("接收到服务端的响应：" + response);  
    }
   
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
        ctx.close();
    }
}

5.启动Netty服务端和客户端的方法说明

一.ServerBoostrap的bind()方法

bind()方法是一个异步方法，调用之后会立即返回一个ChannelFuture，通过ChannelFuture的addListener()方法可以添加监听器监听端口是否绑定成功。

二.ServerBoostrap的handler()方法

childHandler()方法用于处理新连接的数据读写逻辑，handler()方法用于处理服务端启动过程中的一些逻辑，但一般不用该方法。

三.ServerBoostrap的attr()方法

attr()方法可以给服务端Channel也就是NioServerSocketChannel指定自定义属性，然后可以通过channel.attr()取出这些属性。

四.ServerBoostrap的childAttr()方法

childAttr()方法也可以给每个连接指定自定义属性。

五.ServerBoostrap的option()方法

option()方法可以给服务端Channel也就是NioServerSocketChannel设置TCP参数，比如b.option(ChannelOption.SO_BACKLOG, 1024)就是设置临时存放已完成三次握手的请求的队列的最大长度。

六.ServerBoostrap的childOption()方法

childOption()方法可以给每一个连接都设置一些TCP参数，ChannelOption.SO_KEEPALIVE表示是否开启TCP底层的心跳机制，ChannelOption.TCP_NODELAY表示是否开启Nagle算法。如果要求实时性则可以关闭Nagle算法，如果要求减少发送次数和网络交互次数则可以开启Nagle算法。

七.Boostrap的connect()方法

connect()方法是一个异步方法，调用之后会立即返回一个ChannelFuture，通过ChannelFuture的addListener()方法可以添加监听器监听连接是否建立成功。

八.Boostrap的attr()方法

attr()方法可以给客户端Channel也就是NioSocketChannel指定自定义属性。

九.Boostrap的option()方法

option()方法可以给客户端Channel也就是NioSocketChannel设置TCP参数。

6.Netty服务端和客户端使用总结

一.Netty服务端的启动流程

首先创建一个服务端的启动引导类ServerBootstrap，然后给它指定线程模型、IO模型、TCP参数、业务处理逻辑，最后通过它的bind()方法绑定端口后，服务端就启动起来了。

二.Netty客户端的启动流程

首先创建一个客户端的启动引导类Bootstrap，然后给它指定线程模型、IO模型、TCP参数、业务处理逻辑，最后通过它的connect()方法连接上服务端后，客户端就启动起来了。

三.bind()方法和connect()方法

bind()方法和connect()方法都是异步的，这两方法调用后都会立即返回一个ChannelFuture，都可以通过ChannelFuture的addListener()方法添加监听器监听端口是否绑定成功以及连接是否建立成功。

四.Netty的基本组件与BIO的对应

NIO的三大组件是：Buffer、Channel、Selector。NioEventLoop对应于BIO的线程(监听客户端连接 + 处理客户端连接的读写)。Channel对应于BIO的Socket，ByteBuf对应于BIO的IO Bytes，Pipeline对应于BIO的逻辑处理链，ChannelHandler对应于BIO的逻辑处理块。

7.什么是TCP粘包拆包

TCP协议是一个流协议。所谓流，就是没有界限的一串数据。就像河里的流水，它们是连成一片的，其中并没有分界线。

TCP协议的底层并不了解上层业务数据的具体含义，它会根据TCP缓冲区的实际情况进行包的划分。所以在业务上认为，一个完整的包可能会被拆分成多个包进行发送，也有可能把多个小的包封装成一个大的数据包进行发送。这就是TCP粘包和拆包问题。

8.TCP粘包拆包的几种情况

假设客户端发送了两个数据包D1和D2给服务端，由于服务端一次读取到的字节数是不确定的，故可能有如下情况：

情况一：服务端分两次读取到了两个独立的数据包

也就是数据包D1和D2是分开的，没有出现粘包和拆包的情况。

情况二：服务端一次接收到了两个数据包

也就是数据包D1和D2粘在一起了，出现了TCP粘包的情况。

情况三：服务端分两次读取到了两个数据包

第一次读取到了完整的D1包和D2包的部分内容，第二次读取到了D2包的剩余内容，出现了TCP拆包的情况。

情况四：服务端分两次读取到了两个数据包

第一次读取到了D1包的部分内容，第二次读取到了D1包的剩余内容和完整的D2包，出现了TCP拆包的情况。

情况五：服务端分多次读取到两个数据包

如果服务端的TCP接收滑窗非常小，而数据包D1和D2比较大，那么服务端可能要分多次才能将D1和D2包接收完，期间发生了多次拆包。

9.TCP粘包拆包的原因

(1)粘包问题的原因

由于TCP协议是基于三次握手的可靠的传输协议，会让客户端与服务端维持一个连接。所以在连接不断开的情况下，客户端能够持续不断地将多个数据包发往服务端。

但如果发送的数据包太小，则会因Nagle算法而对较小的数据包进行合并再发送。开启Nagle算法会让TCP的网络延迟高一些，当然可以设置关闭Nagle算法。这样服务端在收到数据包时就无法区分哪些数据包是独立的，从而产生粘包问题。

服务端在收到数据包后会将其放到缓冲区中，如果数据包没有及时从缓存区取走，那么下次获取数据包时就可能会出现一次取出多个数据包的情况，从而产生粘包问题。

由于UDP协议本身作为无连接的不可靠的传输协议(适合频繁发送较小的数据包)，所以不会对数据包进行合并发送(也就不会使用Nagle算法)。每一个数据包都是独立和完整的，但服务端接收的缓冲区大小也会影响是否出现粘包问题。

(2)半包问题的原因

产生半包问题的核心原因就是一个数据包被分成了多次接收，比如：

一.可能是IP分片传输导致的

二.可能是传输过程中丢失部分包导致出现半包

三.可能是一个包被分成了两次传输，在获取数据时先取到一部分

四.可能与接收的缓冲区大小有关系，在获取数据时先取到一部分

半包问题更具体的原因有三个，分别如下：

一.应用程序写入数据的字节大小大于Socket套接字发送缓冲区的大小

二.进行MSS大小的TCP分段

MSS是最大报文段长度的缩写，MSS是TCP报文段中的数据字段的最大长度。数据字段加上TCP首部才等于整个TCP报文段，所以MSS并不是TCP报文段的最大长度，而是TCP报文段长度减去TCP首部长度。

三.以太网的payload大于MTU进行IP分片

MTU指一种通信协议的某一层上面所能通过的最大数据包大小。如果IP层有一个数据包要传，而且数据的长度比链路层的MTU大，那么IP层就会进行分片，把数据包分成若干片，让每一片都不超过MTU。注意IP分片可以发生在原始发送端主机上，也可以发生在中间路由器上。

10.粘包问题的解决策略

由于TCP协议的底层无法理解上层的业务数据，所以在底层是无法保证数据包不被拆分和重组的，这个问题只能通过上层的应用协议栈设计来解决。

策略一：消息定长

例如每个报文的大小为固定长度200字节，如果不够，空位补空格。FixedLengthFrameDecoder

策略二：加分割符

在包尾增加回车换行符等特殊分隔符进行分割，如FTP协议。LineBasedFrameDecoder、DelimiterBasedFrameDecoder

策略三：消息带上长度字段

比如将消息分为消息头和消息体，消息头中包含消息的总长度。LengthFieldBasedFrameDecoder

11.拆包的原理

拆包的基本原理就是不断从TCP缓冲区中读取数据，每次读取完都要判断是否是一个完整的数据包。

如果当前读取的数据不足以拼接成一个完整的业务数据包，那就保留该数据，继续从TCP缓冲区中读取，直到得到一个完整的数据包。

如果当前读取的数据加上已读取的数据足以拼接成一个完整的业务数据包，那就将已读取的数据拼接上当前读取的数据，构成一个完整的业务数据包传递到业务逻辑，多余的数据仍然保留，以便和下一次读取的数据进行拼接。

12.粘包问题演示

客户端和服务端链路建立成功后，就触发执行channelActive()方法循环发送100条消息。每发送一条消息就通过writeAndFlush()方法刷新一次，保证每条消息都被写入Channel。

期望服务端应该会收到100条消息，但实际上只收到了两条消息。一条包含"57条消息"，另一条包含"43条消息"。

public class NettyClientHandler extends ChannelInboundHandlerAdapter {
    public void channelActive(ChannelHandlerContext ctx) {
        for (int i = 0; i < 100; i++) {
            ByteBuf buffer = ctx.alloc().buffer();
            buffer.writeBytes("Hello".getBytes());
            ctx.channel().writeAndFlush(buffer);
        }
    }
}

public class NettyServerHandler extends ChannelInboundHandlerAdapter {
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        ByteBuf byteBuf = (ByteBuf) msg;
        System.out.println("服务端读到数据" + byteBuf.toString());
    }
}

13.换行符解码器LineBasedFrameDecoder

LineBasedFrameDecoder也可以称为行拆包器。从字面意思看，发送端发送数据包时，每个数据包之间以换行符作为分隔，接收端通过这个行拆包器将粘过的ByteBuf拆分成一个个完整的应用层数据包。

它的工作原理是依次遍历ByteBuf中的可读字节，然后判断是否有"\n"或者"\r\n"。如果有，就以此位置为结束位置，从可读索引到结束位置区间的字节就组成了一行。

它是以换行符为结束标志的解码器，支持配置单行的最大长度。如果连续读取到最大长度后仍然没有发现换行符，那么就会抛出异常，并忽略之前读取到的异常码流。

//添加换行符解码器
//在绝大多数情况下，这些解码器会添加到对应ChannelPipeline的起始位置
ch.pipeline().addLast(new LineBasedFrameDecoder(1024));

//响应消息要拼接换行符
ByteBuf resp = Unpooled.copiedBuffer(("响应内容" + System.getProperty("line.separator")).getBytes());
ctx.writeAndFlush(resp);

14.分隔符解码器DelimiterBasedFrameDecoder

DelimiterBasedFrameDecoder也可以称为分隔符拆包器，它是行拆包器的通用版本，可以传递两个参数，可以自定义分隔符。

第一个参数1024表示单条消息的最大长度，当达到该长度后仍然没有查找到分隔符则抛异常。第二个参数就是分隔符ByteBuf对象。

//添加分隔符解码器
//在绝大多数情况下，这些解码器会添加到对应ChannelPipeline的起始位置
ByteBuf delimiter = Unpooled.copiedBuffer("$_".getBytes());
ch.pipeline().addLast(new DelimiterBasedFrameDecoder(1024, delimiter));

//响应消息要拼接分隔符
ByteBuf resp = Unpooled.copiedBuffer(("响应内容" + "$_").getBytes());
ctx.writeAndFlush(resp);

15.固定长度解码器FixedLengthFrameDecoder

FixedLengthFrameDecoder也可以称为固定长度拆包器。如果应用层协议非常简单，每个数据包的长度都是固定的，比如100。那么只需要把这个固定长度解码器添加到Pipeline中，Netty就会把一个个长度为100的数据包(ByteBuf)传递到下一个ChannelHandler。

因为利用固定长度解码器，无论一次接收多少数据，都会按照构造函数中设置的固定长度进行解码。如果是半包消息，固定长度解码器会缓存半包消息并等下一个数据包到达后再进行拼包处理。

//添加固定长度解码器
//在绝大多数情况下，这些解码器会添加到对应ChannelPipeline的起始位置
ch.pipeline().addLast(new FixedLengthFrameDecoder(200));

16.基于长度域解码器LengthFieldBasedDecoder

LengthFieldBasedDecoder也可以称为基于长度域的拆包器，这种基于长度域的拆包器是最通用的一种拆包器。只要我们的自定义协议中包含长度域字段，均可以使用这个解码器来实现应用层拆包。

假如一个自定义协议的数据结构如下，那么长度域的偏移量就是4 + 1 + 1 + 1 = 7。

于是就可以根据长度域偏移量和长度域的长度来构造一个解码器，其中第一个参数是数据包的最大长度，第二个参数是长度域的偏移量，第三个参数是长度域的长度。

//添加基于长度域的解码器
//在绝大多数情况下，这些解码器会添加到对应ChannelPipeline的起始位置
ch.pipeline().addLast(new LengthFieldBasedDecoder(Integer.MAX_VALUE, 7, 4));

注意：设计协议时通常会在数据包的开头加上一个魔数，以便可以尽早屏蔽非本协议的客户端。通常会在第一个Handler对每个客户端发过来的数据包做一次快速判断，看是否满足自定义协议。比如可以继承LengthFieldBasedDecoder的decode()方法，然后在decode()之前判断前4字节是否等于定义的魔数。

17.Java序列化的缺点

缺点一：无法跨语言

缺点二：序列化的码流太大

缺点三：序列化性能太差

所以一般使用Protobuf、Hessian、Kyro等工具进行序列化。

18.Netty基本组件与BIO的对应

Channel -> BIO的Socket

ByteBuf -> BIO的IO Bytes

Pipeline -> BIO的逻辑处理链

ChannelHandler -> BIO的逻辑处理块

NioEventLoop -> BIO的线程(监听客户端连接 + 处理客户端连接的读写)