Netty中FastThreadLocal源码分析

Netty中使用FastThreadLocal替代JDK中的ThreadLocal【JAVA】ThreadLocal源码分析，其用法和ThreadLocal 一样，只不过从名字FastThreadLocal来看，其处理效率要比JDK中的ThreadLocal要高

在类加载的时候，先初始化了一个静态成员：

1 private static final int variablesToRemoveIndex = InternalThreadLocalMap.nextVariableIndex();

实际上FastThreadLocal的操作都是通过对InternalThreadLocalMap的操作来实现的，

而InternalThreadLocalMap是UnpaddedInternalThreadLocalMap的子类，UnpaddedInternalThreadLocalMap的定义比较简单：

1 1 class UnpaddedInternalThreadLocalMap {
2 2     static final ThreadLocal<InternalThreadLocalMap> slowThreadLocalMap = new ThreadLocal();
3 3     static final AtomicInteger nextIndex = new AtomicInteger();
4 4     Object[] indexedVariables;
5 5     int futureListenerStackDepth;
6 6     int localChannelReaderStackDepth;
7 7     Map<Class<?>, Boolean> handlerSharableCache;
8 8     IntegerHolder counterHashCode;
9 9     ThreadLocalRandom random;
1010     Map<Class<?>, TypeParameterMatcher> typeParameterMatcherGetCache;
1111     Map<Class<?>, Map<String, TypeParameterMatcher>> typeParameterMatcherFindCache;
1212     StringBuilder stringBuilder;
1313     Map<Charset, CharsetEncoder> charsetEncoderCache;
1414     Map<Charset, CharsetDecoder> charsetDecoderCache;
1515     ArrayList<Object> arrayList;
1616 
1717     UnpaddedInternalThreadLocalMap(Object[] indexedVariables) {
1818         this.indexedVariables = indexedVariables;
1919     }
2020 }

可以看到在类加载时，会初始化一个泛型为InternalThreadLocalMap的JDK的ThreadLocal对象作为其静态成员slowThreadLocalMap ，还有一个原子化的Integer静态成员nextIndex

InternalThreadLocalMap的定义如下：

