Netty学习路线

很早就说要写关于Netty的博客了，如今他来了他来了！！！

第一章 Netty入门

1.1 Netty的概述

1.1.1 Netty简介

官网中的解释：

Netty是一个异步事件驱动的网络应用框架，用于快速开发可维护的高性能协议服务器和客户端。(王婆卖瓜自卖自夸)

1.1.2 谁在使用Netty？

Dubbo、zk、RocketMQ、ElasticSearch、Spring5(对HTTP协议的实现)、GRPC、Spark等大型开源项目都在使用Netty作为底层通讯框架

1.1.3 Netty中的核心概念

Channel
- 管道，其是对Socket的封装，其包含了一组API，大大简化了直接与Socket进行操作的复杂性
EventLoopGroup
- EventLoopGroup 是一个EventLoop池，包含很多的EventLoop
- Netty为每个Channel分配了一个EventLoop，用于处理用户连接请求、对用户请求的处理等所有事件。EventLoop本身只是一个线程驱动，在其生命周期内只会绑定一个线程，让该线程处理一个Channel的所有IO事件。
- 一个Channel一旦与一个EventLoop相绑定，那么在Channel的整个生命周期内是不能改变的。一个EventLoop可以与多个Channel绑定。即Channel与EventLoop的关系是n：1，而EventLoop与线程的关系是1:1。
ServerBootStrap
- 用于配制整个Netty代码，将各个组件关联起来。服务端使用的是ServerBootStrap，而客户端使用的是BootStrap。快速开发的脚手架罢了。。
ChannelHandler与ChannelPipeline
- ChannelHandler是对Channel中数据的处理器，这些处理器可以是系统本身定义好的编解码器，也可以是用户自定义的。这些处理器会被统一添加到一个ChannelPipeline的对象中，然后按照添加的顺序对Channel中的数据进行依次处理。
ChannelFuture
- Netty中所有IO操作都是异步的，即操作不会立即得到返回结果，所以Netty中定义了一个ChannelFuture对象作为这个异步操作的“代言人”，表示异步操作本身。如果想获取到该异步操作的返回值，可以通过该异步操作对象的addListener()方法为该异步操作添加监听器，为其注册回调：当结果出来马上调用执行
- Netty的异步编程模型都是简历在Future与回调概念之上的。

1.1.4 Netty执行流程

1.2 小尝试

目的：

通过该程序达到目的是，对Netty编程的基本结构及流程有所了解
该程序是通过Netty实现Http请求的处理，即接受Http，返回Http响应。

1.2.1 创建工程

? 创建一个普通的Maven的Java工程。

1.2.2 导入依赖

? 仅导入一个netty-all依赖即可

<dependencies>
        <!--netty-all-->
        <dependency>
            <groupId>io.netty</groupId>
            <artifactId>netty-all</artifactId>
            <version>4.1.65.Final</version>
        </dependency>
        <!--lombok依赖-->
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <version>1.18.22</version>
            <scope>provided</scope>
        </dependency>
    </dependencies>

1.2.3 定义服务启动类

public class SomeServer {
    public static void main(String[] args) throws InterruptedException {

        // 用于处理客户端连接请求，将请求发送给childGroup中的eventLoop
        EventLoopGroup parentGroup = new NioEventLoopGroup();
        // 用于处理客户端请求
        EventLoopGroup childGroup = new NioEventLoopGroup();

        try {
            // 用户启动ServerChannel
            ServerBootstrap bootstrap = new ServerBootstrap();
            bootstrap.group(parentGroup, childGroup)  // 指定eventLoopGroup
                    .channel(NioServerSocketChannel.class)  // 指定使用NIO进行通信
                    .childHandler(new SomeChannelInitializer());   // 指定childGroup中的eventLoop所绑定的线程所要处理的处理器

            // 指定当前服务器所监听的端口号
            // bind()方法的执行是异步的
            // sync()方法会使bind()操作与后续的代码的执行由异步变为了同步
            ChannelFuture future = bootstrap.bind(8888).sync();
            // System.out.println("服务器启动成功。监听的端口号为：8888");
            // 关闭Channel
            // closeFuture()的执行是异步的。
            // 当Channel调用了close()方法并关闭成功后才会触发closeFuture()方法的执行
            future.channel().closeFuture().sync();
        } finally {
            // 优雅关闭
            parentGroup.shutdownGracefully();
            childGroup.shutdownGracefully();
        }
    }
}

1.2.4 定义服务端处理器

/ 自定义服务端处理器
// 需求：用户提交一个请求后，在浏览器上就会看到hello netty world
public class SomeServerHandler extends ChannelInboundHandlerAdapter {

    /**
     *  当Channel中有来自于客户端的数据时就会触发该方法的执行
     * @param ctx  上下文对象
     * @param msg   就是来自于客户端的数据
     * @throws Exception
     */
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        // System.out.println("msg = " + msg.getClass());
        // System.out.println("客户端地址 = " + ctx.channel().remoteAddress());

        if(msg instanceof HttpRequest) {
            HttpRequest request = (HttpRequest) msg;
            System.out.println("请求方式：" + request.method().name());
            System.out.println("请求URI：" + request.uri());

            if("/favicon.ico".equals(request.uri())) {
                System.out.println("不处理/favicon.ico请求");
                return;
            }

            // 构造response的响应体
            ByteBuf body = Unpooled.copiedBuffer("hello netty world", CharsetUtil.UTF_8);
            // 生成响应对象
            DefaultFullHttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK, body);
            // 获取到response的头部后进行初始化
            HttpHeaders headers = response.headers();
            headers.set(HttpHeaderNames.CONTENT_TYPE, "text/plain");
            headers.set(HttpHeaderNames.CONTENT_LENGTH, body.readableBytes());

