【netty系列-08】深入Netty组件底层原理和基本实现

Netty系列整体栏目

内容	链接地址
【一】深入理解网络通信基本原理和tcp/ip协议	https://zhenghuisheng.blog.csdn.net/article/details/136359640
【二】深入理解Socket本质和BIO	https://zhenghuisheng.blog.csdn.net/article/details/136549478
【三】深入理解NIO的基本原理和底层实现	https://zhenghuisheng.blog.csdn.net/article/details/138451491
【四】深入理解反应堆模式的种类和具体实现	https://zhenghuisheng.blog.csdn.net/article/details/140113199
【五】深入理解直接内存与零拷贝	https://zhenghuisheng.blog.csdn.net/article/details/140721001
【六】select、poll和epoll多路复用的区别	https://zhenghuisheng.blog.csdn.net/article/details/140795733
【七】深入理解和使用Netty中组件	https://zhenghuisheng.blog.csdn.net/article/details/141166098
【八】深入Netty组件底层原理和基本实现	https://zhenghuisheng.blog.csdn.net/article/details/141685088

深入Netty组件底层原理和基本实现

一，深入理解netty组件
- 1，EventLoopGroup的组成原理
- 2，EventLoopGroup组件
- 3，channel组件
- 4，ChannelPipeline组件
- 5，ChannelHandlerContext
- 6，ChannelHandler以及对应适配器
- - 6.1，出站的read事件
  - 6.2，实现Handler共享

一，深入理解netty组件

在上一篇中讲解了netty的基本使用，在代码中用到了多个组件。如BootStarp，EventLoopGroup，以及NioServerSocketChannel，handler以及pipeline等。接下来这篇主要是对这些组再件做一个详细的解释

1，EventLoopGroup的组成原理

如下图所示，一个 EventLoopGroup 可以管理多个 EventLoop ，每一个EventLoop都会对应一个线程，EventLoop会管理所有对应的channel，channel就是封装的socket，一个channel只会对应一个EventLoop，内部需要触发或者执行什么事件都得是通过相应的EventLoop进行管理
在这里插入图片描述

2，EventLoopGroup组件

在前面大概说了一下，这个EventLoopGroup就是类似于nio中反应堆模式的selector，用于循环的去处理事件，接下来通过本篇文章，详细的描述一下到底什么是EventLoopGroup。还是使用的上一个4.1.42.Final版本，后续都是该版本查看内部的源码以及实现，该接口的组成和基本方法和实现如下

public interface EventLoopGroup extends EventExecutorGroup {
    EventLoop next();

    ChannelFuture register(Channel var1);

    ChannelFuture register(ChannelPromise var1);

    /** @deprecated */
    @Deprecated
    ChannelFuture register(Channel var1, ChannelPromise var2);
}

该接口的父接口的实现图如下，最顶层就是一个任务的实现类，因此在没看具体的源码之前，就能知道这个EventLoop底层应该就是一个线程任务

在这里插入图片描述

EventLoop就是对应一个个线程，去执行多个socket对应的事件或者任务。ChannelPromise就是一个具体的ChannelFuture，通过unsafe方法将这个socket注册到对应的EventLoop中，随后返回。

public ChannelFuture register(ChannelPromise promise) {
    ObjectUtil.checkNotNull(promise, "promise");
    promise.channel().unsafe().register(this, promise);
    return promise;
}

EventLoop的注册实现如下，在socketChannel绑定对应EventLoop时，需要判断EventLoop是否存在，如果存在则通过unsafe方法将这个socket注册到对应的EventLoop中，如果不存在，则将这个 promise 打包成一个任务，丢到线程池中，随后通过这个EventLoop执行

public final void register(EventLoop eventLoop, final ChannelPromise promise) {
            if (eventLoop == null) {
                throw new NullPointerException("eventLoop");
            } else {
                AbstractChannel.this.eventLoop = eventLoop;
                if (eventLoop.inEventLoop()) {
                    this.register0(promise);
                } else {
                    try {
                        eventLoop.execute(new Runnable() {
                            public void run() {
                                //封装成
                                AbstractUnsafe.this.register0(promise);
                            }
                        });
                    } catch (Throwable var4) {
                        ...
                    }
                }

            }
}

