Netty入门（三）：EventLoop

前言

Netty系列索引：

1.Netty入门（一）：ByteBuf

2.Netty入门（二）：Channel

IO相关：

1.Java基础（一）：I/O多路复用模型及Linux中的应用

上文提到，早期基于线程的网络模型，处理高并发的能力非常差，随着请求数量的增多，必须不断新建线程，随之带来的问题就是服务器资源被占满、上下文切换成本过高以及IO阻塞导致的CPU浪费。

而Netty则使用了经典Reactor模型，并重新进行了封装，包括EventLoop、EventGroup等。

EventLoopGroup

EventLoopGroup是一个接口，继承自线程池EventExecutorGroup，并允许注册channel到自身所持有的EventLoop，同时支持按一定规则获取下一个EventLoop。

EventLoopGroup的具体实现有很多，下面以DefaultEventLoopGroup为例，简述一下我的理解

1.ScheduledExecutorService

JDK接口，一个延迟或定时任务的执行器，其实现类ScheduledThreadPoolExecutor主要是利用了延时队列及设置下次执行时间来实现的，这里不再赘述（可以单独开个专题0.0）

2.EventExecutorGroup

接口，Netty自定义的一个线程池，负责复用EventExecutor和执行任务

3.EventLoopGroup

核心接口，EventLoopGroup继承自EventExecutorGroup，代表他是一个线程池。同时他具备将channel注册到EventExecutorGroup的功能，代表他是一个能够真正处理Channel的特殊线程池

4.MultithreadEventExecutorGroup（AbstractEventExecutorGroup）

抽象类，实现自EventExecutorGroup接口，提供了一个简易线程池的实现，其只有一个抽象方法newChild（创建EventExecutor）供子类实现

1
protected MultithreadEventExecutorGroup(int nThreads, Executor executor, EventExecutorChooserFactory chooserFactory, Object... args)

4.1 nThreads

1
private final EventExecutor[] children;

该线程池通过数组存储线程，入参nThreads指定数组大小，并循环调用newChild创建线程。当创建过程中有异常时，会自动调用已创建完成线程的shutdownGracefully方法，进行优雅关闭

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
for (int i = 0; i < nThreads; i ++) {
    boolean success = false;
    try {
        children[i] = 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 ++) {
                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;
                }
            }
        }
    }
}

4.2 EventExecutorChooserFactory

EventExecutorChooserFactory是一个工厂接口，负责创建EventExecutorChooser

其默认实现DefaultEventExecutorChooserFactory会判断当前线程数是否2的n次幂，如果是则返回PowerOfTwoEventExecutorChooser，否则返回GenericEventExecutorChooser

4.3 EventExecutorChooser

1
private final EventExecutorChooserFactory.EventExecutorChooser chooser;

EventExecutorChooser负责根据一定规则从线程池children数组中取得下一个线程

PowerOfTwoEventExecutorChooser：通过&运算，将超出executors.length的位置为0

1
2
3
public EventExecutor next() {
    return executors[idx.getAndIncrement() & executors.length - 1];
}

GenericEventExecutorChooser：通过求余运算，获取有效index

1
2
3
public EventExecutor next() {
    return executors[Math.abs(idx.getAndIncrement() % executors.length)];
}

可以看出当线程数是2的n次幂时，Netty通过与运算优化了效率

5.MultithreadEventLoopGroup

抽象类，继承自MultithreadEventExecutorGroup并实现了EventLoopGroup接口，代表此抽象类是一个可以注册并处理channel的线程池

值得关注的是next方法，他把返回值的类型，进一步限定为EventLoop

1
2
3
public EventLoop next() {
    return (EventLoop) super.next();
}

6.DefaultEventLoopGroup

MultithreadEventLoopGroup的一个默认实现

其核心就是实现了newChild方法返回一个EventLoop extends EventExecutor实例

1
2
3
protected EventLoop newChild(Executor executor, Object... args) throws Exception {
    return new DefaultEventLoop(this, executor);
}

7.总结

说白了EventLoopGroup核心方法，register负责将channel与线程池中某一线程绑定，next负责返回下一个线程供调用方执行任务

EventLoop

EventLoop直译为事件循环，他的职责简单来说就是绑定一个唯一的线程，去执行或调度被分配的任务。

