[网络协议]Netty详解

介绍

1. Netty是由JBoos提供的一个Java网络应用开源框架，它是一个异步的，基于事件驱动的网络应用框架，用于开发高性能，高可靠的网络IO程序
2. Netty主要针对TCP协议下，面向客户端的高并发应用，或者在Peer0toPeer场景下的大量数据持续传输的应用。
3. Netty本质上是一个NIO框架，适用于服务器通信相关的多种应用场景，使用Netty可以帮助我们快速，简单的开发出一个网络应用；

Netty核心API介绍

ChannelHandler及其实现类
ChannelHandler接口定义了许多事件处理的方法，我们可以通过重写这些方法实现具体的业务逻辑，以下是ChannelHandler接口的类图继承关系

Netty开发中需要自定义一个Handler类去实现ChannelHandler接口或者子接口或者接口的实现类，然后通过重写相应的方法来实现业务逻辑，下面是一般都需要重写的方法

// 通道就绪事件
public void channelActive(ChannelHandlerContext ctx);

// 通道读取数据事件
public void channelRead(ChannelHandlerContext ctx,Object msg);
      
// 数据读取完毕事件    
public void channelReadComplete(ChannelHandlerContext ctx);

// 数据发生异常事件    
public void exceptionCaught(ChannelHandlerContext ctx,Throwable cause);

ChannelHandlerContext常用方法

// 关闭通道
ChannelFuture close()
    
 // 刷新 
ChannelOutboundInvoker flush()
    
    
//将数据写到 ChannelPipeline 中当前 ChannelHandler的下一个 ChannelHandler 开始处理（出站）
ChannelFuture writeAndFlush(Object msg)

ChannelPipeline

ChannelPipeline 是一个ChannelHandler的集合，它负责处理和拦截 inbound（请求入站）和 outbound（请求出站）的事件和操作，相当于一个贯穿Netty的责任链

ChannelOption
Netty在创建Channel实例后，一般都需要设置ChannelOption 参数。ChannelOption是Socket的标准参数，而非Netty独创。常用的参数配置有：

● ChannelOption.SO_BACKLOG 对应TCP/IP协议listen函数中的backlog参数，用来初始化服务器的可连接队列大小。服务端处理客户端请求是顺序处理的，所以同一时间只能处理一个客户端连接。多个客户端来的时候，服务端将不能处理的客户端请求放在队列中等待处理，backlog参数指定了队列大小

● ChannelOption.SO_KEEPALIVE 一直保持连接的活动状态，该参数用于设置TCP连接，当设置该选项以后，链接时会测试连接的状态，这个选项用于可能长时间没有数据交流的连接，当设置这个选项以后，如果两个小时没有数据通信时，TCP会自动发送一个活动探测数据报文

ChannelFuture
表示Channel通道中一步IO操作的结果，在Netty中所有的IO操作都是异步的，调用者并不能立刻获取结果，但是可以通过ChannelFuture来获取IO操作的处理状态
常用方法如下：

// 返回当前正在进行IO操作的通道
Channel  channel(); 

// 等待异步操作执行完毕，将异步改为同步
ChannelFuture  sync();

EventLoopGroup和实现类NioEventLoopGroup

EventLoopGroup是一组EventLoop的抽象（可以将EventLoopGroup理解为一个线程池，EventLoop理解为线程池中的线程，Netty为了更好的利用多核CPU的资源，会有多个EventLoop 同时工作，每个EventLoop 维护着一个Selector 实例。

EventLoopGroup提供next接口，可以从组中按照一定规则获取EventLoop 来处理任务，在Netty服务端编程中我们需要提供两个EventLoopGroup：BossEventLoopGroup 和 WorkerEventLoopGroup , BossEventLoop负责接收处理客户端连接，将SocketChannel 封装成 NioSocketChannel 交给 WorkerEventLoop 进行处理，WorkerEventLoop只负责处理读写事件；

一般情况下我们都是用实现类NioEventLoopGroup:

常用方法：

// 构造方法,创建线程组
public NioEventLoopGroup();

