Netty 核心原理四 MultithreadEventExecutorGroup 原理二

ThreadPerTaskExecutor原理

ThreadPerTaskExecutor为MultithreadEventExecutorGroup默认的Executor执行器，该执行器我们在前面看到将会在newChild(executor, args)方法中传递给子执行器完成对子执行器EventExecutor对象的创建，同时用于启动线程来执行子执行器任务。我们看到该执行器同它的名字一样，在execute方法中通过ThreadFactory线程工厂创建一个新的线程来处理传递过来的任务command。而又由于EventExecutor子执行器对象只需要一个线程来处理它的任务（只有一个执行器的组），所以这里很容易推理得出：有多少个子执行器，那么这里就会产生出多少个线程。那么为何不用线程池？线程池用于缓存线程，减少每次执行任务时对线程的创建和销毁时间，那么这里需要么？答案是否定的，因为我们只会提交与EventExecutor数组长度相同的子执行器任务。源码描述如下。

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原理

我们看到默认的线程对象通过poolId和nextId来生成线程名前缀，然后根据构造函数来设置线程的属性：守护线程、线程优先级、线程组。这里需要注意以下两点：

1. 我们对传入的Runnable执行对象进行了包装，将其包装为DefaultRunnableDecorator，目的是为了在任务执行完毕后对ThreadLocal进行清理
2. 创建的线程对象为FastThreadLocalThread

public class DefaultThreadFactory implements ThreadFactory {



 private static final AtomicInteger poolId = new AtomicInteger(); // 线程池ID，用于生成线程名

 private final AtomicInteger nextId = new AtomicInteger(); // 线程ID，用于生成线程名

 private final String prefix; // 生成的线程名前缀

 private final boolean daemon; // 是否为守护线程

 private final int priority;// 线程优先级

 protected final ThreadGroup threadGroup;// 线程组



 // 完整构造器。在对参数进行校验后，保存生成线程前缀、是否为daemon守护线程、优先级、所属线程组

 public DefaultThreadFactory(String poolName, boolean daemon, int priority, ThreadGroup threadGroup) {

  if (poolName == null) {

   throw new NullPointerException("poolName");

  }

  if (priority < Thread.MIN_PRIORITY || priority > Thread.MAX_PRIORITY) {

   throw new IllegalArgumentException(

    "priority: " + priority + " (expected: Thread.MIN_PRIORITY <= priority <= Thread.MAX_PRIORITY)");

  }



  prefix = poolName + '-' + poolId.incrementAndGet() + '-';

  this.daemon = daemon;

  this.priority = priority;

  this.threadGroup = threadGroup;

 }



 // 生成新的线程对象

 @Override

 public Thread newThread(Runnable r) {

  Thread t = newThread(new DefaultRunnableDecorator(r), prefix + nextId.incrementAndGet());

  // 设置线程属性

  try {

   if (t.isDaemon()) {

    if (!daemon) {

     t.setDaemon(false);

    }

   } else {

    if (daemon) {

     t.setDaemon(true);

    }

   }



   if (t.getPriority() != priority) {

    t.setPriority(priority);

   }

  } catch (Exception ignored) {

   // Doesn't matter even if failed to set.

  }

  return t;

 }



 // 创建线程对象

 protected Thread newThread(Runnable r, String name) {

  return new FastThreadLocalThread(threadGroup, r, name);

 }



 // 用于装饰需要执行的Runnable任务，标准的装饰者模式。这里主要对任务进行封装，在任务执行完毕后，移除ThreadLocal的存储，避免内存泄漏

 private static final class DefaultRunnableDecorator implements Runnable {



  private final Runnable r;



  DefaultRunnableDecorator(Runnable r) {

   this.r = r;

  }



  @Override

  public void run() {

   try {

    r.run();

   } finally {

    FastThreadLocal.removeAll();

   }

  }

 }

}

FastThreadLocalThread原理

前面我们看到默认线程工厂创建的线程对象为FastThreadLocalThread，而该线程工厂又是在ThreadPerTaskExecutor执行器中调用，而ThreadPerTaskExecutor线程工厂又在MultithreadEventExecutorGroup中传递给了子执行器，那么很明显，我们的子执行器在操作任务时，创建的线程对象就是FastThreadLocalThread，读者这里注意：Netty中的执行线程对象就是该类的实例。