            // 将响应对象写入到Channel
            // ctx.write(response);
            // ctx.flush();
            // ctx.writeAndFlush(response);
            ctx.writeAndFlush(response)
                    // 添加监听器，响应体发送完毕则直接将Channel关闭
                    .addListener(ChannelFutureListener.CLOSE);
        }
    }

    /**
     *  当Channel中的数据在处理过程中出现异常时会触发该方法的执行
     * @param ctx  上下文
     * @param cause  发生的异常对象
     * @throws Exception
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
        // 关闭Channel
        ctx.close();
    }
}

1.3 socket

前面的工程是一个仅存在服务端的HTTP请求的服务器，而Netty中最为常见的是C/S架构的Socket代码。所以下面我们就来看一个Netty的Socket通信代码。

1.3.1 创建工程

创建一个普通的Maven的Java工程

本例要实现的功能是：客户端连接上服务端后，其马上会向服务端发送一个数据。服务端在接收到数据后，会马上向客户端也恢复一个数据。客户端每收到服务端的一个数据后，便会再向服务端发送一个数据。而服务端每收到客户端的一个数据后，便会再向客户端发送一个数据。如此反复。(maven依赖同上)

1.3.2 定义服务端

定义服务端启动类

// 定义服务端启动类
public class SomeServer {
    public static void main(String[] args) throws InterruptedException {
        NioEventLoopGroup parentGroup = new NioEventLoopGroup();
        NioEventLoopGroup childGroup = new NioEventLoopGroup();
        try {
            ServerBootstrap bootstrap = new ServerBootstrap();
            bootstrap.group(parentGroup, childGroup)
                     .channel(NioServerSocketChannel.class)
                    .childHandler(new ChannelInitializer<SocketChannel>() {

                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            ChannelPipeline pipeline = ch.pipeline();
                            // StringDecoder：字符串解码器，将Channel中的ByteBuf数据解码为String
                            pipeline.addLast(new StringDecoder());
                            // StringEncoder：字符串编码器，将String编码为将要发送到Channel中的ByteBuf
                            pipeline.addLast(new StringEncoder());
                            pipeline.addLast(new SomeServerHandler());
                        }
                    });
            ChannelFuture future = bootstrap.bind(8888).sync();
            System.out.println("服务器已启动");
            future.channel().closeFuture().sync();
        } finally {
            parentGroup.shutdownGracefully();
            childGroup.shutdownGracefully();
        }
    }
}

定义服务端处理器

public class SomeServerHandler extends ChannelInboundHandlerAdapter {

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        // 将来自于客户端的数据显示在服务端控制台
        System.out.println(ctx.channel().remoteAddress() + "，" + msg);
        // 向客户端发送数据
        ctx.channel().writeAndFlush("from server：" + UUID.randomUUID());
        TimeUnit.MILLISECONDS.sleep(500);
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        ctx.close();
    }
}

1.3.3 定义客户端

定义客户端启动类

public class SomeClient {
    public static void main(String[] args) throws InterruptedException {
        NioEventLoopGroup group = new NioEventLoopGroup();
        try {
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group(group)
                    .channel(NioSocketChannel.class)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            ChannelPipeline pipeline = ch.pipeline();
                            pipeline.addLast(new StringDecoder());
                            pipeline.addLast(new StringEncoder());
                            pipeline.addLast(new SomeClientHandler());
                        }
                    });

            ChannelFuture future = bootstrap.connect("localhost", 8888).sync();
            future.channel().closeFuture().sync();
        } finally {
            group.shutdownGracefully();
        }
    }
}

定义客户端处理器

public class SomeClientHandler extends SimpleChannelInboundHandler<String> {

    // msg的消息类型与类中的泛型类型是一致的
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {
        System.out.println(ctx.channel().remoteAddress() + "，" + msg);
        ctx.channel().writeAndFlush("from client：" + LocalDateTime.now());
        TimeUnit.MILLISECONDS.sleep(500);
    }

    // 当Channel被激活后会触发该方法的执行
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        ctx.channel().writeAndFlush("from client：begin talking");
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        ctx.close();
    }
}

第二章 TCP的拆包与粘包

2.1 拆包粘包简介

Netty在基于TCP协议的网络通信中，存在拆包与粘包情况。拆包与粘包同时发生在数据的发送方与接收方两方
发送方通过网络每发送一批二进制数据包，那么这次所发送的数据包就称为一帧，即Frame。在进行基于TCP的网络传输时，TCP协议会将用户真正要发送的数据根据当前缓存的实际情况对其进行拆分或重组，变为用于网络传输的Frame。在Netty中就是将ByteBuf中的数据拆分或重组为二进制的Frame。而接收方则需要将接收到的Frame中的数据进行重组或拆分，重新恢复为发送方发送时的ByteBuf数据。
具体场景描述
- 发送方发送的ByteBuf较大，在传输之前会被TCP底层拆分为多个Frame进行发送，这个过程称为发送拆包；接收方在接收到需要将这些Frame进行合并，这个合并的过程称为接收方粘包。
- 发送方的ByteBuf较小，无法形成一个Frame，此时TCP底层会将很多的这样的小的ByteBuf合并为一个Frame进行传输，这个合并的过程称为发送方的粘包；接收方在接收到这个Frame后需要进行拆包，拆分出多个原来的小的ByteBuf，这个拆分的过程称为接收方拆包。
- 当一个Frame无法放入整数倍个ByteBuf时，最后一个ByteBuf会发生拆包。这个ByteBuf中的一部分进入到了一个Frame中，另一部分被放入到了另一个Frame中。这个过程就是发送方拆包。但对于将这些ByteBuf放入到一个Frame的过程，就是发送方粘包；当接收方在接收到两个Frame后，对于第一个Frame的最后部分，与第二个