可以得知 EventLoop 继承于OrderedEventExecutor，结合OrderedEventExecutor的实现类图分析，可以得知该类的父类就是一个Executor的的线程池

public interface EventLoop extends OrderedEventExecutor, EventLoopGroup {
    EventLoopGroup parent();
}

在这里插入图片描述

以一个具体实现的单例的EventLoop为例，该接口继承了 SingleThreadEventExecutor 任务执行器和 EventLoop 接口

public abstract class SingleThreadEventLoop extends SingleThreadEventExecutor implements EventLoop {
    ...
}

在这个 SingleThreadEventExecutor 抽象类中，我把一些重要的属性和参数列在下面，如一默认最大的线程任务数，存放的任务队列等

//默认的最大线程任务数
static final int DEFAULT_MAX_PENDING_EXECUTOR_TASKS = Math.max(16,
      SystemPropertyUtil.getInt("io.netty.eventexecutor.maxPendingTasks", Integer.MAX_VALUE));
//存放线程的队列
private final Queue<Runnable> taskQueue;
//可见线程
private volatile Thread thread;
//计数器
private final CountDownLatch threadLock = new CountDownLatch(1);

除此之外，还有线程的一些状态等，如未启动，就绪，运行，阻塞，终止等状态

    private static final int ST_NOT_STARTED = 1;
    private static final int ST_STARTED = 2;
    private static final int ST_SHUTTING_DOWN = 3;
    private static final int ST_SHUTDOWN = 4;
    private static final int ST_TERMINATED = 5;

因此EventLoop的线程实现，就是类似于在jdk中的线程池的实现，里面既有线程，又有队列。因此在使用这个Event Loop的流程大致如下：首先会判断当前执行的线程是否为EventLoop线程，如果是则直接将当前的Channel注册到该EventLoop中，这样可以减少这个上下文的切换；如果当前执行的线程不是EventLoop线程，那么就会将这个注册的任务打包成一个队列去完成，有点类似于加入到线程池中去排队异步执行。

为什么要通过队列去通过这个EventLoop去执行这个任务，而不是直接使用主线程去执行任务，原因是为了解决这个并发问题。netty为了解决这个问题，特意指定了每个channel只能由对应的EventLoop去管理和执行，因此就不能由其他线程或者当前主线程去执行注册或者执行任务的事件。每次由一个EventLoop去管理多个channel，这样就可以保证每个channel执行的安全性。

总而言之就是一句话：EventLoopGroup负责管理EventLoop，EventLoop负责管理Channel

3，channel组件

在上面提到了Channel所有的动作和行为都得由对应的EventLoop去触发对应的事件，接下来分析在Netty中的这个Channel的具体实现。在分析之前，不管是Netty的channel还是Nio中ServerSocket，其内部都是对底层的 Socket 进行操作。

public interface Channel extends AttributeMap, ChannelOutboundInvoker, Comparable<Channel> {
    ...
}

在该接口中，其内部的有的方法如下，其部分方法详细描述如下，如是否注册，绑定地址，绑定eventLoop等

在这里插入图片描述

方法	详情
EventLoop eventLoop()	返回与该 `Channel` 关联的 `EventLoop`，负责处理该 `Channel` 的所有 I/O 操作
Channel parent()	返回父 `Channel`，例如，`ServerSocketChannel` 是父 `Channel`
ChannelConfig config()	用于配置 `Channel` 的参数，如 `TCP_NODELAY`, `SO_KEEPALIVE` 等
boolean isOpen()	判断 `Channel` 是否打开（未关闭）。一个打开的 `Channel` 是可以接收和发送数据的
boolean isRegistered()	判断 `Channel` 是否已经注册到 `EventLoop`
ChannelFuture bind(SocketAddress localAddress)	绑定到一个本地地址，用于服务器端 `Channel`
ChannelFuture connect(SocketAddress remoteAddress)	连接到远程地址，用于客户端 `Channel`。
ChannelFuture close()	关闭 `Channel`，释放资源。
Channel read()	请求从 `Channel` 中读取数据，通常是由 `ChannelHandler` 自动触发的。
ChannelFuture write(Object msg)	向 `Channel` 中写入数据，`writeAndFlush()` 会立即将消息刷出到远程对端。
ChannelPipeline pipeline()	返回与 `Channel` 关联的 `ChannelPipeline`

