Netty源码-01-NioEventLoopGroup

标签：NioEventLoop Netty nThreads NioEventLoopGroup 源码线程 executor public

一定义

摘自源码JavaDoc

/**
 * The {@link EventExecutorGroup} is responsible for providing the {@link EventExecutor}'s to use
 * via its {@link #next()} method. Besides this, it is also responsible for handling their
 * life-cycle and allows shutting them down in a global fashion.
 *
 */

EventExecutor是NioEventLoop的抽象
NioEventLoopGroup是NioEventLoop管理器
- 管理NioEventLoop事件循环器
- 通过next()方法选择一个NioEventLoop事件循环器使用(NioEventLoop线程)

二类图

截屏2022-11-07 上午10.52.09

从继承关系上可以看出NioEventLoopGroup具有任务管理功能，类似线程池的一种实现。

三源码

1 事件循环器选择

1.1 Demo

package io.netty.example.basic.eventloopgroup;

import io.netty.channel.EventLoop;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;

/**
 *
 * @since 2022/11/7
 * @author dingrui
 */
public class NioEventLoopGroupTest00 {

    public static void main(String[] args) {
        EventLoopGroup group = new NioEventLoopGroup(3);
        for (int i = 0; i < 4; i++) {
            EventLoop el = group.next();
            System.out.println(el);
        }
    }
}

执行结果为

io.netty.channel.nio.NioEventLoop@2ac1fdc4
io.netty.channel.nio.NioEventLoop@5f150435
io.netty.channel.nio.NioEventLoop@1c53fd30
io.netty.channel.nio.NioEventLoop@2ac1fdc4

可以猜测出来两点

NioEventLoopGroup构造方法形参可以控制定义EventLoop数量
next()方法从EventLoop集合中选择一个
next()方法选择策略具有一定负载均衡功能，大概率是通过定义计数器进行集合取模

1.2 构造方法

// NioEventLoopGroup.java
public NioEventLoopGroup(int nThreads) {
    this(nThreads, (Executor) null);
}

public NioEventLoopGroup(int nThreads, Executor executor) {
        /**
         * executor用于开启NioEventLoop线程所需要的线程执行器
         * SelectorProvider.provider()用于创建selector
         */
        this(nThreads, executor, SelectorProvider.provider());
    }

public NioEventLoopGroup(int nThreads, Executor executor, final SelectorProvider selectorProvider) {
        this(nThreads, executor, selectorProvider, DefaultSelectStrategyFactory.INSTANCE);
    }

public NioEventLoopGroup(int nThreads,
                             Executor executor, // null
                             final SelectorProvider selectorProvider, // 创建Java的NIO复用器
                             final SelectStrategyFactory selectStrategyFactory
    ) {
        super(nThreads, executor, selectorProvider, selectStrategyFactory, RejectedExecutionHandlers.reject());
    }

// MultithreadEventLoopGroup.java
protected MultithreadEventLoopGroup(int nThreads,
                                        Executor executor, // null
                                        Object... args // [SelectorProvider SelectStrategyFactory RejectedExecutionHandlers]
    ) {
        super(nThreads == 0 ? DEFAULT_EVENT_LOOP_THREADS : nThreads, executor, args);
    }

// MultithreadEventExecutorGroup.java
protected MultithreadEventExecutorGroup(int nThreads,
                                            Executor executor, // null
                                            Object... args // [SelectorProvider SelectStrategyFactory RejectedExecutionHandlers]
    ) {
        this(nThreads, executor, DefaultEventExecutorChooserFactory.INSTANCE, args);
    }