// 断开连接，关闭线程
public Future<?> shutdownGracefully();

ServerBootstrap和Bootstrap
ServerBootstrap 是 Netty 中的服务器端启动助手，通过它可以完成服务器端的各种配置； Bootstrap 是 Netty 中的客户端启动助手，通过它可以完成客户端的各种配置。

常用方法：

# 该方法用于服务器端，用来设置两个 EventLoop
# 其中：parentGroup  对应的就是：BossEventLoopGroup
#	   childGroup   对应的就是：WorkerEventLoopGroup
public ServerBootstrap group(EventLoopGroup parentGroup, EventLoopGroup childGroup)
    
    
#该方法用于客户端，用来设置一个 EventLoop
public B group(EventLoopGroup group)
    
#该方法用来设置一个服务器端的通道 实现
public B channel(Class<? extends C> channelClass)
    
#用来给 ServerChannel 添加配置
public B option(ChannelOption option, T value)
    
#用来给接收到的通道添加配置
public ServerBootstrap childOption(ChannelOption childOption, T value)

    
#该方法用来设置业务 处理类（自定义的 handler)
public ServerBootstrap childHandler(ChannelHandler childHandler)
    
# 该方法用于服务器端，用来设置占用的端口号
public ChannelFuture bind(int inetPort)
    
# 该方法用于客户端，用来连 接服务器端
public ChannelFuture connect(String inetHost, int inetPort)

Unpooled类（缓冲区Buffer工具类）
这是 Netty 提供的一个专门用来操作缓冲区的工具类，
常用方法如下：

#通过给定的数据 和 字符编码返回一个 ByteBuf 对象（类似于 NIO 中的 ByteBuffer 对象）
public static ByteBuf copiedBuffer(CharSequence string, Charset charset)

Netty入门案例

引入Netty依赖坐标

        <dependency>
            <groupId>io.netty</groupId>
            <artifactId>netty-all</artifactId>
            <version>4.1.69.Final</version>
        </dependency>
        

服务端实现

package cn.hu.netty.primer;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;

/**
 * @Author: hu.chen
 * @Description: Netty 入门案例，服务端实现
 * @DateTime: 2021/12/23 1:08 PM
 **/
public class NettyServer {

    public static void main(String[] args) throws InterruptedException {
        //1： 创建BossGroup线程组  --- 用来处理网络连接事件
        //设置 bossGroup 线程的个数，这个一般设置为1个
        // 如果不设置，默认是：你的电脑的处理器线程数*2
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);

        //2： 创建 WorkerGroup线程组 ---- 用来处理网络读写事件
        //设置 workerGroup 线程的个数，这个在double中设置的是200
        // 如果不设置，默认是：你的电脑的处理器线程数*2
        EventLoopGroup workerGroup = new NioEventLoopGroup(20);

        //3： 创建服务端的启动助手
        ServerBootstrap serverBootstrap = new ServerBootstrap();

        //4： 将两个线程组绑定到 ServerBootstrap 中
        serverBootstrap.group(bossGroup,workerGroup);

        //5： 设置服务端通道为：NIO
        serverBootstrap.channel(NioServerSocketChannel.class);

        //6：设置相应的参数--设置bossGroup 等待队列的大小
        serverBootstrap.option(ChannelOption.SO_BACKLOG,128);

        //6：设置相应的参数--设置workerGroup 开启连接的探活
        serverBootstrap.childOption(ChannelOption.SO_KEEPALIVE,true);

        //7：创建一个通道初始化对象
        serverBootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
            @Override
            protected void initChannel(SocketChannel channel) {
                //8：向pipeline中添加自定义业务处理handier
                channel.pipeline().addLast(new MyServerHandler());
            }
        });

        //9：启动服务端，并绑定端口,同时将异步改为同步
        ChannelFuture channelFuture = serverBootstrap.bind(8989).sync();

        System.err.println("Netty服务端启动成功");
        //10：关闭通道--(并不是真正意义上的关闭，而是监听通道关闭的状态）
        channelFuture.channel().closeFuture().sync();
        //11: 关闭连接池
        bossGroup.shutdownGracefully();
        workerGroup.shutdownGracefully();
    }
}