我们看到该类继承自Thread类，因为Thread类定义了一个Java线程对象，所以只能通过继承该类对其进行扩展。我们看到这里面主要扩展了对InternalThreadLocalMap的操作，那么InternalThreadLocalMap又是什么？它和线程自带的ThreadLocal.ThreadLocalMap有何区别？别急，先了解这个类，下一节笔者会详细介绍，只需要知道这里扩展了这个InternalThreadLocalMap类即可。源码如下。

public class FastThreadLocalThread extends Thread {



 private InternalThreadLocalMap threadLocalMap;





 public final InternalThreadLocalMap threadLocalMap() {

  return threadLocalMap;

 }



 public final void setThreadLocalMap(InternalThreadLocalMap threadLocalMap) {

  this.threadLocalMap = threadLocalMap;

 }

}

FastThreadLocal原理

我们看看这个名字，前面叫做FastThreadLocalThread，这里又叫FastThreadLocal，无外乎就是Fast，快，到底哪里快呢？这还得从JDK自带的ThreadLocal.ThreadLocalMap说起。我们知道ThreadLocalMap为JDK的ThreadLocal类中的一个内部类，我们可以创建一个ThreadLocal对象，然后在线程中保存线程本地变量。那么我们来看看这个放入过程：

1. 根据ThreadLocalMap是否存在来决定是否创建ThreadLocalMap集合
2. 在放入过程中通过key.threadLocalHashCode & (len-1)计算当前ThreadLocal与value的映射放入的下标。同理，在获取时也会该算法来获取下标处对应的Entry映射

public class ThreadLocal<T> {

 // 设置线程本地变量

 public void set(T value) { 

  Thread t = Thread.currentThread(); // 获取当前线程

  ThreadLocalMap map = getMap(t); // 获取线程map

  if (map != null) // map不为空，那么直接放入

   map.set(this, value);

  else

   createMap(t, value); // 否则创建线程map

 }



 // 获取线程map。可以看到这里直接取线程对象的ThreadLocal.ThreadLocalMap对象

 ThreadLocalMap getMap(Thread t) {

  return t.threadLocals;

 }



 // 创建线程map，可以看到这里初始化了ThreadLocalMap对象

 void createMap(Thread t, T firstValue) {

  t.threadLocals = new ThreadLocalMap(this, firstValue);

 }



 // 用于处理ThreadLocal和value映射的线程map

 static class ThreadLocalMap {

  

  private Entry[] table; // 保存映射的数组

  

  ThreadLocalMap(ThreadLocal<?> firstKey, Object firstValue) {

   table = new Entry[INITIAL_CAPACITY]; // 创建数组

   int i = firstKey.threadLocalHashCode & (INITIAL_CAPACITY - 1); // 计算当前ThreadLocal中对应value的下标

   table[i] = new Entry(firstKey, firstValue); // 将映射关系放入下标所在的entry中

   size = 1;

   setThreshold(INITIAL_CAPACITY);

  }

  

  // 将指定的映射关系放入对应的数组下标

  private void set(ThreadLocal<?> key, Object value) {

   Entry[] tab = table;

   int len = tab.length;

   int i = key.threadLocalHashCode & (len-1); // 根据hash值计算该映射关系应该放置的索引下标

   ... // 省略掉发生冲突二次寻址的过程

   tab[i] = new Entry(key, value); // 创建新的entry

   ...

  }

  

  // 根据ThreadLocal对象来获取映射值

  private Entry getEntry(ThreadLocal<?> key) {

   int i = key.threadLocalHashCode & (table.length - 1);

   ...

  }

 }

}

我们先来看FastThreadLocal类的原理。我们先来看放入方法。根据源码我们得出以下结论：

1. FastThreadLocal中的variablesToRemoveIndex下标通常为0，同时在该下标处存放着一个Set集合，该集合用于保存所有在InternalThreadLocalMap中保存值的FastThreadLocal
2. 保存FastThreadLocal的映射值的核心类为InternalThreadLocalMap
3. 在生成FastThreadLocal对象时，将会给FastThreadLocal生成一个唯一的下标index，通过该下标，我们就不需要像原生JDK的ThreadLocalMap一样，通过计算hash值来取余数找到需要放置的下标
4. 在放置对象时，我们可以根据是否放置的值为InternalThreadLocalMap.UNSET占位符来决定是否从InternalThreadLocalMap中移除映射
5. 在移除映射时，我们先将映射值从InternalThreadLocalMap取出，然后设置InternalThreadLocalMap中对应该FastThreadLocal index下标处为UNSET占位对象。然后根据移除的对象值是否存在来决定回调onRemoval钩子函数

