【Netty】EventLoopGroup

EventLoop继承关系

EventLoopGroup bossGroup = new NioEventLoopGroup();
EventLoopGroup workerGroup = new NioEventLoopGroup();
ServerBootstrap b = new ServerBootstrap();
// 设置EventLoopGroup
b.group(bossGroup, workerGroup);

NioEventLoopGroup

1. NioEventLoopGroup是一个NioEventLoop组，它有一个EventExecutor[] 成员变量，数组中的每一个元素是NioEventLoop类型的对象，所以NioEventLoopGroup可以看做是一系列NioEventLoop的集合。

2. NioEventLoopGroup初始化的时候可以指定线程数，如果没有指定，将获取系统可以的处理器数量*2作为线程数，这个线程数代表着NioEventLoopGroup可以有多少个NioEventLoop。

3. NioEventLoopGroup的构造函数中，有一个Executor类型的参数，默认使用ThreadPerTaskExecutor实现，因为NioEventLoop中也有一个Executor，暂且将这个NioEventLoopGroup的Executor称为全局的Executor，在实例化NioEventLoop的时候，传入了这个全局的Executor。

   • ThreadPerTaskExecutor的execute方法会通过ThreadFactory线程工厂创建一个线程，并且启动该线程，EventLoop中就是通过它来创建线程的。
   • 看到Executor首先会想起Java中的线程池，将需要执行的任务添加到线程池，由线程池管理并执行任务。

4. NioEventLoopGroup中有一个EventExecutorChooser选择器，因为EventExecutor[] 数组中有多个NioEventLoop对象，选择器就是来选择使用哪个NioEventLoop。

NioEventLoopGroup初始化

public class NioEventLoopGroup extends MultithreadEventLoopGroup {
    /**
     * NioEventLoopGroup构造函数
     */
    public NioEventLoopGroup() {
        this(0);
    }
    // 带有线程数的构造函数
    public NioEventLoopGroup(int nThreads) {
        this(nThreads, (Executor) null);
    }
    // 带有线程数和Executor的构造函数
    public NioEventLoopGroup(int nThreads, Executor executor) {
        this(nThreads, executor, SelectorProvider.provider());
    }
    // 带有线程数、Executor和SelectorProvider的构造函数
    public NioEventLoopGroup(
            int nThreads, Executor executor, final SelectorProvider selectorProvider) {
        this(nThreads, executor, selectorProvider, DefaultSelectStrategyFactory.INSTANCE);
    }
   // 带有线程数、Executor、SelectorProvider和SelectStrategyFactory的构造函数
    public NioEventLoopGroup(int nThreads, Executor executor, final SelectorProvider selectorProvider,
                             final SelectStrategyFactory selectStrategyFactory) {
        // 调用父类的构造函数，在MultithreadEventLoopGroup中实现
        super(nThreads, executor, selectorProvider, selectStrategyFactory, RejectedExecutionHandlers.reject());
    }
   
}

MultithreadEventLoopGroup

public abstract class MultithreadEventLoopGroup extends MultithreadEventExecutorGroup implements EventLoopGroup {
    private static final int DEFAULT_EVENT_LOOP_THREADS;
    // EventExecutor线程组
    private final EventExecutor[] children;
    // 注意这个成员变量在父类MultithreadEventExecutorGroup类中
    private final EventExecutorChooserFactory.EventExecutorChooser chooser;
  