protected MultithreadEventExecutorGroup(int nThreads, // 标识着group中有几个EventLoop
                                            Executor executor, // null
                                            EventExecutorChooserFactory chooserFactory, // DefaultEventExecutorChooserFactory.INSTANCE
                                            Object... args // [SelectorProvider SelectStrategyFactory RejectedExecutionHandlers]
    ) {
        if (executor == null)
            executor = new ThreadPerTaskExecutor(this.newDefaultThreadFactory()); // 构造一个executor线程执行器 一个任务对应一个线程(线程:任务=1:n)

        /**
         * 构建NioEventLoop
         * NioEventLoop children数组 线程池中的线程数组
         */
        this.children = new EventExecutor[nThreads];

        for (int i = 0; i < nThreads; i ++) { // 实例化children数组中的每一个元素
            boolean success = false;
            try {
                /**
                 * children是EventExecutor数组 也就是NioEventLoop集合
                 * {@link MultithreadEventExecutorGroup#newChild(Executor, Object...)}方法是抽象方法 具体实现子类关注
                 * 因为是NioEventLoopGroup调用的该方法 所以最终实现在NioEventLoopGroup中
                 *
                 * 将每个EventLoop和每个线程关联起来
                 */
                children[i] = this.newChild(executor, args);
                success = true;
            } catch (Exception e) {
                // TODO: Think about if this is a good exception type
                throw new IllegalStateException("failed to create a child event loop", e);
            } finally {
                if (!success) {
                    for (int j = 0; j < i; j ++) { // 但凡有一个child实例化失败 就把已经成功实例化的线程进行shutdown shutdown是异步操作
                        children[j].shutdownGracefully();
                    }

                    for (int j = 0; j < i; j ++) {
                        EventExecutor e = children[j];
                        try {
                            while (!e.isTerminated()) {
                                e.awaitTermination(Integer.MAX_VALUE, TimeUnit.SECONDS);
                            }
                        } catch (InterruptedException interrupted) {
                            // Let the caller handle the interruption.
                            Thread.currentThread().interrupt(); // 把中断状态设置回去 交给关心的线程来处理
                            break;
                        }
                    }
                }
            }
        }

        /**
         * 创建线程选择器
         * 线程选择策略
         * NioEventLoop都绑定一个chooser对象 作为线程选择器 通过这个线程选择器 为每一个channel分配不同的线程
         */
        this.chooser = chooserFactory.newChooser(children);

        final FutureListener<Object> terminationListener = new FutureListener<Object>() { // 设置一个listener用来监听线程池中的termination事件 给线程池中的每一个线程都设置这个listener 当监听到所有线程都terminate以后 这个线程池就算真正的terminate了
            @Override
            public void operationComplete(Future<Object> future) throws Exception {
                if (terminatedChildren.incrementAndGet() == children.length)
                    terminationFuture.setSuccess(null);
            }
        };

        for (EventExecutor e: children)
            e.terminationFuture().addListener(terminationListener);

        Set<EventExecutor> childrenSet = new LinkedHashSet<EventExecutor>(children.length);
        Collections.addAll(childrenSet, children);
        readonlyChildren = Collections.unmodifiableSet(childrenSet); // 只读集合
    }

其中比较重要的几个点

1.2.1 线程执行器Executor

// MultithreadEventExecutorGroup.java
if (executor == null) // 线程执行器
            executor = new ThreadPerTaskExecutor(this.newDefaultThreadFactory()); // 构造一个executor线程执行器 一个任务对应一个线程(线程:任务=1:n)

ThreadFactory仅仅是线程创建的顶层抽象，提供一些必要线程属性

非守护线程(main线程结束任务线程可以继续执行)
线程优先级

// ThreadPerTaskExecutor
public final class ThreadPerTaskExecutor implements Executor { // 只有一个execute方法 每来一个任务就新建一个线程 这个线程池不是给NioEventLoopGroup使用的 而是给NioEventLoop使用的

    /**
     * 负责创建线程
     */
    private final ThreadFactory threadFactory;

    public ThreadPerTaskExecutor(ThreadFactory threadFactory) {
        this.threadFactory = threadFactory;
    }

    @Override
    public void execute(Runnable command) {
        threadFactory.newThread(command).start(); // 为每个任务新建一个线程 这个execute方法就是用来开启NioEventLoop线程用的
    }
}

也就是将来在某个时机出发executor的execute()方法，为任务创建线程并执行。

此处的线程并非直接使用的Java实现，而是在此基础上做了一些优化。

1.2.2 事件循环器

// MultithreadEventExecutorGroup.java
/**
         * 构建NioEventLoop
         * NioEventLoop children数组 线程池中的线程数组
         */
        this.children = new EventExecutor[nThreads];

for (int i = 0; i < nThreads; i ++) { // 根据NioEventLoopGroup构造器指定的数量创建NioEventLoop 也就是指定数量的线程数(线程的创建动作延迟到任务提交时)
                /**
                 * 初始化NioEventLoop事件循环器集合 也就是多个线程
                 */
                children[i] = this.newChild(executor, args);
            }

1.2.3 线程选择器

// MultithreadEventExecutorGroup.java
/**
         * 创建线程选择器
         * 线程选择策略
         * NioEventLoopGroup都绑定一个chooser对象 作为线程选择器 通过这个线程选择器 为每一个channel发生的读写IO分配不同的线程进行处理
         */
        this.chooser = chooserFactory.newChooser(children);

