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

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

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

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

而InternalThreadLocalMap是UnpaddedInternalThreadLocalMap的子类,UnpaddedInternalThreadLocalMap的定义比较简单:

 1 class UnpaddedInternalThreadLocalMap {
 2     static final ThreadLocal<InternalThreadLocalMap> slowThreadLocalMap = new ThreadLocal();
 3     static final AtomicInteger nextIndex = new AtomicInteger();
 4     Object[] indexedVariables;
 5     int futureListenerStackDepth;
 6     int localChannelReaderStackDepth;
 7     Map<Class<?>, Boolean> handlerSharableCache;
 8     IntegerHolder counterHashCode;
 9     ThreadLocalRandom random;
10     Map<Class<?>, TypeParameterMatcher> typeParameterMatcherGetCache;
11     Map<Class<?>, Map<String, TypeParameterMatcher>> typeParameterMatcherFindCache;
12     StringBuilder stringBuilder;
13     Map<Charset, CharsetEncoder> charsetEncoderCache;
14     Map<Charset, CharsetDecoder> charsetDecoderCache;
15     ArrayList<Object> arrayList;
16
17     UnpaddedInternalThreadLocalMap(Object[] indexedVariables) {
18         this.indexedVariables = indexedVariables;
19     }
20 }

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

InternalThreadLocalMap的定义如下:

1 public final class InternalThreadLocalMap extends UnpaddedInternalThreadLocalMap {
2     private static final InternalLogger logger = InternalLoggerFactory.getInstance(InternalThreadLocalMap.class);
3     private static final int DEFAULT_ARRAY_LIST_INITIAL_CAPACITY = 8;
4     private static final int STRING_BUILDER_INITIAL_SIZE = SystemPropertyUtil.getInt("io.netty.threadLocalMap.stringBuilder.initialSize", 1024);
5     private static final int STRING_BUILDER_MAX_SIZE;
6     public static final Object UNSET = new Object();
7     private BitSet cleanerFlags;

InternalThreadLocalMap的nextVariableIndex方法:

1 public static int nextVariableIndex() {
2     int index = nextIndex.getAndIncrement();
3     if (index < 0) {
4         nextIndex.decrementAndGet();
5         throw new IllegalStateException("too many thread-local indexed variables");
6     } else {
7         return index;
8     }
9 }

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

FastThreadLocal对象的初始化:

1 private final int index = InternalThreadLocalMap.nextVariableIndex();
2
3 public FastThreadLocal() {
4 }

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

首先看到set方法:

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

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

1 public static InternalThreadLocalMap get() {
2     Thread thread = Thread.currentThread();
3     return thread instanceof FastThreadLocalThread ? fastGet((FastThreadLocalThread)thread) : slowGet();
4 }

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

 1 public class FastThreadLocalThread extends Thread {
 2     private final boolean cleanupFastThreadLocals;
 3     private InternalThreadLocalMap threadLocalMap;
 4
 5     public FastThreadLocalThread() {
 6         this.cleanupFastThreadLocals = false;
 7     }
 8
 9     public FastThreadLocalThread(Runnable target) {
10         super(FastThreadLocalRunnable.wrap(target));
11         this.cleanupFastThreadLocals = true;
12     }
13
14     public FastThreadLocalThread(ThreadGroup group, Runnable target) {
15         super(group, FastThreadLocalRunnable.wrap(target));
16         this.cleanupFastThreadLocals = true;
17     }
18
19     public FastThreadLocalThread(String name) {
20         super(name);
21         this.cleanupFastThreadLocals = false;
22     }
23
24     public FastThreadLocalThread(ThreadGroup group, String name) {
25         super(group, name);
26         this.cleanupFastThreadLocals = false;
27     }
28
29     public FastThreadLocalThread(Runnable target, String name) {
30         super(FastThreadLocalRunnable.wrap(target), name);
31         this.cleanupFastThreadLocals = true;
32     }
33
34     public FastThreadLocalThread(ThreadGroup group, Runnable target, String name) {
35         super(group, FastThreadLocalRunnable.wrap(target), name);
36         this.cleanupFastThreadLocals = true;
37     }
38
39     public FastThreadLocalThread(ThreadGroup group, Runnable target, String name, long stackSize) {
40         super(group, FastThreadLocalRunnable.wrap(target), name, stackSize);
41         this.cleanupFastThreadLocals = true;
42     }
43
44     public final InternalThreadLocalMap threadLocalMap() {
45         return this.threadLocalMap;
46     }
47
48     public final void setThreadLocalMap(InternalThreadLocalMap threadLocalMap) {
49         this.threadLocalMap = threadLocalMap;
50     }
51
52     public boolean willCleanupFastThreadLocals() {
53         return this.cleanupFastThreadLocals;
54     }
55
56     public static boolean willCleanupFastThreadLocals(Thread thread) {
57         return thread instanceof FastThreadLocalThread && ((FastThreadLocalThread)thread).willCleanupFastThreadLocals();
58     }
59 }

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

fastGet方法:

1 private static InternalThreadLocalMap fastGet(FastThreadLocalThread thread) {
2    InternalThreadLocalMap threadLocalMap = thread.threadLocalMap();
3     if (threadLocalMap == null) {
4         thread.setThreadLocalMap(threadLocalMap = new InternalThreadLocalMap());
5     }
6
7     return threadLocalMap;
8 }

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

1 private InternalThreadLocalMap() {
2     super(newIndexedVariableTable());
3 }

先调用newIndexedVariableTable方法:

1 private static Object[] newIndexedVariableTable() {
2     Object[] array = new Object[32];
3     Arrays.fill(array, UNSET);
4     return array;
5 }

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

再来看slowGet方法:

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

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

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

1 private boolean setKnownNotUnset(InternalThreadLocalMap threadLocalMap, V value) {
2     if (threadLocalMap.setIndexedVariable(this.index, value)) {
3         addToVariablesToRemove(threadLocalMap, this);
4         return true;
5     } else {
6         return false;
7     }
8 }

首先调用了InternalThreadLocalMap的setIndexedVariable方法:

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

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

expandIndexedVariableTableAndSet方法:

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

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

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

setIndexedVariable成立后,setKnownNotUnset继续调用addToVariablesToRemove方法:

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

上面说过variablesToRemoveIndex恒为0,调用InternalThreadLocalMap的indexedVariable方法:

1 public Object indexedVariable(int index) {
2     Object[] lookup = this.indexedVariables;
3     return index < lookup.length ? lookup[index] : UNSET;
4 }

由于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集合
那么就有如下关系:

Netty中FastThreadLocal源码分析-LMLPHP

回到FastThreadLocal的set方法,在setKnownNotUnset成立后,调用registerCleaner方法:

1 private void registerCleaner(InternalThreadLocalMap threadLocalMap) {
2     Thread current = Thread.currentThread();
3     if (!FastThreadLocalThread.willCleanupFastThreadLocals(current) && !threadLocalMap.isCleanerFlagSet(this.index)) {
4         threadLocalMap.setCleanerFlag(this.index);
5     }
6 }

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

1 public boolean isCleanerFlagSet(int index) {
2     return this.cleanerFlags != null && this.cleanerFlags.get(index);
3 }

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

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

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

1 public void setCleanerFlag(int index) {
2     if (this.cleanerFlags == null) {
3         this.cleanerFlags = new BitSet();
4     }
5
6     this.cleanerFlags.set(index);
7 }

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

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

下来看FastThreadLocal的get方法:

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

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

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

initialValue()方法就是对外提供的,需要手动覆盖:

1 protected V initialValue() throws Exception {
2     return null;
3 }

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

remove方法:

1 public final void remove() {
2     this.remove(InternalThreadLocalMap.getIfSet());
3 }

getIfSet方法:

1 public static InternalThreadLocalMap getIfSet() {
2     Thread thread = Thread.currentThread();
3     return thread instanceof FastThreadLocalThread ? ((FastThreadLocalThread)thread).threadLocalMap() : (InternalThreadLocalMap)slowThreadLocalMap.get();
4 }

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

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

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

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

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

接着调用removeFromVariablesToRemove方法:

1 private static void removeFromVariablesToRemove(InternalThreadLocalMap threadLocalMap, FastThreadLocal<?> variable) {
2     Object v = threadLocalMap.indexedVariable(variablesToRemoveIndex);
3     if (v != InternalThreadLocalMap.UNSET && v != null) {
4         Set<FastThreadLocal<?>> variablesToRemove = (Set)v;
5         variablesToRemove.remove(variable);
6     }
7 }

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

1 protected void onRemoval(V value) throws Exception {
2 }


removeAll方法,是一个静态方法:

 1 public static void removeAll() {
 2     InternalThreadLocalMap threadLocalMap = InternalThreadLocalMap.getIfSet();
 3     if (threadLocalMap != null) {
 4         try {
 5             Object v = threadLocalMap.indexedVariable(variablesToRemoveIndex);
 6             if (v != null && v != InternalThreadLocalMap.UNSET) {
 7                 Set<FastThreadLocal<?>> variablesToRemove = (Set)v;
 8                 FastThreadLocal<?>[] variablesToRemoveArray = (FastThreadLocal[])variablesToRemove.toArray(new FastThreadLocal[0]);
 9                 FastThreadLocal[] var4 = variablesToRemoveArray;
10                 int var5 = variablesToRemoveArray.length;
11
12                 for(int var6 = 0; var6 < var5; ++var6) {
13                     FastThreadLocal<?> tlv = var4[var6];
14                     tlv.remove(threadLocalMap);
15                 }
16             }
17         } finally {
18             InternalThreadLocalMap.remove();
19         }
20
21     }
22 }

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

FastThreadLocal源码分析到此结束。

06-03 02:51