    static {
        // 设置默认的EventLoop线程数：先从系统配置中获取io.netty.eventLoopThreads的值，如果没有配置，就获取系统可以的处理器数量*2作为线程数
        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);
        }
    }

    protected MultithreadEventLoopGroup(int nThreads, Executor executor, Object... args) {
        super(nThreads == 0 ? DEFAULT_EVENT_LOOP_THREADS : nThreads, executor, args);
    }
  
   /**
     * 最终会执行这个构造函数
     */
    protected MultithreadEventExecutorGroup(int nThreads, Executor executor,
                                            EventExecutorChooserFactory chooserFactory, Object... args) {
        // 如果线程数小于等于0抛出异常
        if (nThreads <= 0) {
            throw new IllegalArgumentException(String.format("nThreads: %d (expected: > 0)", nThreads));
        }
        // 一个executor线程池，用来创建线程，默认使用ThreadPerTaskExecutor类型的执行器NioEventLoop中也有一个Executor，暂且将这个Executor称为NioEventLoopGroup全局的executor
        if (executor == null) {
            executor = new ThreadPerTaskExecutor(newDefaultThreadFactory());
        }
        // 创建EventExecutor线程组
        children = new EventExecutor[nThreads];
        for (int i = 0; i < nThreads; i ++) {
            boolean success = false;
            try {
                // 为每一个创建EventExecutor完成实例化，在newChild方法中可以看到，创建的是NioEventLoop类型的对象
                children[i] = newChild(executor, args);
                success = true;
            } catch (Exception e) {
                throw new IllegalStateException("failed to create a child event loop", e);
            } finally {
                // 如果出现异常，对EventExecutor进行关闭处理
                if (!success) {
                    for (int j = 0; j < i; j ++) {
                        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) {
                            Thread.currentThread().interrupt();
                            break;
                        }
                    }
                }
            }
        }
        // 初始化选择器ExecutorChooser
        chooser = chooserFactory.newChooser(children);
        // 为每一个EventExecutor添加一个关闭的监听器
        final FutureListener<Object> terminationListener = new FutureListener<Object>() {
            @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);
        }
        // 存放所有的EventExecutor Set集合
        Set<EventExecutor> childrenSet = new LinkedHashSet<EventExecutor>(children.length);
        Collections.addAll(childrenSet, children);
        readonlyChildren = Collections.unmodifiableSet(childrenSet);
    }
}

// ThreadPerTaskExecutor
public final class ThreadPerTaskExecutor implements Executor {
    private final ThreadFactory threadFactory;

    public ThreadPerTaskExecutor(ThreadFactory threadFactory) {
        if (threadFactory == null) {
            throw new NullPointerException("threadFactory");
        }
        this.threadFactory = threadFactory;
    }

    @Override
    public void execute(Runnable command) {
        // 通过线程工厂创建线程
        threadFactory.newThread(command).start();
    }
}
// DefaultThreadFactory
public class DefaultThreadFactory implements ThreadFactory {
  @Override
    public Thread newThread(Runnable r) {
       // 创建线程
        Thread t = newThread(FastThreadLocalRunnable.wrap(r), prefix + nextId.incrementAndGet());
        try {
            if (t.isDaemon() != daemon) {
                t.setDaemon(daemon);
            }

            if (t.getPriority() != priority) {
                t.setPriority(priority);
            }
        } catch (Exception ignored) {
        }
        return t;
    }
}
// NioEventLoopGroup
public class NioEventLoopGroup extends MultithreadEventLoopGroup {
    // 实例化EventExecutor数组的每一个对象
    @Override
    protected EventLoop newChild(Executor executor, Object... args) throws Exception {
        EventLoopTaskQueueFactory queueFactory = args.length == 4 ? (EventLoopTaskQueueFactory) args[3] : null;
        // 创建NioEventLoop，这里传入了NioEventLoopGroup的executor
        return new NioEventLoop(this, executor, (SelectorProvider) args[0],
            ((SelectStrategyFactory) args[1]).newSelectStrategy(), (RejectedExecutionHandler) args[2], queueFactory);
    }
}

// DefaultEventExecutorChooserFactory
public final class DefaultEventExecutorChooserFactory implements EventExecutorChooserFactory {

    public static final DefaultEventExecutorChooserFactory INSTANCE = new DefaultEventExecutorChooserFactory();

    @SuppressWarnings("unchecked")
    @Override
    public EventExecutorChooser newChooser(EventExecutor[] executors) {
        // 实例化ExecutorChooser，判断数组长度是否是2的整数次幂
        if (isPowerOfTwo(executors.length)) {
            return new PowerOfTwoEventExecutorChooser(executors);
        } else {
            return new GenericEventExecutorChooser(executors);
        }
    }
}

