netty源码之NioEventLoop

Posted 2021-08-28 better_hui

目录

一、初始化

1、NioEventLoopGroup最终都是调用父类MultithreadEventLoopGroup的构造器

2、通过父类MultiThreadEventExecutorGroup构造器创建NioEventLoopGroup实例，内部维护了一个SingleThreadEventExecutor类型的数组，通过newChild()方法进行实例化

3、newChild()方法实例化的是NioEventLoop类型的EventLoop对象

4、NioEventLoop 打开一个selector

二、EventLoop 与 Channel的关系

1、客户端在通过bootstrap启动时会创建一个Channel实例并进行初始化，在初始化过程中会绑定一个NioEventLoop

2、NioEventLoopGroup调用父类的register()方法将获取一个EventLoop与通道绑定

3、最终调用了 AbstractChannel#AbstractUnsafe.register 后完成了 Channel 和 EventLoop 的关联，将获取的EventLoop值赋值给AbstractChannel内的一个eventLoop属性

4、调用AbstractChannel的register0()方法最终完成通道的注册

---

一、初始化

    ServerBootstrap b = new ServerBootstrap();
    
    EventLoopGroup bossGroup = new NioEventLoopGroup();    //创建多线程线程池
    
    EventLoopGroup workGroup = new NioEventLoopGroup();    //创建多线程线程池
    
    b.group(bossGroup,workGroup); 
    

服务端的acceptor ,处理IO事件分两种：监听请求建立连接 和 处理连接 。

在reactor多线程模型下 ， 两种事件都是有同一个线程处理。

在reactor主从线程模型下 ， bossGroup负责处理监听连接事件 ， workerGroup负责处理连接

1、NioEventLoopGroup最终都是调用父类MultithreadEventLoopGroup的构造器

    protected MultithreadEventLoopGroup(int nThreads, ThreadFactory threadFactory, Object... args) {
        super(nThreads == 0? DEFAULT_EVENT_LOOP_THREADS : nThreads, threadFactory, args);
    }

2、通过父类MultiThreadEventExecutorGroup构造器创建NioEventLoopGroup实例，内部维护了一个SingleThreadEventExecutor类型的数组，通过newChild()方法进行实例化

    protected MultithreadEventExecutorGroup(int nThreads, Executor executor,
                                                EventExecutorChooserFactory chooserFactory, Object... args) {
            // 创建一个线程池
            executor = new ThreadPerTaskExecutor(newDefaultThreadFactory());
            // 创建一个NioEventLoop数组
            children = new EventExecutor[nThreads];
    
            for (int i = 0; i < nThreads; i ++) {
                children[i] = newChild(executor, args);
            }
            // 生成一个线程选择器
            chooser = chooserFactory.newChooser(children);
    }

3、newChild()方法实例化的是NioEventLoop类型的EventLoop对象

NioEventLoop类继承了SingleThreadEventExecutor类 ， 该类种有一个thread属性 ，用以绑定本地线程

    protected EventLoop newChild(Executor executor, Object... args) throws Exception {
            return new NioEventLoop(this, executor, (SelectorProvider) args[0],
                ((SelectStrategyFactory) args[1]).newSelectStrategy(), (RejectedExecutionHandler) args[2]);
    }
    
    
    

4、NioEventLoop 打开一个selector

    NioEventLoop(NioEventLoopGroup parent, Executor executor, SelectorProvider selectorProvider,
                 SelectStrategy strategy, RejectedExecutionHandler rejectedExecutionHandler) {
        super(parent, executor, false, DEFAULT_MAX_PENDING_TASKS, rejectedExecutionHandler);
        if (selectorProvider == null) {
            throw new NullPointerException("selectorProvider");
        }
        if (strategy == null) {
            throw new NullPointerException("selectStrategy");
        }
        provider = selectorProvider;
        final SelectorTuple selectorTuple = openSelector();
        selector = selectorTuple.selector;
        unwrappedSelector = selectorTuple.unwrappedSelector;
        selectStrategy = strategy;
    }

二、EventLoop 与 Channel的关系

1、客户端在通过bootstrap启动时会创建一个Channel实例并进行初始化，在初始化过程中会绑定一个NioEventLoop

        -> AbstractBootstrap.initAndRegister()
        -> channelFactory.newChannel()  
        -> group().register(channel) 
        -> MultiThreadEventLoopGroup.regiser()

2、NioEventLoopGroup调用父类的register()方法将获取一个EventLoop与通道绑定

    public ChannelFuture register(Channel channel) {
        return this.next().register(channel);
    }

3、最终调用了 AbstractChannel#AbstractUnsafe.register 后完成了 Channel 和 EventLoop 的关联，将获取的EventLoop值赋值给AbstractChannel内的一个eventLoop属性

    public final void register(EventLoop eventLoop, final ChannelPromise promise) {
        // 删除条件检查.
        ...
        AbstractChannel.this.eventLoop = eventLoop;
    
        if (eventLoop.inEventLoop()) {
            register0(promise);
        } else {
            try {
                eventLoop.execute(new OneTimeTask() {
                    @Override
                    public void run() {
                        register0(promise);
                    }
                });
            } catch (Throwable t) {
                ...
            }
        }
    }

4、调用AbstractChannel的register0()方法最终完成通道的注册

    private void register0(ChannelPromise promise) {
        boolean firstRegistration = neverRegistered;
        doRegister();
        neverRegistered = false;
        registered = true;
        safeSetSuccess(promise);
        pipeline.fireChannelRegistered();
        // Only fire a channelActive if the channel has never been registered. This prevents firing
        // multiple channel actives if the channel is deregistered and re-registered.
        if (firstRegistration && isActive()) {
            pipeline.fireChannelActive();
        }
    }
    
    
    protected void doRegister() throws Exception {
        // 省略错误处理
        selectionKey = javaChannel().register(eventLoop().selector, 0, this);
    }