public class FastThreadLocal<V> {



 private static final int variablesToRemoveIndex = InternalThreadLocalMap.nextVariableIndex(); // 该类创建时生成的唯一id，通常为0



 private final int index; // 当前FastThreadLocal在InternalThreadLocalMap中的下标



 public FastThreadLocal() { // 创建对象时生成唯一的下标

  index = InternalThreadLocalMap.nextVariableIndex(); 

 }



 public final void set(V value) {

  if (value != InternalThreadLocalMap.UNSET) { // value不是占位对象UNSET，那么直接设置

   set(InternalThreadLocalMap.get(), value);

  } else { // 否则执行清理操作

   remove();

  }

 }



 public final void set(InternalThreadLocalMap threadLocalMap, V value) {

  if (value != InternalThreadLocalMap.UNSET) { // 由于该方法是通用方法，所以这里还得检测一次占位对象

   if (threadLocalMap.setIndexedVariable(index, value)) { // 注意：这里直接根据index下标设置值即可

    addToVariablesToRemove(threadLocalMap, this); // 设置成功，那么将当前FastThreadLocal对象添加到threadLocalMap对应的variablesToRemoveIndex下标处，表明需要进行清理的FastThreadLocal（这里通常为0下标）

   }

  } else {

   remove(threadLocalMap);

  }

 }



 // 将FastThreadLocal<?> variable添加到InternalThreadLocalMap threadLocalMap对应的下标处

 private static void addToVariablesToRemove(InternalThreadLocalMap threadLocalMap, <?> variable) {

  Object v = threadLocalMap.indexedVariable(variablesToRemoveIndex); // 获取对应的下标值

  Set<FastThreadLocal<?>> variablesToRemove;

  if (v == InternalThreadLocalMap.UNSET || v == null) { // 如果集合不存在，那么创建集合

   variablesToRemove = Collections.newSetFromMap(new IdentityHashMap<FastThreadLocal<?>, Boolean>());

   threadLocalMap.setIndexedVariable(variablesToRemoveIndex, variablesToRemove);

  } else { // 否则将已经存在的v转为Set<FastThreadLocal<?>>集合

   variablesToRemove = (Set<FastThreadLocal<?>>

        Set<FastThreadLocal<?>>) v;

  }

  // 将FastThreadLocal对象放入集合

  variablesToRemove.add(variable);

 }



 // 清理当前FastThreadLocal在InternalThreadLocalMap中的存储

 public final void remove(InternalThreadLocalMap threadLocalMap) {

  if (threadLocalMap == null) { 

   return;

  }

  Object v = threadLocalMap.removeIndexedVariable(index); // 从threadLocalMap对应的index下标中移除该对象，并设置为UNSET，返回该对象值

  removeFromVariablesToRemove(threadLocalMap, this); // 由于该FastThreadLocal的映射值已经从InternalThreadLocalMap中移除，那么需要将其从下标为variablesToRemoveIndex处的Set集合中移除



  if (v != InternalThreadLocalMap.UNSET) { // 如果移除成功，那么回调钩子方法（默认为空，子类可以重写来完成自己的逻辑）

   try {

    onRemoval((V) v);

   } catch (Exception e) {

    PlatformDependent.throwException(e);

   }

  }

 }

 

 // 将指定的FastThreadLocal从InternalThreadLocalMap中对应下标为variablesToRemoveIndex（通常为0）中的Set集合中移除

 private static void removeFromVariablesToRemove(

   InternalThreadLocalMap threadLocalMap, FastThreadLocal<?> variable) {

  // 获取set集合并掉用其remove方法移除

  Object v = threadLocalMap.indexedVariable(variablesToRemoveIndex); 

  if (v == InternalThreadLocalMap.UNSET || v == null) {

   return;

  }

  Set<FastThreadLocal<?>> variablesToRemove = (Set<FastThreadLocal<?>>) v;

  variablesToRemove.remove(variable);

 }

 

 // 直接移除操作，可以看到直接获取当前线程的InternalThreadLocalMap对象，然后调用上述的remove(InternalThreadLocalMap threadLocalMap) 方法