4，ChannelPipeline组件

channelPipeline是用于存放channelhandler的容器，每一个channel都有一个自身对应的channelPipeline。

public interface ChannelPipeline
        extends ChannelInboundInvoker, ChannelOutboundInvoker, Iterable<Entry<String, ChannelHandler>> {}

依旧是如同上面的channel一样，先了解一下这个channelPipeline接口部分api的使用

在这里插入图片描述

方法	详情
addFirst / addLast	将Handler处理器添加到头部或者尾部
remove / removelast / removefirst	将Handler移除 / 移除首个 / 者最后一个
get / first() / last()	获取任意一个 / 获取第一个 / 获取最后一个
fireChannelRegistered	入站事件，入站handler处理器注册事件
fireChannelRead	入站事件，入站handler处理器读取事件
bind /connect /write /flush	出站事件，handler绑定、连接、写、刷新事件

在 ChannelPipeline 中，支持出站事件和入站事件，顾名思义，对应的就是接收并处理请求以及处理并想要请求。

举个例子，如客户端想服务端发起请求，然后向服务端发送gzip压缩的base64编码的数据，服务端需要获取并解压，解码数据，然后做出对应的响应，然后将数据就行base64编码，再gzip压缩。入站事件就是接收请求，gzip解压，base64解码；出战事件就是做出响应，base64编码，gzip压缩。每一个事件对应的就是一个Handler，需要在对应的 ChannelHandler 中编写具体的事件

在这里插入图片描述

5，ChannelHandlerContext

在pipeline中，内部采用的是双向链表的结构，除了灵活的插入和删除的操作之外呢，最主要的是可以在一个pipeline中支持出站和入站事件，这样在入站时可以从前往后的遍历所有Handler结点，出站时可以从后往前的遍历所有Handler的结点，这样就支持双向遍历。除此之外，由于内部采用的是责任链模式，在执行next结点或者指定结点时，支持直接从当前结点往前找或者往后找，不需要每次都从前往后找。因此双向链表的优势远大于单向链表。

当然这个双向链表是如何实现的呢，其实他也是借助了Lisked链表的方式，具体的实体数据存在列表中，前驱指针和后驱指针等存放在Node节点中。在netty中，使用了 Context 上下文的方式存储着对应的结点，例如 AbstractChannelHandlerContext 抽象类中，就定义类next和prev。

abstract class AbstractChannelHandlerContext implements ChannelHandlerContext, ResourceLeakHint {
    volatile AbstractChannelHandlerContext next;
    volatile AbstractChannelHandlerContext prev;
}

上面的gzip解压，base64编码等就是一个具体的入站Handler；base64编码，gzip压缩就是一个具体的出站事件。

在这里插入图片描述

当然在这个 AbstractChannelHandlerContext 抽象类中，这个上下文也不仅仅是维护上下文链表的关系，同时也有数据在这个pipeline中流动。如在上一篇文章中有讲到，通过ctx的 writeAndFlush 写事件将数据写入到上下文中，然后将数据发送给对端。

@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
    ctx.writeAndFlush(Unpooled.copiedBuffer("Hello Netty", CharsetUtil.UTF_8));
}

除了上面这种写数据之外，还能通过以下两种方式将数据写入到上下文中发送给对端。也就是说既可以直接通过本上将数据写入到上下文中发送给对端，也可以通过pipeline或者channel的方式将数据写入到上下文中发送给对端

ctx.pipeline()
ctx.channel()

真正的区别在于是否要遍历整个pipiline中的出站事件。举个例子，依旧是下图三个pipeline，假设入站事件就是一个gzip的解压事件，如果此时解压成功，那么流程继续往下走没问题，如果此时解压失败，那么就会涉及到是否直接将报错返回，还是得继续往下走，把所有的出站事件走一遍的问题。

在这里插入图片描述

如果业务有强制要求，就是说就算报错，也得将报错信息先base64编码然后gzip压缩将数据返回，如果业务没这种要求，那么就可以直接找前面出站handler将事件返回即可。如果是直接通过context将数据写入到上下文的话，那么在发生gzip报错的时候，那么就会直接往前找对应的出站Handler即可，这样可以提高整个流程的效率；如果是使用的pipeline或者channel的话，就算第一步的入站handler出现异常报错，也得从后往前将全部的出站Handler的事件走一遍，再将结果返回，这种方式可以使得整体的返回结果更统一和规范，缺点就是耗时长。当然无论使用哪种方式r，都能体现出使用双向链表的优势。

