Netty Review - 核心组件扫盲

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

package com.artisan.netty4.server;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.util.CharsetUtil;

/**
 * @author 小工匠
 * @version 1.0
 * @description: 自定义的Handler需要继承Netty规定好的HandlerAdapter才能被Netty框架所关联
 * @mark: show me the code , change the world
 */
@ChannelHandler.Sharable
public class ArtisanServerHandler extends ChannelInboundHandlerAdapter {

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        //获取客户端发送过来的消息
        ByteBuf byteBuf = (ByteBuf) msg;
        System.out.println("收到客户端" + ctx.channel().remoteAddress() + "发送的消息：" + byteBuf.toString(CharsetUtil.UTF_8));
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        //发送消息给客户端
        ctx.writeAndFlush(Unpooled.copiedBuffer(">>>>>>msg sent from server 2 client.....", CharsetUtil.UTF_8));
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        //发生异常，关闭通道
        ctx.close();
    }
}
    ```



【启动类 】

```java

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

    // 创建两个线程组

    EventLoopGroup bossGroup = new NioEventLoopGroup();

    EventLoopGroup workerGroup = new NioEventLoopGroup();

    try {

        // 创建服务端的启动对象，设置参数

        ServerBootstrap serverBootstrap = new ServerBootstrap();

        // 设置两个线程组

        serverBootstrap.group(bossGroup, workerGroup)

                // 设置服务端通道类型实现

                .channel(NioServerSocketChannel.class)

                // 设置bossGroup线程队列的连接个数

                .option(ChannelOption.SO_BACKLOG, 128)

                // 设置workerGroup保持活动连接状态

                .childOption(ChannelOption.SO_KEEPALIVE, true)

                // 使用匿名内部类的形式初始化通道对象

                .childHandler(new ChannelInitializer<SocketChannel>() {

                    @Override

                    protected void initChannel(SocketChannel socketChannel) throws Exception {

                        // 给pipeline管道设置处理器

                        socketChannel.pipeline().addLast(new ArtisanServerHandler());

                    }

                });// 给workerGroup的EventLoop对应的管道设置处理器

        System.out.println("服务端已经准备就绪...");

        // 绑定端口，启动服务

        ChannelFuture channelFuture = serverBootstrap.bind(9999).sync();

        // 对关闭通道进行监听

        channelFuture.channel().closeFuture().sync();

    } finally {

        bossGroup.shutdownGracefully();

        workerGroup.shutdownGracefully();

    }

}

----

### Client

【Handler 】

```javascript

@Override

public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {

    //接收服务端发送过来的消息

    ByteBuf byteBuf = (ByteBuf) msg;

    System.out.println("收到服务端" + ctx.channel().remoteAddress() + "的消息：" + byteBuf.toString(CharsetUtil.UTF_8));

}

【启动类 】

```javascript

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

    NioEventLoopGroup eventExecutors = new NioEventLoopGroup();

    try {

        //创建bootstrap对象，配置参数

        Bootstrap bootstrap = new Bootstrap();

        //设置线程组

        bootstrap.group(eventExecutors)

                //设置客户端的通道实现类型

                .channel(NioSocketChannel.class)

                //使用匿名内部类初始化通道

                .handler(new ChannelInitializer<SocketChannel>() {

                    @Override

                    protected void initChannel(SocketChannel ch) throws Exception {

                        //添加客户端通道的处理器

                        ch.pipeline().addLast(new ArtisanClientHandler());

                    }

                });

        System.out.println("客户端准备就绪");

        //连接服务端

        ChannelFuture channelFuture = bootstrap.connect("127.0.0.1", 9999).sync();

        //对通道关闭进行监听

        channelFuture.channel().closeFuture().sync();

    } finally {

        //关闭线程组

        eventExecutors.shutdownGracefully();

    }

}

先启动服务端，再启动客户端