服务端业务类

package cn.hu.netty.primer;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;

import java.nio.charset.StandardCharsets;

/**
 * @Author: hu.chen
 * @Description: 服务端业务处理Handler
 * @DateTime: 2021/12/23 1:39 PM
 **/
public class MyServerHandler extends ChannelInboundHandlerAdapter {

    /**
     *  通道读取数据事件
     * @param ctx
     * @param msg 客户端发送过来的消息
     */
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg){
        // 将消息进行转换
        ByteBuf byteBuf= (ByteBuf) msg;
        System.err.println("客户端说："+byteBuf.toString(StandardCharsets.UTF_8));
    }


    /**
     * 通道业务处理完毕事件 (当我们的业务逻辑处理完成后会调用此方法）可以在这里给客户端进行响应
     * @param ctx 上下文对象
     */
    @Override
    public void channelReadComplete(ChannelHandlerContext ctx){
        // 给客户端响应消息
        ctx.writeAndFlush(Unpooled.copiedBuffer("你好我是Netty服务端".getBytes(StandardCharsets.UTF_8)));
    }


    /**
     * 通道发生异常
     * @param ctx 上下文
     * @param cause 异常对象
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause){

        System.err.println("出现异常了。。。。。。。。。。。");
        // 打印异常信息
        cause.printStackTrace();

        // 关闭通道
        ctx.close();
    }
}

客户端实现

package cn.hu.netty.primer;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;

/**
 * @Author: hu.chen
 * @Description: Netty 客户端代码
 * @DateTime: 2021/12/23 2:01 PM
 **/
public class NettyClient {

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

        //1：创建线程组
        EventLoopGroup eventExecutors = new NioEventLoopGroup(5);

        //2：创建客户端启动助手
        Bootstrap bootstrap=new Bootstrap();

        //3：设置线程组
        bootstrap.group(eventExecutors);

        //5： 设置客户端通道为：NIO
        bootstrap.channel(NioSocketChannel.class);

        //6：创建一个通道初始化对象
        bootstrap.handler(new ChannelInitializer<SocketChannel>() {
            @Override
            protected void initChannel(SocketChannel channel) throws Exception {
                //8：向pipeline中添加自定义业务处理handier
                channel.pipeline().addLast(new MyClientandler());
            }
        });

        //7: 启动客户端绑定端口，同时将异步改为同步
        ChannelFuture channelFuture = bootstrap.connect("127.0.0.1", 8989).sync();

        //8：关闭通道
        channelFuture.channel().closeFuture().sync();

        //9：关闭连接池
        eventExecutors.shutdownGracefully();
    }
}

客户端业务类

package cn.hu.netty.primer;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;

import java.nio.charset.StandardCharsets;

/**
 * @Author: hu.chen
 * @Description: 客户端业务处理Handler
 * @DateTime: 2021/12/23 1:39 PM
 **/
public class MyClientandler extends ChannelInboundHandlerAdapter {


    /**
     *  通道就绪事件
     * @param ctx 上下文
     */
    @Override
    public void channelActive(ChannelHandlerContext ctx){

        // 向服务端发送数据
        ctx.writeAndFlush(Unpooled.copiedBuffer("你好，我是Netty客户端".getBytes(StandardCharsets.UTF_8)));

    }

    /**
     *  通道读取数据事件
     * @param ctx
     * @param msg 客户端发送过来的消息
     */
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg){
        // 将消息进行转换
        ByteBuf byteBuf= (ByteBuf) msg;
        System.err.println("服务端说："+byteBuf.toString(StandardCharsets.UTF_8));


        // 本次通话结束，关闭通道
        ctx.close();
    }


    /**
     * 通道发生异常
     * @param ctx 上下文
     * @param cause 异常对象
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause){

        System.err.println("出现异常了。。。。。。。。。。。");
        // 打印异常信息
        cause.printStackTrace();

        // 关闭通道
        ctx.close();
    }
}