6，ChannelHandler以及对应适配器

在谈完上面的这些基本的固定组件，在实际开发中，我们最主要写的就是一个个 channelHandler 事件。如前面文章例子中定义了一个实现接口ChannelInboundHandlerAdapter的 NettyServerChannelHandler

public class NettyServerChannelHandler extends ChannelInboundHandlerAdapter {
	...
}

在这里插入图片描述

ChannelInboundHandlerAdapter 顾名思义就是一个均衡的适配器，即实现了入站事件，也实现了出站事件的一些方法。通过适配器方式，可以在开发中只需要去继承这个适配器类，而不需要去就行实现对应的ChannelHandler接口

在该适配器中，只做了一件事情，就是将一些数据或者动作就行传递。里面所有的方法中，都用了fire开头，如fireChannelRegistered

public class ChannelInboundHandlerAdapter extends ChannelHandlerAdapter implements ChannelInboundHandler {
    @Override
    public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
        ctx.fireChannelRegistered();
    }
}

6.1，出站的read事件

一般入站事件中调用read方法，出站事件中调用write写方法。但是在出站的接口中，同时也提供了一个read方法，这就得回归到这个pipeline中的这张图，在pipeline中流转的不仅仅是数据，而且可能是动作。理解这点还是得回归到nio，也就是说客户端先在服务端注册一个感兴趣的事件，然后通过selector轮询器一直去扫描这些事件，当服务端这边有线程空闲的时候就会去触发这个事件，那么就会将这个事件交给感兴趣的线程去操作。

public interface ChannelOutboundHandler extends ChannelHandler {
    ...
    void read(ChannelHandlerContext ctx) throws Exception;
}

在这里插入图片描述

在netty中是通过 EventLoop 去触发这个感兴趣的事件的，那么当事件被触发时，就需要从EventLoop去通知对应的感兴趣的线程，那么这个过程中，eventLoop就类似于一个服务端，对应的感兴趣的线程就是一个服务端，那么就会将这个触发事件的通知打包成一个出站的 ChannelOutboundHandler 事件，因为这个操作时起点是一个发送者，因此在出站事件中，也提供了这个 read 事件。这个事件是一个比较特殊的用法，因为不像上面的数据流动主要针对的是双端的Socket之间的通信，而这个出站的read事件主要是针对服务端内部通过 EventLoop 去通知内部线程去响应对应的事件。

6.2，实现Handler共享

上面谈到了channelHandler会由对应的Eventoop进行管理，因此每一个Handler内部都相互隔离，属于是线程安全的。但是如果有需求需要设计一个共享的handler如何实现，其实在ChannerHandler内部已实现。在 ChannelHandler 接口中，写了一个自定义注解的 Sharable 接口，当在handler上面声明了是共享之后，那么所有的handler都能用这个共享Handler

public interface ChannelHandler {
    ...
        
    @Inherited
    @Documented
    @Target(ElementType.TYPE)
    @Retention(RetentionPolicy.RUNTIME)
    @interface Sharable {
        // no value
    }
}

如在下面的 NettyShareChannelHandler 类中加上这个 @ChannelHandler.Sharable 接口，那么该类就能成为一个共享handler。在该接口中的read方法中，对所有的请求数量进行统计

@ChannelHandler.Sharable
@Slf4j
public class NettyShareChannelHandler extends ChannelDuplexHandler {
    AtomicInteger increment = new AtomicInteger(0);

    @Override
    public void read(ChannelHandlerContext ctx) throws Exception {
        int count = increment.incrementAndGet();
        log.info("接收到的请求总数为:" + count);
        super.read(ctx);
    }
}

在服务端中，也不需要再去手动的new Handler，只需要外部定义好将该对象加入即可。

//创建共享对象
NettyShareChannelHandler nettyShareChannelHandler = new NettyShareChannelHandler();
//部分伪代码
socketChannel.pipeline()
.addLast(nettyShareChannelHandler) 		//加入共享事件
.addLast(new NettyServerChannelHandler());     //将事件加入到管道中