NioEventLoop

1. NioEventLoop中包含了一个Selector
2. NioEventLoop中有三种任务队列：
   • 普通任务队列taskQueue，一些核心的任务就是放在taskQueue中的
   • 定时任务队列scheduledTaskQueue，处理一些定时任务
   • 收尾任务队列tailTasks，做一些收尾的工作
3. 每个NioEventLoop中也有一个Executor，因为NioEventLoop是Executor的子类，所以NioEventLoop本身也是一个Executor
4. ThreadExecutorMap中维护了线程和EventLoop的绑定关系，记录了每个线程绑定到了哪个EventLoop

public final class NioEventLoop extends SingleThreadEventLoop {
  
    /**
     * selector
     */
    private Selector selector;
    private Selector unwrappedSelector;
    
    // 构造函数
    NioEventLoop(NioEventLoopGroup parent, Executor executor, SelectorProvider selectorProvider,
                 SelectStrategy strategy, RejectedExecutionHandler rejectedExecutionHandler,
                 EventLoopTaskQueueFactory queueFactory) {
        // 调用父类SingleThreadEventLoop的构造函数
        super(parent, executor, false, newTaskQueue(queueFactory), newTaskQueue(queueFactory),
                rejectedExecutionHandler);
        if (selectorProvider == null) {
            throw new NullPointerException("selectorProvider");
        }
        // SelectStrategy判空
        if (strategy == null) {
            throw new NullPointerException("selectStrategy");
        }
        provider = selectorProvider;
        final SelectorTuple selectorTuple = openSelector();
        // 封装的selector
        selector = selectorTuple.selector;
        // 原始的selector
        unwrappedSelector = selectorTuple.unwrappedSelector;
        selectStrategy = strategy;
    }
    // 创建任务队列
    private static Queue<Runnable> newTaskQueue(EventLoopTaskQueueFactory queueFactory) {
        return queueFactory == null?newTaskQueue0(DEFAULT_MAX_PENDING_TASKS):queueFactory.newTaskQueue(DEFAULT_MAX_PENDING_TASKS);
    }
}

// io.netty.channel.SingleThreadEventLoop
public abstract class SingleThreadEventLoop extends SingleThreadEventExecutor implements EventLoop {
    // tailTask
    private final Queue<Runnable> tailTasks;
    // 构造函数
    protected SingleThreadEventLoop(EventLoopGroup parent, Executor executor,
                                    boolean addTaskWakesUp, Queue<Runnable> taskQueue, Queue<Runnable> tailTaskQueue,
                                    RejectedExecutionHandler rejectedExecutionHandler) {
        // 调用SingleThreadEventExecutor父类构造函数
        super(parent, executor, addTaskWakesUp, taskQueue, rejectedExecutionHandler);
        // 设置 tailTask
        tailTasks = ObjectUtil.checkNotNull(tailTaskQueue, "tailTaskQueue");
    }
}

// io.netty.util.concurrent.SingleThreadEventExecutor
public abstract class SingleThreadEventExecutor extends AbstractScheduledEventExecutor implements OrderedEventExecutor {
    // 普通任务任务队列taskQueue
    private final Queue<Runnable> taskQueue;
    private final Executor executor;
    private final RejectedExecutionHandler rejectedExecutionHandler;

    protected SingleThreadEventExecutor(EventExecutorGroup parent, Executor executor,
                                        boolean addTaskWakesUp, Queue<Runnable> taskQueue,
                                        RejectedExecutionHandler rejectedHandler) {
        // 调用父类的构造函数
        super(parent);
        this.addTaskWakesUp = addTaskWakesUp;
        this.maxPendingTasks = DEFAULT_MAX_PENDING_EXECUTOR_TASKS;
        // 初始化NioEventLoop的executor，this.executor 指的是当前对象的executor，为了防止混淆，先看作是NioEventLoop的executor，apply(executor, this)中的executor，是NioEventLoopGroup实例化的时候创建的全局的Executor
        this.executor = ThreadExecutorMap.apply(executor, this);
        // 设置taskQueue
        this.taskQueue = ObjectUtil.checkNotNull(taskQueue, "taskQueue");
        // 拒绝策略
        rejectedExecutionHandler = ObjectUtil.checkNotNull(rejectedHandler, "rejectedHandler");
    }
}