 public final void remove() {

  remove(InternalThreadLocalMap.getIfSet());

 }

可能读者会想：下标为0处的集合到底有何用？为何在当前线程添加映射时需要将FastThreadLocal放入该集合。我们继续看removeAll静态方法的移除操作。根据源码我们得出结论：

1. 我们在当前线程中可以创建多个FastThreadLocal，而一个线程只有一个InternalThreadLocalMap
2. 当我们在添加FastThreadLocal映射值时，就会将其放入InternalThreadLocalMap下标为0处的set集合
3. 这时当我们需要移除该线程所有的FastThreadLocal映射时，就可以获取该集合将其中的保存的InternalThreadLocalMap的值全部移除

public class FastThreadLocal<V> {

 public static void removeAll() {

  InternalThreadLocalMap threadLocalMap = InternalThreadLocalMap.getIfSet(); // 获取当前线程的InternalThreadLocalMap

  if (threadLocalMap == null) {

   return;

  }

  try {

   Object v = threadLocalMap.indexedVariable(variablesToRemoveIndex); // 获取下标为0的set集合，该集合保留了当前线程所有的FastThreadLocal变量

   if (v != null && v != InternalThreadLocalMap.UNSET) { // 集合存在，那么遍历调用FastThreadLocal的remove方法移除

    @SuppressWarnings("unchecked")

    Set<FastThreadLocal<?>> variablesToRemove = (Set<FastThreadLocal<?>>) v;

    FastThreadLocal<?>[] variablesToRemoveArray =

     variablesToRemove.toArray(new FastThreadLocal[variablesToRemove.size()]);

    for (FastThreadLocal<?> tlv: variablesToRemoveArray) {

     tlv.remove(threadLocalMap);

    }

   }

  } finally { // 最后调用InternalThreadLocalMap的remove静态方法，完成最终的清理操作（其实就是将FastThreadLocalThread的InternalThreadLocalMap threadLocalMap置空，因为这个map已经无用了）

   InternalThreadLocalMap.remove();

  }

 }

}

FastThreadLocal的核心是InternalThreadLocalMap，我们继续来看InternalThreadLocalMap的原理。根据源码我们得出以下结论：

1. InternalThreadLocalMap继承自UnpaddedInternalThreadLocalMap，对于保存index和value的核心变量Object[] indexedVariables在UnpaddedInternalThreadLocalMap中定义
2. 我们知道保存变量的核心为InternalThreadLocalMap，那么这时我们可以知道FastThreadLocalThread中保存有InternalThreadLocalMap的引用，那么这时移除和获取都可以直接通过该引用操作。那么如果是普通Thread线程呢？那么我们也可以兼容：只需要将InternalThreadLocalMap放入到普通的Thread类的ThreadLocalMap即可。这时我们在UnpaddedInternalThreadLocalMap中定义了ThreadLocal<InternalThreadLocalMap> slowThreadLocalMap来完成该操作

class UnpaddedInternalThreadLocalMap {

 static final ThreadLocal<InternalThreadLocalMap> slowThreadLocalMap = new ThreadLocal<InternalThreadLocalMap>(); // 兼容原生的JDK ThreadLocal ，注意这里的类型为InternalThreadLocalMap，这是何意？继续往下看



 static final AtomicInteger nextIndex = new AtomicInteger(); // 原子性生成下一个索引下标



 Object[] indexedVariables; // 核心变量，用于保存ThreadLocal的变量值（还记得前面的FastThreadLocal的index下标么）



 UnpaddedInternalThreadLocalMap(Object[] indexedVariables) { 

  this.indexedVariables = indexedVariables;

 }

}



public final class InternalThreadLocalMap extends UnpaddedInternalThreadLocalMap {

 public static final Object UNSET = new Object(); // 表明未使用的InternalThreadLocalMap下标占位符



 // 构造器初始化父类的Object[] indexedVariables数组，可以看到默认数组长度为32

 private InternalThreadLocalMap() { 

  super(newIndexedVariableTable());

 }



 private static Object[] newIndexedVariableTable() {

  Object[] array = new Object[32];

  Arrays.fill(array, UNSET);

  return array;

 }



 // 获取当前线程所属的InternalThreadLocalMap对象。根据当前线程类型是否为FastThreadLocalThread，从而调用fastGet或者slowGet获取InternalThreadLocalMap。因为只有FastThreadLocalThread内部直接拥有InternalThreadLocalMap引用。