netty中提供了2种实现

PowerOfTwoEventExecutorChooser
GenericEventExecutorChooser

1.3 next

// DefaultEventExecutorChooserFactory.java
private final EventExecutor[] executors; // 在EventLoopGroup构造器中初始化的EventLoop数组

// DefaultEventExecutorChooserFactory.java::PowerOfTwoEventExecutorChooser
/**
         * next()方法的实现就是选择下一个线程的方法
         * 如果线程数是2的倍数 通过位运算 效率高
         */
        @Override
        public EventExecutor next() { // 线程池线程数是2的幂次方 位运算
            return this.executors[idx.getAndIncrement() & this.executors.length - 1];
        }

// DefaultEventExecutorChooserFactory.java::GenericEventExecutorChooser
/*
         * 线程数不是2的倍数 采用绝对值取模的方式 效率一般
         */
        @Override
        public EventExecutor next() { // 线程池线程数量不是2的幂次方 采用取模方式
            return this.executors[(int) Math.abs(idx.getAndIncrement() % this.executors.length)];
        }

原理比较简单，本质就是对数组长度取模，在一定场景下用位运算替代取模运算。

也就EventLoopGroup这个管理器按照轮询方式从NioEventLoop数组中选择事件循环器执行操作，起到了一定的负载效果。

2 (定时)任务提交

2.1 Demo

// Demo.java
package io.netty.example.basic.eventloopgroup;

import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;

import java.util.concurrent.TimeUnit;

/**
 *
 * @since 2022/11/7
 * @author dingrui
 */
public class NioEventLoopGroupTest01 {

    public static void main(String[] args) {
        EventLoopGroup group = new NioEventLoopGroup();
        group.execute(() -> {
            System.out.println("execute task...");
        });
        group.submit(() -> {
            System.out.println("submit task...");
        });
        group.schedule(() -> {
                    System.out.println("schedule task...");
                },
                5_000,
                TimeUnit.MILLISECONDS);
        System.out.println("main thread end");
    }
}

执行结果

main thread end
submit task...
execute task...

上面已经分析过

EventLoopGroup仅仅是一个管理器，本身不具备执行任务能力
管理器暴露next()选择其内部真正的执行单元(事件循环器)进行工作
每个EventLoop都是一个线程，真正处理任务的执行单元，非守护线程
NioEventLoop是特定处理NIO多路复用这种场景的具体实现，因此还通过组合了Java中Selector多路复用器实现基于事件的响应模式React

2.2 任务执行单元

下面这几个方法都是EventLoopGroup管理器通过选择器chooser选择真正的执行单元EventLoop事件循环器来处理的

execute(...)
submit(...)
schedule(...)
scheduleAtFixedRate(...)
scheduleWithFixedDelay(...)

以execute(...)方法为例

// Demo.java
group.execute(() -> {
            System.out.println("execute task...");
        });

// AbstractEventExecutorGroup.java
public void execute(Runnable command) {
        this.next().execute(command);
    }

// MultithreadEventLoopGroup.java
@Override
    public EventLoop next() { // 从线程池NioEventLoopGroup中选择一个线程NioEventLoop
        return (EventLoop) super.next();
    }

// MultithreadEventExecutorGroup
@Override
    public EventExecutor next() {
        return this.chooser.next();
    }

那么真正执行execute(...)方法的就是EventLoop。

标签：NioEventLoop,Netty,nThreads,NioEventLoopGroup,源码,线程,executor,public
From： https://www.cnblogs.com/miss-u/p/16897723.html

Netty源码-01-NioEventLoopGroup

一定义

二类图

三源码

1 事件循环器选择

1.1 Demo

1.2 构造方法

1.2.1 线程执行器Executor

1.2.2 事件循环器

1.2.3 线程选择器

1.3 next

2 (定时)任务提交

2.1 Demo

2.2 任务执行单元

相关文章

赞助商

阅读排行

Netty源码-01-NioEventLoopGroup

一 定义

二 类图

三 源码

1 事件循环器选择

1.1 Demo

1.2 构造方法

1.2.1 线程执行器Executor

1.2.2 事件循环器

1.2.3 线程选择器

1.3 next

2 (定时)任务提交

2.1 Demo

2.2 任务执行单元

相关文章

赞助商

阅读排行

一定义

二类图

三源码