// io.netty.util.concurrent.AbstractScheduledEventExecutor
public abstract class AbstractScheduledEventExecutor extends AbstractEventExecutor {
    // 定时任务队列
    PriorityQueue<ScheduledFutureTask<?>> scheduledTaskQueue;
}

ThreadExecutorMap

ThreadExecutorMap中记录了每个线程绑定的EventLoop，从名字上也能看出，它记录的是Thread线程和Executor的对应关系。

public final class ThreadExecutorMap {
    // FastThreadLocal类型的，记录每个线程绑定的EventLoop(EventLoop是EventExecutor的子类)
    private static final FastThreadLocal<EventExecutor> mappings = new FastThreadLocal<EventExecutor>();

    // 设置当前线程关联的EventExecutor
    private static void setCurrentEventExecutor(EventExecutor executor) {
        mappings.set(executor);
    }
    // executor是NioEventLoopGroup实例化的时候创建的全局的Executor，eventExecutor当前正在创建的EventLoop（NioEventLoop）对象
    public static Executor apply(final Executor executor, final EventExecutor eventExecutor) {
        ObjectUtil.checkNotNull(executor, "executor");
        ObjectUtil.checkNotNull(eventExecutor, "eventExecutor");
        // 创建一个Executor
        return new Executor() {
            @Override
            public void execute(final Runnable command) {
                // 全局的executor的execute方法会创建一个线程，并且启动线程，所以apply返回Runnable对象后，会交给全局的executor来创建线程并启动
                executor.execute(apply(command, eventExecutor));
            }
        };
    }

    public static Runnable apply(final Runnable command, final EventExecutor eventExecutor) {
        ObjectUtil.checkNotNull(command, "command");
        ObjectUtil.checkNotNull(eventExecutor, "eventExecutor");
        return new Runnable() {
            @Override
            public void run() {
                // 设置当前线程关联的eventExecutor
                setCurrentEventExecutor(eventExecutor);
                try {
                    // 运行任务
                    command.run();
                } finally {
                    setCurrentEventExecutor(null);
                }
            }
        };
    }
}

Channel初始化与注册

以Channel初始化与注册的过程为例，看一下EventLoop的使用：

回到AbstractBootstrap的initAndRegister初始化并注册Channel的方法，可以看到获取了EventLoopGroup，然后调用它的register方法进行注册。

EventLoopGroup的register实际上是通过选择器，从EventLoop数组中选择了一个EventLoop返回，最终调用EventLoop的register方法进行注册。

// AbstractBootstrap
final ChannelFuture initAndRegister() {
    // ...
    // 获取EventLoopGroup，调用register进行注册
    ChannelFuture regFuture = config().group().register(channel);
    // ...
}

// AbstractBootstrapConfig
@SuppressWarnings("deprecation")
public final EventLoopGroup group() {
    return bootstrap.group();
}

// MultithreadEventLoopGroup
@Override
public ChannelFuture register(Channel channel) {
    //调用了next()方法获取一个EventLoop
    return next().register(channel);
}
// 获取EventLoop
@Override
public EventLoop next() {
    // 调用父类的next方法,进入到MultithreadEventExecutorGroup的next方法
    return (EventLoop) super.next();
}

// MultithreadEventExecutorGroup
@Override
public EventExecutor next() {
    // 选择一个EventLoop
    return chooser.next();
}
// DefaultEventExecutorChooserFactory
private static final class PowerOfTwoEventExecutorChooser implements EventExecutorChooser {
    private final AtomicInteger idx = new AtomicInteger();
    private final EventExecutor[] executors;

    PowerOfTwoEventExecutorChooser(EventExecutor[] executors) {
        this.executors = executors;
    }

    @Override
    public EventExecutor next() {
        // 选择一个EventLoop
        return executors[idx.getAndIncrement() & executors.length - 1];
    }
}