11 public final class InternalThreadLocalMap extends UnpaddedInternalThreadLocalMap {
22     private static final InternalLogger logger = InternalLoggerFactory.getInstance(InternalThreadLocalMap.class);
33     private static final int DEFAULT_ARRAY_LIST_INITIAL_CAPACITY = 8;
44     private static final int STRING_BUILDER_INITIAL_SIZE = SystemPropertyUtil.getInt("io.netty.threadLocalMap.stringBuilder.initialSize", 1024);
55     private static final int STRING_BUILDER_MAX_SIZE;
66     public static final Object UNSET = new Object();
77     private BitSet cleanerFlags;

InternalThreadLocalMap的nextVariableIndex方法：

11 public static int nextVariableIndex() {
22     int index = nextIndex.getAndIncrement();
33     if (index < 0) {
44         nextIndex.decrementAndGet();
55         throw new IllegalStateException("too many thread-local indexed variables");
66     } else {
77         return index;
88     }
99 }

这是一个CAS滞后自增操作，获取nextIndex自增前的值，那么variablesToRemoveIndex初始化时就是0，且恒为0，nextIndex此时变成了1

FastThreadLocal对象的初始化：

11 private final int index = InternalThreadLocalMap.nextVariableIndex();
22 
33 public FastThreadLocal() {
44 }

由上面可知，index成员恒等于nextVariableIndex的返回值，nextIndex 的CAS操作保障了每个FastThreadLocal对象的index是不同的

首先看到set方法：

1 1 public final void set(V value) {
2 2     if (value != InternalThreadLocalMap.UNSET) {
3 3         InternalThreadLocalMap threadLocalMap = InternalThreadLocalMap.get();
4 4         if (this.setKnownNotUnset(threadLocalMap, value)) {
5 5             this.registerCleaner(threadLocalMap);
6 6         }
7 7     } else {
8 8         this.remove();
9 9     }
1010 
1111 }

只要set的value不是InternalThreadLocalMap.UNSET，会先调用InternalThreadLocalMap的get方法：

11 public static InternalThreadLocalMap get() {
22     Thread thread = Thread.currentThread();
33     return thread instanceof FastThreadLocalThread ? fastGet((FastThreadLocalThread)thread) : slowGet();
44 }

判断当前线程是否是FastThreadLocalThread，是则调用fastGet，否则调用slowGet
FastThreadLocalThread是经过包装后的Thread：

1 1 public class FastThreadLocalThread extends Thread {
2 2     private final boolean cleanupFastThreadLocals;
3 3     private InternalThreadLocalMap threadLocalMap;
4 4 
5 5     public FastThreadLocalThread() {
6 6         this.cleanupFastThreadLocals = false;
7 7     }
8 8 
9 9     public FastThreadLocalThread(Runnable target) {
1010         super(FastThreadLocalRunnable.wrap(target));
1111         this.cleanupFastThreadLocals = true;
1212     }
1313 
1414     public FastThreadLocalThread(ThreadGroup group, Runnable target) {
1515         super(group, FastThreadLocalRunnable.wrap(target));
1616         this.cleanupFastThreadLocals = true;
1717     }
1818 
1919     public FastThreadLocalThread(String name) {
2020         super(name);
2121         this.cleanupFastThreadLocals = false;
2222     }
2323 
2424     public FastThreadLocalThread(ThreadGroup group, String name) {
2525         super(group, name);
2626         this.cleanupFastThreadLocals = false;
2727     }
2828 
2929     public FastThreadLocalThread(Runnable target, String name) {
3030         super(FastThreadLocalRunnable.wrap(target), name);
3131         this.cleanupFastThreadLocals = true;
3232     }
3333 
3434     public FastThreadLocalThread(ThreadGroup group, Runnable target, String name) {
3535         super(group, FastThreadLocalRunnable.wrap(target), name);
3636         this.cleanupFastThreadLocals = true;
3737     }
3838 
3939     public FastThreadLocalThread(ThreadGroup group, Runnable target, String name, long stackSize) {
4040         super(group, FastThreadLocalRunnable.wrap(target), name, stackSize);
4141         this.cleanupFastThreadLocals = true;
4242     }
4343 
4444     public final InternalThreadLocalMap threadLocalMap() {
4545         return this.threadLocalMap;
4646     }
4747 
4848     public final void setThreadLocalMap(InternalThreadLocalMap threadLocalMap) {
4949         this.threadLocalMap = threadLocalMap;
5050     }
5151 
5252     public boolean willCleanupFastThreadLocals() {
5353         return this.cleanupFastThreadLocals;
5454     }
5555 
5656     public static boolean willCleanupFastThreadLocals(Thread thread) {
5757         return thread instanceof FastThreadLocalThread && ((FastThreadLocalThread)thread).willCleanupFastThreadLocals();
5858     }
5959 }

如果看过我之前写的ThreadLocal源码分析，看到这就明白，JDK的ThreadLocal中很重要的一点是在Thread类中有一个ThreadLocalMap类型的成员，每个线程都维护这一张ThreadLocalMap，通过ThreadLocalMap来和ThreadLocal对象产生映射关系；而这里和JDK同理绑定的就是InternalThreadLocalMap。

fastGet方法：

11 private static InternalThreadLocalMap fastGet(FastThreadLocalThread thread) {
22    InternalThreadLocalMap threadLocalMap = thread.threadLocalMap();
33     if (threadLocalMap == null) {
44         thread.setThreadLocalMap(threadLocalMap = new InternalThreadLocalMap());
55     }
66 
77     return threadLocalMap;
88 }

这里也和JDK的ThreadLocal类似，判断FastThreadLocalThread 线程的threadLocalMap成员是否为null，若是null，则先创建一个InternalThreadLocalMap实例：

11 private InternalThreadLocalMap() {
22     super(newIndexedVariableTable());
33 }

先调用newIndexedVariableTable方法：

11 private static Object[] newIndexedVariableTable() {
22     Object[] array = new Object[32];
33     Arrays.fill(array, UNSET);
44     return array;
55 }

创建了一个大小为32的数组，并且用UNSET这个Object填充了整个数组，然后调用UnpaddedInternalThreadLocalMap的构造，令indexedVariables成员保存该数组

再来看slowGet方法：

1 1 private static InternalThreadLocalMap slowGet() {
2 2     ThreadLocal<InternalThreadLocalMap> slowThreadLocalMap = UnpaddedInternalThreadLocalMap.slowThreadLocalMap;
3 3     InternalThreadLocalMap ret = (InternalThreadLocalMap)slowThreadLocalMap.get();
4 4     if (ret == null) {
5 5         ret = new InternalThreadLocalMap();
6 6         slowThreadLocalMap.set(ret);
7 7     }
8 8 
9 9     return ret;
1010 }

可以看到，其实这里为了提高效率，并没有直接使用JDK的ThreadLocal，而是给当前非FastThreadLocalThread线程绑定了一个ThreadLocal<InternalThreadLocalMap>对象，避免直接使用JDK的ThreadLocal效率低。

回到FastThreadLocal的set方法，在取得到了当前线程的InternalThreadLocalMap成员后，调用setKnownNotUnset方法：

11 private boolean setKnownNotUnset(InternalThreadLocalMap threadLocalMap, V value) {
22     if (threadLocalMap.setIndexedVariable(this.index, value)) {
33         addToVariablesToRemove(threadLocalMap, this);
44         return true;
55     } else {
66         return false;
77     }
88 }

首先调用了InternalThreadLocalMap的setIndexedVariable方法：

1 1 public boolean setIndexedVariable(int index, Object value) {
2 2     Object[] lookup = this.indexedVariables;
3 3     if (index < lookup.length) {
4 4         Object oldValue = lookup[index];
5 5         lookup[index] = value;
6 6         return oldValue == UNSET;
7 7     } else {
8 8         this.expandIndexedVariableTableAndSet(index, value);
9 9         return true;
1010     }
1111 }

因为index是不可更改的常量，所以这里有两种情况：
当indexedVariables这个Object数组的长度大于index时，直接将value放在indexedVariables数组下标为index的位置，返回oldValue是否等于UNSET，若是不等于UNSET，说明已经set过了，直进行替换，若是等于UNSET，还要进行后续的registerCleaner
当indexedVariables这个Object数组的长度小于等于index时，调用expandIndexedVariableTableAndSet方法扩容

expandIndexedVariableTableAndSet方法：

1 1 private void expandIndexedVariableTableAndSet(int index, Object value) {
2 2     Object[] oldArray = this.indexedVariables;
3 3     int oldCapacity = oldArray.length;
4 4     int newCapacity = index | index >>> 1;
5 5     newCapacity |= newCapacity >>> 2;
6 6     newCapacity |= newCapacity >>> 4;
7 7     newCapacity |= newCapacity >>> 8;
8 8     newCapacity |= newCapacity >>> 16;
9 9     ++newCapacity;
1010     Object[] newArray = Arrays.copyOf(oldArray, newCapacity);
1111     Arrays.fill(newArray, oldCapacity, newArray.length, UNSET);
1212     newArray[index] = value;
1313     this.indexedVariables = newArray;
1414 }

如果读过HashMap源码的话对上述的位运算操作因该不陌生，这个位运算产生的newCapacity的值是大于oldCapacity的最小的二的整数幂（【Java】HashMap中的tableSizeFor方法）

然后申请一个newCapacity大小的数组，将原数组的内容拷贝到新数组，并且用UNSET填充剩余部分，还是将value放在下标为index的位置，用indexedVariables保存新数组。

setIndexedVariable成立后，setKnownNotUnset继续调用addToVariablesToRemove方法：

1 1 private static void addToVariablesToRemove(InternalThreadLocalMap threadLocalMap, FastThreadLocal<?> variable) {
2 2     Object v = threadLocalMap.indexedVariable(variablesToRemoveIndex);
3 3     Set variablesToRemove;
4 4     if (v != InternalThreadLocalMap.UNSET && v != null) {
5 5         variablesToRemove = (Set)v;
6 6     } else {
7 7         variablesToRemove = Collections.newSetFromMap(new IdentityHashMap());
8 8         threadLocalMap.setIndexedVariable(variablesToRemoveIndex, variablesToRemove);
9 9     }
1010 
1111     variablesToRemove.add(variable);
1212 }

上面说过variablesToRemoveIndex恒为0，调用InternalThreadLocalMap的indexedVariable方法：

11 public Object indexedVariable(int index) {
22     Object[] lookup = this.indexedVariables;
33     return index < lookup.length ? lookup[index] : UNSET;
44 }

由于variablesToRemoveIndex恒等于0，所以这里判断indexedVariables这个Object数组是否为空，若是为空，则返回第0个元素，若不是则返回UNSET

在addToVariablesToRemove中，接着对indexedVariables的返回值进行了判断，
判断不是UNSET，并且不等于null，则说明是set过的，然后将刚才的返回值强转为Set类型
若上述条件不成立，创建一个IdentityHashMap，将其包装成Set赋值给variablesToRemove，然后调用InternalThreadLocalMap的setIndexedVariable方法，这里就和上面不一样了，上面是将value放在下标为index的位置，而这里是将Set放在下标为0的位置。

看到这，再结合上面来看，其实已经有一个大致的想法了，一开始在set时，是将value放在InternalThreadLocalMap的Object数组下标为index的位置，然后在这里获取下标为0的Set，说明value是暂时放在下标为index的位置，然后判断下标为0的位置有没有Set，若是有，取出这个Set ，将当前FastThreadLocal对象放入Set中，则说明这个Set中存放的是FastThreadLocal集合
那么就有如下关系：

回到FastThreadLocal的set方法，在setKnownNotUnset成立后，调用registerCleaner方法：

11 private void registerCleaner(InternalThreadLocalMap threadLocalMap) {
22     Thread current = Thread.currentThread();
33     if (!FastThreadLocalThread.willCleanupFastThreadLocals(current) && !threadLocalMap.isCleanerFlagSet(this.index)) {
44         threadLocalMap.setCleanerFlag(this.index);
55     }
66 }

willCleanupFastThreadLocals的返回值在前面FastThreadLocalThread的初始化时就确定了，看到isCleanerFlagSet方法：

11 public boolean isCleanerFlagSet(int index) {
22     return this.cleanerFlags != null && this.cleanerFlags.get(index);
33 }

cleanerFlags 是一个BitSet对象，在InternalThreadLocalMap初始化时是null，
若不是第一次的set操作，则根据index，获取index在BitSet对应位的值

这里使用BitSet，使其持有的位和indexedVariables这个Object数组形成了一一对应关系，每一位都是0和1代表当前indexedVariables的对应下标位置的使用情况，0表示没有使用对应UNSET，1则代表有value

在上面条件成立的情况下，调用setCleanerFlag方法：

11 public void setCleanerFlag(int index) {
22     if (this.cleanerFlags == null) {
33         this.cleanerFlags = new BitSet();
44     }
55 
66     this.cleanerFlags.set(index);
77 }

逻辑比较简单，判断cleanerFlags是否初始化，若没有，则立即初始化，再将cleanerFlags中对应index位的值设为1；

这里通过registerCleaner直接标记了所有set了value的下标可，为以后的removeAll 清除提高效率。

下来看FastThreadLocal的get方法：

1 1 public final V get() {
2 2     InternalThreadLocalMap threadLocalMap = InternalThreadLocalMap.get();
3 3     Object v = threadLocalMap.indexedVariable(this.index);
4 4     if (v != InternalThreadLocalMap.UNSET) {
5 5         return v;
6 6     } else {
7 7         V value = this.initialize(threadLocalMap);
8 8         this.registerCleaner(threadLocalMap);
9 9         return value;
1010     }
1111 }

和上面一样，先取得当前线程持有的InternalThreadLocalMap ，调用indexedVariable方法，根据当前FastThreadLocal的index定位，判断是否是UNSET（set过），若没有set过则和JDK一样调用initialize先set：

1 1 private V initialize(InternalThreadLocalMap threadLocalMap) {
2 2     Object v = null;
3 3 
4 4     try {
5 5         v = this.initialValue();
6 6     } catch (Exception var4) {
7 7         PlatformDependent.throwException(var4);
8 8     }
9 9 
1010     threadLocalMap.setIndexedVariable(this.index, v);
1111     addToVariablesToRemove(threadLocalMap, this);
1212     return v;
1313 }

initialValue（）方法就是对外提供的，需要手动覆盖：

11 protected V initialValue() throws Exception {
22     return null;
33 }

后面的操作就和set的逻辑一样。

remove方法：

11 public final void remove() {
22     this.remove(InternalThreadLocalMap.getIfSet());
33 }

getIfSet方法：

11 public static InternalThreadLocalMap getIfSet() {
22     Thread thread = Thread.currentThread();
33     return thread instanceof FastThreadLocalThread ? ((FastThreadLocalThread)thread).threadLocalMap() : (InternalThreadLocalMap)slowThreadLocalMap.get();
44 }

和上面的get方法思路相似，只不过在这里如果获取不到不会创建
然后调用remove重载：

1 1 public final void remove(InternalThreadLocalMap threadLocalMap) {
2 2     if (threadLocalMap != null) {
3 3         Object v = threadLocalMap.removeIndexedVariable(this.index);
4 4         removeFromVariablesToRemove(threadLocalMap, this);
5 5         if (v != InternalThreadLocalMap.UNSET) {
6 6             try {
7 7                 this.onRemoval(v);
8 8             } catch (Exception var4) {
9 9                 PlatformDependent.throwException(var4);
1010             }
1111         }
1212 
1313     }
1414 }

先检查threadLocalMap是否存在，若存在才进行后续操作：
调用removeIndexedVariable方法：

1 1 public Object removeIndexedVariable(int index) {
2 2     Object[] lookup = this.indexedVariables;
3 3     if (index < lookup.length) {
4 4         Object v = lookup[index];
5 5         lookup[index] = UNSET;
6 6         return v;
7 7     } else {
8 8         return UNSET;
9 9     }
1010 }

和之前的setIndexedVariable逻辑相似，只不过现在是把index位置的元素设置为UNSET

接着调用removeFromVariablesToRemove方法：

11 private static void removeFromVariablesToRemove(InternalThreadLocalMap threadLocalMap, FastThreadLocal<?> variable) {
22     Object v = threadLocalMap.indexedVariable(variablesToRemoveIndex);
33     if (v != InternalThreadLocalMap.UNSET && v != null) {
44         Set<FastThreadLocal<?>> variablesToRemove = (Set)v;
55         variablesToRemove.remove(variable);
66     }
77 }

之前说过variablesToRemoveIndex恒为0，在Object数组中下标为0存储的Set<FastThreadLocal<?>>集合（不为UNSET情况下），从集合中，将当前FastThreadLocal移除掉
最后调用了onRemoval方法，该方法需要由用户去覆盖：

11 protected void onRemoval(V value) throws Exception {
22 }

removeAll方法，是一个静态方法：

1 1 public static void removeAll() {
2 2     InternalThreadLocalMap threadLocalMap = InternalThreadLocalMap.getIfSet();
3 3     if (threadLocalMap != null) {
4 4         try {
5 5             Object v = threadLocalMap.indexedVariable(variablesToRemoveIndex);
6 6             if (v != null && v != InternalThreadLocalMap.UNSET) {
7 7                 Set<FastThreadLocal<?>> variablesToRemove = (Set)v;
8 8                 FastThreadLocal<?>[] variablesToRemoveArray = (FastThreadLocal[])variablesToRemove.toArray(new FastThreadLocal[0]);
9 9                 FastThreadLocal[] var4 = variablesToRemoveArray;
1010                 int var5 = variablesToRemoveArray.length;
1111 
1212                 for(int var6 = 0; var6 < var5; ++var6) {
1313                     FastThreadLocal<?> tlv = var4[var6];
1414                     tlv.remove(threadLocalMap);
1515                 }
1616             }
1717         } finally {
1818             InternalThreadLocalMap.remove();
1919         }
2020 
2121     }
2222 }

首先获取当前线程的InternalThreadLocalMap，若是存在继续后续操作：
通过indexedVariable方法，取出Object数组中下标为0的Set集合（如果不是UNSET情况下），将其转换为FastThreadLocal数组，遍历这个数组调用上面的remove方法。

FastThreadLocal源码分析到此结束。