可见一个EventLoop实例可以为多个channel服务，而为了最大化利用资源，Netty使用池化技术将EventLoop放入EventLoopGroup中管理。

EventLoop的具体实现有很多，先看下DefaultEventLoop的类图，会发现他和DefaultEventLoopGroup的类图很像，都继承了EventLoopGroup接口，但其最大的不同是红框所示，他还继承了EventExecutor，下面主要讲一下多出来的这部分到底是干了什么

1.EventExecutor

接口，定义了一个事件执行器，主要方法如下

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
/**
 * 直接返回自身
 */
@Override
EventExecutor next();

/**
 * 返回所属线程池
 */
EventExecutorGroup parent();

/**
 * 判断当前线程是否是当前EventLoop绑定的线程
 */
boolean inEventLoop();

/**
 * 判断传入线程是否是当前EventLoop绑定的线程
 */
boolean inEventLoop(Thread thread);

（还涉及一些Future异步编程的一些东西，太复杂了后续再填坑吧0.0）

2.AbstractScheduledEventExecutor （AbstractEventExecutor）

抽象类，简单定义了一个支持延迟（定时）任务的执行器

1
2
3
4
   //延时队列
   PriorityQueue<ScheduledFutureTask<?>> scheduledTaskQueue;
//下一个任务id
   long nextTaskId;

重要方法scheduled

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
private <V> ScheduledFuture<V> schedule(final ScheduledFutureTask<V> task) {
    if (inEventLoop()) {
    	//如果和执行器绑定的线程一致，直接放入延时队列中
        scheduleFromEventLoop(task);
    } else {
    	//获取任务最晚执行时间
        final long deadlineNanos = task.deadlineNanos();
        // task will add itself to scheduled task queue when run if not expired
        if (beforeScheduledTaskSubmitted(deadlineNanos)) {
        	//放入线程池执行
            execute(task);
        } else {
        	//与execute类似，但不保证任务会在执行非延迟任务或执行程序关闭之前运行，默认实现只是委托给execute(Runnable) 
            lazyExecute(task);
            // Second hook after scheduling to facilitate race-avoidance
            if (afterScheduledTaskSubmitted(deadlineNanos)) {
                execute(WAKEUP_TASK);
            }
        }
    }

    return task;
}

3.SingleThreadEventExecutor

抽象类，定义了一个单线程顺序执行器

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
  private void execute(Runnable task, boolean immediate) {
      boolean inEventLoop = inEventLoop();
      //添加到任务队列
      addTask(task);
      if (!inEventLoop) {
      	//启动线程
          startThread();
          //如果线程池已经关闭，调用拒绝方法
          if (isShutdown()) {
              boolean reject = false;
              try {
                  if (removeTask(task)) {
                      reject = true;
                  }
              } catch (UnsupportedOperationException e) {
                  // The task queue does not support removal so the best thing we can do is to just move on and
                  // hope we will be able to pick-up the task before its completely terminated.
                  // In worst case we will log on termination.
              }
              if (reject) {
                  reject();
              }
          }
      }
//唤醒线程
      if (!addTaskWakesUp && immediate) {
          wakeup(inEventLoop);
      }
  }

4.SingleThreadEventLoop

1
public abstract class SingleThreadEventLoop extends SingleThreadEventExecutor implements EventLoop

EventLoop的抽象基类，负责在单线程中执行所有被提交的任务，同时具有注册和处理channle的能力

5.DefaultEventLoop

单线程任务执行器的默认实现，主要就是其实现的run方法

1
2
3
4
5
6
7
8
9
10
11
12
13
14
protected void run() {
    //循环阻塞的获取任务，知道被通知关闭
    for (;;) {
        Runnable task = takeTask();
        if (task != null) {
            task.run();
            updateLastExecutionTime();
        }

        if (confirmShutdown()) {
            break;
        }
    }
}

6.总结

通过以上分析，不难看出Netty首先定义了自己的线程池（EventExectorGroup）和执行器（EventExector），然后通过继承的方式定义了线程池（EventLoopGroup）和执行器（EventLoop），从而添加了处理（注册）channel的能力。

You gotta grab what you can when you can.
机不可失，时不我待。