AbstractChannel的register进行注册：

1. 可以看到将当前的Channel绑定到了选择EventLoop上。
2. 判断当前线程是否是NioEventLoop中的线程,首次启动注册Channel时，当前线程不是NioEventLoop的线程，所以会走到else的代码中，向EventLoop中添加了一个注册任务。

public abstract class AbstractChannel extends DefaultAttributeMap implements Channel {
        @Override
        public final void register(EventLoop eventLoop, final ChannelPromise promise) {
            if (eventLoop == null) {
                throw new NullPointerException("eventLoop");
            }
            // 判断是否已经注册过
            if (isRegistered()) {
                promise.setFailure(new IllegalStateException("registered to an event loop already"));
                return;
            }
            if (!isCompatible(eventLoop)) {
                promise.setFailure(
                        new IllegalStateException("incompatible event loop type: " + eventLoop.getClass().getName()));
                return;
            }
            // 设置当前channel的eventLoop，也就是将channel绑定到了EventLoop上
            AbstractChannel.this.eventLoop = eventLoop;
            // 当前线程是否是EventLoop中的线程
            if (eventLoop.inEventLoop()) {
                // 注册channel
                register0(promise);
            } else {
                try {
                    // 将注册任务放入EventLoop中执行，SingleThreadEventExecutor中实现了execute方法
                    eventLoop.execute(new Runnable() {
                        @Override
                        public void run() {
                            // 注册channel
                            register0(promise);
                        }
                    });
                } catch (Throwable t) 
                     // 省略...
                }
            }
        }
}
// AbstractEventExecutor
public abstract class AbstractEventExecutor extends AbstractExecutorService implements EventExecutor {
    @Override
    public boolean inEventLoop() {
        return inEventLoop(Thread.currentThread());
    }
    // SingleThreadEventExecutor中实现
    @Override
    public boolean inEventLoop(Thread thread) {
        // 判断当前线程与EventLoop中的线程是否相等
        return thread == this.thread;
    }
}
// SingleThreadEventExecutor
public abstract class SingleThreadEventExecutor extends AbstractScheduledEventExecutor implements OrderedEventExecutor {
    private volatile Thread thread;
}

SingleThreadEventExecutor

SingleThreadEventExecutor实现了execute方法，进行任务添加：

1. 将需要执行的任务放到了taskQueue任务队列中
2. 如果不是EventLoop线程（首次启动Channel不是EventLoop线程），调用startThread开启了一个线程，开启方式是通过调用EventLoop的executor，向线程池中执行任务，最终是通过EventLoopGroup的executor的execute方法会创建线程并且启动线程。
   • 之前提到过EventLoop也是一个Executor，所以它本身也可以接受任务，实现了execute方法，这个方法就是在SingleThreadEventExecutor中实现的。
   • EventLoop中有一个Executor的成员变量，它的execute方法是通过EventLoopGroup的Executor创建线程使用的。
3. 将NioEventLoop的thread成员变量设置为当前获取到的线程，实现了当前线程与NioEventLoop的绑定

//io.netty.util.concurrent.SingleThreadEventExecutor
public abstract class SingleThreadEventExecutor extends AbstractScheduledEventExecutor implements OrderedEventExecutor {
    // 添加任务,这个execute方法是EventLoop自己的
    @Override
    public void execute(Runnable task) {
        if (task == null) {
            throw new NullPointerException("task");
        }
        // 是否是EventLoop中的线程
        boolean inEventLoop = inEventLoop();
        // 将任务添加到taskQueue
        addTask(task);
        // 如果不是EventLoop线程
        if (!inEventLoop) {
            // 开启线程
            startThread();
            if (isShutdown()) {
                boolean reject = false;
                try {
                    if (removeTask(task)) {
                        reject = true;
                    }
                } catch (UnsupportedOperationException e) {
                }
                if (reject) {
                    reject();
                }
            }
        }

        if (!addTaskWakesUp && wakesUpForTask(task)) {
            wakeup(inEventLoop);
        }
    }
  