![请在此添加图片描述](https://developer.qcloudimg.com/http-save/yehe-8916337/5f94153e197f152b565ba087520a00af.png?qc_blockWidth=960&qc_blockHeight=211)

![请在此添加图片描述](https://developer.qcloudimg.com/http-save/yehe-8916337/c2f56173de76b4297036d665b1b39540.png?qc_blockWidth=970&qc_blockHeight=196)

----

## Netty 重要组件

### taskQueue任务队列

如果Handler处理器有一些长时间的业务处理，可以交给taskQueue异步处理。

我们在`ArtisanServerHandler#channelRead`中添加如下代码

```javascript

@Override

public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {

    //获取客户端发送过来的消息

    ByteBuf byteBuf = (ByteBuf) msg;

    System.out.println("收到客户端" + ctx.channel().remoteAddress() + "发送的消息：" + byteBuf.toString(CharsetUtil.UTF_8));

    //获取到线程池eventLoop，添加线程，执行

    ctx.channel().eventLoop().execute(() -> {

        //长时间操作，不至于长时间的业务操作导致Handler阻塞

        try {

            Thread.sleep(1000);

        } catch (InterruptedException e) {

            throw new RuntimeException(e);

        }

        System.out.println(Thread.currentThread().getName() + " - 长时间的业务处理");

    });

}

![请在此添加图片描述](https://developer.qcloudimg.com/http-save/yehe-8916337/7cdb698f5f615df5a990211c4f03f498.png?qc_blockWidth=1431&qc_blockHeight=901)

![请在此添加图片描述](https://developer.qcloudimg.com/http-save/yehe-8916337/bd368dd84f3e1515b769045efd3136c0.png?qc_blockWidth=1431&qc_blockHeight=983)

----

### scheduleTaskQueue延时任务队列

![请在此添加图片描述](https://developer.qcloudimg.com/http-save/yehe-8916337/84479598da5bc5dc979d0a8089bcc989.png?qc_blockWidth=1431&qc_blockHeight=951)

![请在此添加图片描述](https://developer.qcloudimg.com/http-save/yehe-8916337/ddf86a2f742ee252567262d79e5b19e6.png?qc_blockWidth=1431&qc_blockHeight=969)

----

### Future异步机制

```javascript

这个ChannelFuture对象是用来做什么的呢？

ChannelFuture提供操作完成时一种异步通知的方式。一般在Socket编程中，等待响应结果都是同步阻塞的，而Netty则不会造成阻塞，因为ChannelFuture是采取类似观察者模式的形式进行获取结果。

请看一段代码演示：

```javascript

            @Override

            public void operationComplete(ChannelFuture channelFuture) throws Exception {

                if (channelFuture.isSuccess()) {

                    System.out.println("连接成功");

                } else {

                    System.out.println("连接失败");

                }

            }

        });

----

### Bootstrap与ServerBootStrap

![请在此添加图片描述](https://developer.qcloudimg.com/http-save/yehe-8916337/ab7f2bfaf39b9ac49292fd3ece30b7c1.png?qc_blockWidth=713&qc_blockHeight=472)

都是继承于`AbstractBootStrap`抽象类，所以大致上的配置方法都相同。

一般来说，使用Bootstrap创建启动器的步骤可分为以下几步：

![请在此添加图片描述](https://developer.qcloudimg.com/http-save/yehe-8916337/b7292d6c9924a9299f8eb949f2a31cd6.png?qc_blockWidth=1431&qc_blockHeight=1270)

----

### group()

```javascript

- bossGroup 用于监听客户端连接，**专门负责与客户端创建连接，并把连接注册到workerGroup的Selector中**。
- workerGroup用于处理每一个连接发生的读写事件

一般创建线程组直接new：

```javascript

默认线程数cpu核数的两倍 。 在`MultithreadEventLoopGroup`定义 `NettyRuntime.availableProcessors() * 2`

```javascript

static {

    DEFAULT_EVENT_LOOP_THREADS = Math.max(1, SystemPropertyUtil.getInt(

            "io.netty.eventLoopThreads", NettyRuntime.availableProcessors() * 2));

    if (logger.isDebugEnabled()) {

        logger.debug("-Dio.netty.eventLoopThreads: {}", DEFAULT_EVENT_LOOP_THREADS);

    }

}

通过源码可以看到，默认的线程数是cpu核数的两倍。假设想自定义线程数，可以使用有参构造器：

```javascript

----

### channel()

这个方法用于设置通道类型，当建立连接后，会根据这个设置创建对应的Channel实例。

-  `NioSocketChannel`： **异步非阻塞的客户端 TCP Socket 连接**。
 
-  `NioServerSocketChannel`： **异步非阻塞的服务器端 TCP Socket 连接**。
 

常用的就是这两个通道类型，因为是异步非阻塞的。所以是首选。

----

-  `OioSocketChannel`： 同步阻塞的客户端 TCP Socket 连接 **(已废弃)**。
 
-  `OioServerSocketChannel`： 同步阻塞的服务器端 TCP Socket 连接 **(已废弃)** 。
 

```javascript

    .channel(OioServerSocketChannel.class)//设置服务端通道实现类型

    .channel(OioSocketChannel.class)//设置客户端的通道实现类型

-  `NioSctpChannel`： 异步的客户端 Sctp（Stream Control Transmission Protocol，流控制传输协议）连接。
 
-  `NioSctpServerChannel`： 异步的 Sctp 服务器端连接。
 只能在linux环境下才可以启动
 

### option()与childOption()

-   `option()`设置的是服务端用于接收进来的连接，也就是boosGroup线程。
 
-   `childOption()`是提供给父管道接收到的连接，也就是workerGroup线程。
 

列举一下常用的参数

`SocketChannel`参数，也就是childOption()常用的参数：

- SO\_RCVBUF Socket参数，TCP数据接收缓冲区大小。
- TCP\_NODELAY TCP参数，立即发送数据，默认值为Ture。
- SO\_KEEPALIVE Socket参数，连接保活，默认值为False。启用该功能时，TCP会主动探测空闲连接的有效性。

`ServerSocketChannel`参数，也就是option()常用参数：

- SO\_BACKLOG Socket参数，服务端接受连接的队列长度，如果队列已满，客户端连接将被拒绝。默认值，Windows为200，其他为128。

### ChannelPipeline

 `ChannelPipeline`是Netty处理请求的责任链，`ChannelHandler`则是具体处理请求的处理器。实际上每一个channel都有一个处理器的流水线

**在Bootstrap中childHandler()方法需要初始化通道，实例化一个ChannelInitializer，这时候需要重写initChannel()初始化通道的方法，装配流水线就是在这个地方进行**。

代码演示如下：

```javascript

@Override

protected void initChannel(SocketChannel socketChannel) throws Exception {

    //给pipeline管道设置自定义的处理器

    socketChannel.pipeline().addLast(new MyServerHandler());

}

处理器Handler主要分为两种：

-   `ChannelInboundHandlerAdapter`(入站处理器): 入站指的是数据从底层java NIO Channel到Netty的Channel。
 
-   `ChannelOutboundHandler`(出站处理器) ：出站指的是通过Netty的Channel来操作底层的java NIO Channel
 

`ChannelInboundHandlerAdapter`处理器常用的事件有：

- 注册事件 `fireChannelRegistered`。
- 连接建立事件 `fireChannelActive`。
- 读事件和读完成事件 `fireChannelRead`、`fireChannelReadComplete`。
- 异常通知事件 `fireExceptionCaught`。
- 用户自定义事件 `fireUserEventTriggered`。
- Channel 可写状态变化事件 `fireChannelWritabilityChanged`。
- 连接关闭事件 `fireChannelInactive`。

`ChannelOutboundHandler`处理器常用的事件有：

- 端口绑定 `bind`。
- 连接服务端 `connect`。
- 写事件 `write`。
- 刷新时间 `flush`。
- 读事件 `read`。
- 主动断开连接 `disconnect`。
- 关闭 channel 事件 `close`
- 还有一个类似的`handler()`，主要用于装配parent通道，也就是bossGroup线程。一般情况下，都用不上这个方法

----

### bind()

**提供用于服务端或者客户端绑定服务器地址和端口号，默认是异步启动。如果加上sync()方法则是同步**。

有五个同名的重载方法，作用都是用于绑定地址端口号。

![请在此添加图片描述](https://developer.qcloudimg.com/http-save/yehe-8916337/d5c52dd4e35949288c9344267f21707f.png?qc_blockWidth=928&qc_blockHeight=259)

----

### 优雅地关闭EventLoopGroup

```javascript

会关闭所有的child Channel。关闭之后，释放掉底层的资源。

----

### Channle

#### Channel是什么

```javascript

翻译大意：一种连接到网络套接字或能进行读、写、连接和绑定等I/O操作的组件。

```javascript

channel为用户提供：

- 通道当前的状态（例如它是打开？还是已连接？）
- channel的配置参数（例如接收缓冲区的大小）
- channel支持的IO操作（例如读、写、连接和绑定），以及处理与channel相关联的所有IO事件和请求的ChannelPipeline。

----

#### 获取channel的状态

![请在此添加图片描述](https://developer.qcloudimg.com/http-save/yehe-8916337/90fbe797e59826ccc774ba8914bc31f0.png?qc_blockWidth=926&qc_blockHeight=721)

```javascript

以上就是获取channel的四种状态的方法。

----

#### 获取channel的配置参数

![请在此添加图片描述](https://developer.qcloudimg.com/http-save/yehe-8916337/9c5dd39ceb14e3870bb1201b71561b0f.png?qc_blockWidth=931&qc_blockHeight=719)

获取单条配置信息，使用getOption()， ：

```javascript

获取多条配置信息，使用getOptions() ：

```javascript

 System.out.println("Key = " + entry.getKey() + ", Value = " + entry.getValue());

输出

```javascript

完整代码如下

```javascript

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

    // 创建两个线程组

    EventLoopGroup bossGroup = new NioEventLoopGroup();

    EventLoopGroup workerGroup = new NioEventLoopGroup();

    try {

        // 创建服务端的启动对象，设置参数

        ServerBootstrap serverBootstrap = new ServerBootstrap();

        // 设置两个线程组

        serverBootstrap.group(bossGroup, workerGroup)

                // 设置服务端通道类型实现

                .channel(NioServerSocketChannel.class)

                // 设置bossGroup线程队列的连接个数

                .option(ChannelOption.SO_BACKLOG, 128)

                // 设置workerGroup保持活动连接状态

                .childOption(ChannelOption.SO_KEEPALIVE, true)

                // 使用匿名内部类的形式初始化通道对象

                .childHandler(new ChannelInitializer<SocketChannel>() {

                    @Override

                    protected void initChannel(SocketChannel socketChannel) throws Exception {

                        // 给pipeline管道设置处理器

                        socketChannel.pipeline().addLast(new ArtisanServerHandler());

                    }

                });// 给workerGroup的EventLoop对应的管道设置处理器

        System.out.println("服务端已经准备就绪...");

        // 绑定端口，启动服务

        ChannelFuture channelFuture = serverBootstrap.bind(9999).sync();

        channelFuture.addListener(new ChannelFutureListener() {

            @Override

            public void operationComplete(ChannelFuture channelFuture) throws Exception {

                if (channelFuture.isSuccess()) {

                    System.out.println("连接成功");

                } else {

                    System.out.println("连接失败");

                }

            }

        });

        // 获取单个配置信息

        Integer option = channelFuture.channel().config().getOption(ChannelOption.SO_BACKLOG);

        System.out.println(option);

        // 获取多条配置信息

        Map<ChannelOption<?>, Object> options = channelFuture.channel().config().getOptions();

        for (Map.Entry<ChannelOption<?>, Object> entry : options.entrySet()) {

            System.out.println("Key = " + entry.getKey() + ", Value = " + entry.getValue());

        }

        // 对关闭通道进行监听

        channelFuture.channel().closeFuture().sync();

    } finally {

        bossGroup.shutdownGracefully();

        workerGroup.shutdownGracefully();

    }

}

----

#### channel支持的IO操作

##### 写操作

这里演示从服务端写消息发送到客户端

```javascript

@Override

public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {

    //发送消息给客户端

    ctx.writeAndFlush(Unpooled.copiedBuffer(">>>>>>msg sent from server 2 client.....", CharsetUtil.UTF_8));

}

![请在此添加图片描述](https://developer.qcloudimg.com/http-save/yehe-8916337/7bdf0a8451af920a311f533884bde6c4.png?qc_blockWidth=1366&qc_blockHeight=523)

----

##### 连接操作

```javascript

----

##### 通过channel获取ChannelPipeline，并做相关的处理：

```javascript

----

### Selector

Netty中的Selector也和NIO的Selector是一样的，就是**用于监听事件，管理注册到Selector中的channel，实现多路复用器**。

![请在此添加图片描述](https://developer.qcloudimg.com/http-save/yehe-8916337/b7b8e99a364ba4314fbee1120d68085e.png?qc_blockWidth=1431&qc_blockHeight=1320)

### PiPeline与ChannelPipeline

![请在此添加图片描述](https://developer.qcloudimg.com/http-save/yehe-8916337/1e4cba4b171f0557fa70d888bd13c341.png?qc_blockWidth=1421&qc_blockHeight=454)

我们知道可以在channel中装配ChannelHandler流水线处理器，那一个channel不可能只有一个channelHandler处理器，肯定是有很多的，既然是很多channelHandler在一个流水线工作，肯定是有顺序的。

于是pipeline就出现了，pipeline相当于处理器的容器。初始化channel时，把channelHandler按顺序装在pipeline中，就可以实现按序执行channelHandler了。

在一个Channel中，只有一个ChannelPipeline。该pipeline在Channel被创建的时候创建。ChannelPipeline包含了一个ChannelHander形成的列表，且所有ChannelHandler都会注册到ChannelPipeline中。

----

### ChannelHandlerContext

![请在此添加图片描述](https://developer.qcloudimg.com/http-save/yehe-8916337/d4328540b45c02b8b89a018e523a5dfc.png?qc_blockWidth=540&qc_blockHeight=230)


 在Netty中，Handler处理器是由我们定义的，上面讲过通过集成入站处理器或者出站处理器实现。这时如果我们想在Handler中获取pipeline对象，或者channel对象，怎么获取呢。

于是Netty设计了这个ChannelHandlerContext上下文对象，就可以拿到channel、pipeline等对象，就可以进行读写等操作。

通过类图，ChannelHandlerContext是一个接口，下面有三个实现类。

![请在此添加图片描述](https://developer.qcloudimg.com/http-save/yehe-8916337/5bf1568eefcb3ef8040cbe5730bf9f05.png?qc_blockWidth=1360&qc_blockHeight=390)

实际上ChannelHandlerContext在pipeline中是一个链表的形式

```javascript

this.channel = ObjectUtil.checkNotNull(channel, "channel");

succeededFuture = new SucceededChannelFuture(channel, null);

voidPromise =  new VoidChannelPromise(channel, true);

//设置头结点head，尾结点tail

tail = new TailContext(this);

head = new HeadContext(this);

head.next = tail;

tail.prev = head;