 public static InternalThreadLocalMap get() {

  Thread thread = Thread.currentThread();

  if (thread instanceof FastThreadLocalThread) { 

   return fastGet((FastThreadLocalThread) thread);

  } else {

   return slowGet();

  }

 }



 // 从当前FastThreadLocalThread属性变量中获取InternalThreadLocalMap对象，如果不存在，那么在这里将会进行初始化

 private static InternalThreadLocalMap fastGet(FastThreadLocalThread thread) {

  InternalThreadLocalMap threadLocalMap = thread.threadLocalMap();

  if (threadLocalMap == null) {

   thread.setThreadLocalMap(threadLocalMap = new InternalThreadLocalMap());

  }

  return threadLocalMap;

 }



 // 当前线程不是FastThreadLocalThread，那么从线程原生的ThreadLocalMap中获取InternalThreadLocalMap，当然，如果这里不存在，那么将会初始化InternalThreadLocalMap对象，然后再将其放入原生的ThreadLocal中

 private static InternalThreadLocalMap slowGet() {

  ThreadLocal<InternalThreadLocalMap> slowThreadLocalMap = UnpaddedInternalThreadLocalMap.slowThreadLocalMap;

  InternalThreadLocalMap ret = slowThreadLocalMap.get();

  if (ret == null) {

   ret = new InternalThreadLocalMap();

   slowThreadLocalMap.set(ret);

  }

  return ret;

 }



 // 将当前线程中的InternalThreadLocalMap移除。这里同样分为FastThreadLocalThread和普通线程。如果是FastThreadLocalThread那么直接设置保存的引用为null即可，如果是普通Thread，那么直接调用原生ThreadLocal的remove方法即可

 public static void remove() {

  Thread thread = Thread.currentThread();

  if (thread instanceof FastThreadLocalThread) {

   ((FastThreadLocalThread) thread).setThreadLocalMap(null);

  } else {

   slowThreadLocalMap.remove();

  }

 }



}

小结

我们这里主要对比下普通的ThreadLocal和FastThreadLocal的原理：

1. 普通的ThreadLocalMap在存储和移除ThreadLocal和Value值的映射时需要进行hash取余操作
2. FastThreadLocal在初始化对象时就保留了一个index下标，这时存储和移除时就不需要取余，只需要通过保存的index下标获取即可，这就是为什么叫做FastThreadLocal

shutdownGracefully方法原理

该方法将由外部线程调用，用于结束执行器组中的子执行器。我们看到该方法的实现较为简单，仅仅遍历了子执行器数组，然后调用他们的shutdownGracefully方法，通过返回terminationFuture对象，我们在前面看到过该对象将会在所有子执行器完成关闭后，设置为完成状态。源码描述如下。

public Future<?> shutdownGracefully(long quietPeriod, long timeout, TimeUnit unit) {

 for (EventExecutor l: children) {

  l.shutdownGracefully(quietPeriod, timeout, unit);

  return terminationFuture();

 }

}



public Future<?> terminationFuture() { // 返回terminationFuture实例

 return terminationFuture;

}

awaitTermination方法原理

该方法用于同步等待所有子执行器全部终止。timeout参数用于指定等待时长。我们看到实现过程如下：

1. 根据超时时间和单位计算出基于当前时间退出等待的截止时间deadline
2. 遍历所有的子执行器，并调用他们的awaitTermination方法等待，在调用子类方法过程中，对等待的时间进行修正，因为每次等待过后将会消耗一段等待时间，所以需要减掉该时间
3. 等待过程中对deadline进行检测，如果发现超时，那么将退出等待
4. 调用isTerminated方法返回子执行器是否已经终止

public boolean awaitTermination(long timeout, TimeUnit unit)

 throws InterruptedException {

 long deadline = System.nanoTime() + unit.toNanos(timeout);

 loop: for (EventExecutor l: children) {

  for (;;) {

   long timeLeft = deadline - System.nanoTime();

   if (timeLeft <= 0) {

    break loop;

   }

   if (l.awaitTermination(timeLeft, TimeUnit.NANOSECONDS)) {

    break;

   }

  }

 }

 return isTerminated();

}



public boolean isTerminated() { // 判断子执行器是否终止，可以看到遍历自执行器数组，如果有一个未终止，那么均返回false

 for (EventExecutor l: children) {

  if (!l.isTerminated()) {

   return false;

  }

 }

 return true;

}