    // 开启线程
    private void startThread() {
        // 是否未启动状态
        if (state == ST_NOT_STARTED) {
            // 设置为启动状态
            if (STATE_UPDATER.compareAndSet(this, ST_NOT_STARTED, ST_STARTED)) {
                boolean success = false;
                try {
                    // 开启线程
                    doStartThread();
                    success = true;
                } finally {
                    if (!success) {
                        // 如果失败，状态设置未启动
                        STATE_UPDATER.compareAndSet(this, ST_STARTED, ST_NOT_STARTED);
                    }
                }
            }
        }
    }
    // 开启线程
    private void doStartThread() {
        assert thread == null;
        // 调用EventLoop的executor，向线程池中添加任务，execute方法会进入到ThreadExecutorMap类中apply方法实例化Executor时实现的execute方法
        executor.execute(new Runnable() {
            @Override
            public void run() {
                // 这里将NioEventLoop的thread成员变量设置为当前获取到的线程，也是实现了当前线程与NioEventLoop的绑定
                thread = Thread.currentThread();
                if (interrupted) {
                    thread.interrupt();
                }
                boolean success = false;
                updateLastExecutionTime();
                try {
                    // 执行核心任务，在NioEventLoop中实现
                    SingleThreadEventExecutor.this.run();
                    success = true;
                } catch (Throwable t) {
                    logger.warn("Unexpected exception from an event executor: ", t);
                } finally {
                  // 省略了部分代码
                }
            }
        });
    }

    // 向taskQueue队列中添加任务
    protected void addTask(Runnable task) {
        if (task == null) {
            throw new NullPointerException("task");
        }
        if (!offerTask(task)) {
            reject(task);
        }
    }
    final boolean offerTask(Runnable task) {
        if (isShutdown()) {
            reject();
        }
        // 添加到taskQueue
        return taskQueue.offer(task);
    }
}

ThreadExecutorMap

在NioEventLoop实例化的时候可以知道，NioEventLoop的Executor是在ThreadExecutorMap中完成实例化的，所以调用NioEventLoop的Executor的execute方法就会进入ThreadExecutorMap中实例化Executor时重写的execute方法，在这个方法中，又会调用EventLoopGroup的executor创建线程并启动线程。

public static Executor apply(final Executor executor, final EventExecutor eventExecutor) {
     ObjectUtil.checkNotNull(executor, "executor");
     ObjectUtil.checkNotNull(eventExecutor, "eventExecutor");
     // 创建一个Executor对象
     return new Executor() {
         // 并重写execute方法
         @Override
         public void execute(final Runnable command) {
             // EventLoopGroup的executor的execute方法会创建一个线程，并且启动线程，所以apply返回Runnable对象后，会交给全局的executor来创建线程并启动
             executor.execute(apply(command, eventExecutor));
         }
     };
 }

总结：

到这里，应该已经知道EventLoop和EventLoopGroup的作用是什么了，再次总结一下：

1. NioEventLoopGroup是一个NioEventLoop组，里面包含多个NioEventLoop，可以看做是一系列NioEventLoop的集合

2. NioEventLoopGroup初始化会实例化一个Executor，它是用来创建线程使用的，可以看做是一个创建线程的线程池

3. NioEventLoopGroup会通过选择策略，类似于负载均衡算法从NioEventLoop数组中选取一个NioEventLoop

4.NioEventLoop中有三个任务队列，分别是taskQueue(处理普通任务)、tailTasks(收尾工作处理)和scheduledTaskQueue(定时任务队列)

5.NioEventLoop是Executor的子类，所以它本身也是一个Executor，可以向NioEventLoop中提交任务，任务提交后会添加到任务队列中等待处理

6.NioEventLoop中也有一个Executor，以成员变量的形式存在，它可以调用NioEventLoopGroup的Executor来创建线程，执行任务

Netty版本：4.1.42.Final

参考：

Netty（八）源码解析 之 NioEventLoopGroup和NioEventLoop源码分析、NioEventLoop绑定线程的流程分析