netty5笔记-concurrent-FastThreadLocal

深藏阁楼爱情的钟 2022-05-28 01:38 181阅读 0赞

并发的一个很大的朋友就是锁，因为锁可以在并发的情况下保护一些公共的变量；并发的一个很大的敌人也是锁，因为锁带来了很大的性能开销。

很多时候我们必须加锁来应对并发带来的线程安全问题，而ThreadLocal则给我们开辟了另外一个思路，将数据保存在线程的私有字段中，使一个线程无法读到其他线程的数据， 这样的数据存储方式，自然就线程安全。 那么ThreadLocal本身还有没有可以挖掘的优化点呢，netty告诉你，有！

我们知道ThreadLocal的实现是在Thread里放一个类似map（虽然叫map，但实际上是纯数组实现）的数据结构来存数据，而这个自定义的map要查找一个元素主要分成两步：1、通过元素的hashcode计算得到对应的index，判断对应index上是否有数据，如果有数据且key==yourThreadLocal则匹配成功；2、否则按顺序依次挪到下一个位置进行对比，直到找到yourThreadLocal。 这样查询一个元素可能需要很多步。同样插入或移除一个值，操作步数也比较多。

下面来看看netty君是怎么做的，首先基本思路和jdk的ThreadLocal相同，在一个线程上增加一个私有字段存数据。然而原有的ThreadLocal并没有提供定制的接口，咋办呢？netty继承Thread实现了一个FastThreadLocalThread，该FastThreadLocalThread实现了FastThreadLocalAccess接口：

**\[java\]** [view plain][] [copy][view plain]

1.  **public** **interface** FastThreadLocalAccess \{  
2.      InternalThreadLocalMap threadLocalMap();  
3.      **void** setThreadLocalMap(InternalThreadLocalMap threadLocalMap);  
4.  \}

该接口暴露两个方法用来获取和设置InternalThreadLocalMap ， InternalThreadLocalMap就是数据存储的实现了。 从构造方法可以看到InternalThreadLocalMap与ThreadLocalMap一样都采取了数组的方式存储元素。

**\[java\]** [view plain][] [copy][view plain]

1.  **private** InternalThreadLocalMap() \{  
2.      **super**(newIndexedVariableTable());  
3.  \}  
4.    
5.  **private** **static** Object\[\] newIndexedVariableTable() \{  
6.      Object\[\] array = **new** Object\[32\];  
7.      Arrays.fill(array, UNSET);  
8.      **return** array;  
9.  \}

InternalThreadLocalMap根据netty的实际情况还加入了一些其他的字段做一些定制化的的数据缓存，这样在ThreadLocal的基础上功能又增强了。不过话说回来ThreadLocal毕竟是公共的解决方案，不太可能放这些乱七八糟的东西。来看看这些netty引入的字段：

**\[java\]** [view plain][] [copy][view plain]

1.  **static** ThreadLocal<InternalThreadLocalMap> slowThreadLocalMap;  
2.  **static** **final** AtomicInteger nextIndex = **new** AtomicInteger();  
3.  /\*\* Used by \{@link FastThreadLocal\} \*/  
4.  Object\[\] indexedVariables;  
5.    
6.  // Core thread-locals  
7.  **int** futureListenerStackDepth;  
8.  **int** localChannelReaderStackDepth;  
9.  Map<Class<?>, Boolean> handlerSharableCache;  
10. IntegerHolder counterHashCode;  
11. ThreadLocalRandom random;  
12. Map<Class<?>, TypeParameterMatcher> typeParameterMatcherGetCache;  
13. Map<Class<?>, Map<String, TypeParameterMatcher>> typeParameterMatcherFindCache;  
14.   
15. // String-related thread-locals  
16. StringBuilder stringBuilder;  
17. Map<Charset, CharsetEncoder> charsetEncoderCache;  
18. Map<Charset, CharsetDecoder> charsetDecoderCache;

indexedVariables相关方法都是采用直接操作index的方式，效率极高：

**\[java\]** [view plain][] [copy][view plain]

1.  // 直接根据index获取  
2.  **public** Object indexedVariable(**int** index) \{  
3.      Object\[\] lookup = indexedVariables;  
4.      **return** index < lookup.length? lookup\[index\] : UNSET;  
5.  \}  
6.    
7.  /\*\* 
8.   \* @return \{@code true\} if and only if a new thread-local variable has been created 
9.   \*/  
10. **public** **boolean** setIndexedVariable(**int** index, Object value) \{  
11.     Object\[\] lookup = indexedVariables;  
12.     **if** (index < lookup.length) \{  
13.         Object oldValue = lookup\[index\];  
14.         lookup\[index\] = value;  
15.         **return** oldValue == UNSET;  
16.     \} **else** \{  
17.         // 如果要查询的index不在indexedVariables范围，则需要先扩展在设置              
18.         expandIndexedVariableTableAndSet(index, value);  
19.         **return** **true**;  
20.     \}  
21. \}  
22.   
23. **private** **void** expandIndexedVariableTableAndSet(**int** index, Object value) \{  
24.     Object\[\] oldArray = indexedVariables;  
25.     **final** **int** oldCapacity = oldArray.length;  
26.     // newCapacity-> 32,64,128,256,512....  
27.     **int** newCapacity = index;  
28.     newCapacity |= newCapacity >>>  1;  
29.     newCapacity |= newCapacity >>>  2;  
30.     newCapacity |= newCapacity >>>  4;  
31.     newCapacity |= newCapacity >>>  8;  
32.     newCapacity |= newCapacity >>> 16;  
33.     newCapacity ++;  
34.   
35.     Object\[\] newArray = Arrays.copyOf(oldArray, newCapacity);  
36.     Arrays.fill(newArray, oldCapacity, newArray.length, UNSET);  
37.     newArray\[index\] = value;  
38.     indexedVariables = newArray;  
39. \}  
40. // remove方法直接在对应位置上设置UNSET  
41. **public** Object removeIndexedVariable(**int** index) \{  
42.     Object\[\] lookup = indexedVariables;  
43.     **if** (index < lookup.length) \{  
44.         Object v = lookup\[index\];  
45.         lookup\[index\] = UNSET;  
46.         **return** v;  
47.     \} **else** \{  
48.         **return** UNSET;  
49.     \}  
50. \}  
51.   
52. **public** **boolean** isIndexedVariableSet(**int** index) \{  
53.     Object\[\] lookup = indexedVariables;  
54.     **return** index < lookup.length && lookup\[index\] != UNSET;  
55. \}

这段代码简单得紧，所有操作都是直接的index查找，这样的话netty的FastThreadLocal元素操作就只有一步。然而这个index是如何得来的？ 答案就是上面出现过的nextIndex，注意nextIndex是静态的，因为ThreadLocal正确的用法就是声明成static。 在整个进程中可能存在很多个ThreadLocal，对于jdk的ThreadLocal，识别ThreadLocal的方式是引用的对比，即key == yourThreadLocal的方式。而netty实现的FastThreadLocal则采用全局序列号的方式，采用AtomicInteger分配可以保证index的唯一性，同时index从0开始递增，符合数据的下标模式，因此FastThreadLocal采用的index就可以和数组完美的结合， 相比ThreadLocal查找一个元素还需要考虑冲突的情况来说，的确是更先进。

知道了上面这一段原理的介绍，后面的代码已经不重要了，不过我们还是来八一八吧，时间不多的同学可以选择不往下看了（虽然后面还有一个重要的点）。

先看看几个辅助的方法：

**\[java\]** [view plain][] [copy][view plain]

1.  // 该方法与get()不同的是,get()在查找不到数据是会进行初始化，而这个方法则是直接返回  
2.  **public** **static** InternalThreadLocalMap getIfSet() \{  
3.      Thread thread = Thread.currentThread();  
4.      InternalThreadLocalMap threadLocalMap;  
5.      **if** (thread **instanceof** FastThreadLocalAccess) \{  
6.          threadLocalMap = ((FastThreadLocalAccess) thread).threadLocalMap();  
7.      \} **else** \{  
8.          ThreadLocal<InternalThreadLocalMap> slowThreadLocalMap = UnpaddedInternalThreadLocalMap.slowThreadLocalMap;  
9.          **if** (slowThreadLocalMap == **null**) \{  
10.             threadLocalMap = **null**;  
11.         \} **else** \{  
12.             threadLocalMap = slowThreadLocalMap.get();  
13.         \}  
14.     \}  
15.     **return** threadLocalMap;  
16. \}  
17.   
18. **public** **static** InternalThreadLocalMap get() \{  
19.     Thread thread = Thread.currentThread();  
20.     **if** (thread **instanceof** FastThreadLocalAccess) \{  
21.         // 如果实现了FastThreadLocalAccess，则采用netty的算法  
22.         **return** fastGet((FastThreadLocalAccess) thread);  
23.     \} **else** \{  
24.         // 没有未实现FastThreadLocalAccess，则降级使用jdk的ThreadLocal来存放InternalThreadLocalMap  
25.         **return** slowGet();  
26.     \}  
27. \}  
28.   
29. **private** **static** InternalThreadLocalMap fastGet(FastThreadLocalAccess thread) \{  
30.     InternalThreadLocalMap threadLocalMap = thread.threadLocalMap();  
31.     **if** (threadLocalMap == **null**) \{  
32.         thread.setThreadLocalMap(threadLocalMap = **new** InternalThreadLocalMap());  
33.     \}  
34.     **return** threadLocalMap;  
35. \}  
36.   
37. **private** **static** InternalThreadLocalMap slowGet() \{  
38.     // slowGet采用了java的ThreadLocal来管理InternalThreadLocalMap；  
39.     // 整个方案相对于直接用ThreadLocal来说又多了一步InternalThreadLocalMap的查询，但差距很小。  
40.     // 好处是统一了FastThreadLocal的模型，同时上文中也提到InternalThreadLocalMap做了很多功能上的增强。  
41.     ThreadLocal<InternalThreadLocalMap> slowThreadLocalMap = UnpaddedInternalThreadLocalMap.slowThreadLocalMap;  
42.     **if** (slowThreadLocalMap == **null**) \{  
43.         UnpaddedInternalThreadLocalMap.slowThreadLocalMap =  
44.                 slowThreadLocalMap = **new** ThreadLocal<InternalThreadLocalMap>();  
45.     \}  
46.   
47.     InternalThreadLocalMap ret = slowThreadLocalMap.get();  
48.     **if** (ret == **null**) \{  
49.         ret = **new** InternalThreadLocalMap();  
50.         slowThreadLocalMap.set(ret);  
51.     \}  
52.     **return** ret;  
53. \}

看完了辅助方法，我们回到FastThreadLocal看看具体的实现：

**\[java\]** [view plain][] [copy][view plain]

1.  **private** **final** **int** index;  
2.    
3.  **public** FastThreadLocal() \{  
4.      // 初始化时分配一个全局唯一的index  
5.      index = InternalThreadLocalMap.nextVariableIndex();  
6.  \}  
7.    
8.  **public** **final** V get() \{  
9.      // InternalThreadLocalMap.get()获取到与当前线程关联的InternalThreadLocalMap, 通过该map来查询具体数据  
10.     **return** get(InternalThreadLocalMap.get());  
11. \}  
12.   
13. **public** **final** V get(InternalThreadLocalMap threadLocalMap) \{  
14.     // 直接采用index下标访问threadLocalMap中数组的指定位置元素  
15.     Object v = threadLocalMap.indexedVariable(index);  
16.     **if** (v != InternalThreadLocalMap.UNSET) \{  
17.         **return** (V) v;  
18.     \}  
19.     // 不存在值则初始化  
20.     **return** initialize(threadLocalMap);  
21. \}  
22.   
23. **private** V initialize(InternalThreadLocalMap threadLocalMap) \{  
24.     V v = **null**;  
25.     **try** \{  
26.         v = initialValue();  
27.     \} **catch** (Exception e) \{  
28.         PlatformDependent.throwException(e);  
29.     \}  
30.     // 初始化后再设置，下次就不用再初始化  
31.     threadLocalMap.setIndexedVariable(index, v);  
32.     addToVariablesToRemove(threadLocalMap, **this**);  
33.     **return** v;  
34. \}  
35.   
36. /\*\* 
37.  \* Set the value for the current thread. 
38.  \*/  
39. **public** **final** **void** set(V value) \{  
40.     **if** (value != InternalThreadLocalMap.UNSET) \{  
41.         set(InternalThreadLocalMap.get(), value);  
42.     \} **else** \{  
43.         remove();  
44.     \}  
45. \}  
46.   
47. /\*\* 
48.  \* Set the value for the specified thread local map. The specified thread local map must be for the current thread. 
49.  \*/  
50. **public** **final** **void** set(InternalThreadLocalMap threadLocalMap, V value) \{  
51.    // UNSET表示移除   
52.    **if** (value != InternalThreadLocalMap.UNSET) \{  
53.         **if** (threadLocalMap.setIndexedVariable(index, value)) \{  
54.             addToVariablesToRemove(threadLocalMap, **this**);  
55.         \}  
56.     \} **else** \{  
57.         remove(threadLocalMap);  
58.     \}  
59. \}  
60.   
61. /\*\* 
62.  \* Sets the value to uninitialized; a proceeding call to get() will trigger a call to initialValue(). 
63.  \*/  
64. **public** **final** **void** remove() \{  
65.     remove(InternalThreadLocalMap.getIfSet());  
66. \}  
67.   
68. /\*\* 
69.  \* Sets the value to uninitialized for the specified thread local map; 
70.  \* a proceeding call to get() will trigger a call to initialValue(). 
71.  \* The specified thread local map must be for the current thread. 
72.  \*/  
73. @SuppressWarnings("unchecked")  
74. **public** **final** **void** remove(InternalThreadLocalMap threadLocalMap) \{  
75.     **if** (threadLocalMap == **null**) \{  
76.         **return**;  
77.     \}  
78.     // 直接使用下标进行remove  
79.     Object v = threadLocalMap.removeIndexedVariable(index);  
80.     removeFromVariablesToRemove(threadLocalMap, **this**);  
81.   
82.     **if** (v != InternalThreadLocalMap.UNSET) \{  
83.         **try** \{  
84.             onRemoval((V) v);  
85.         \} **catch** (Exception e) \{  
86.             PlatformDependent.throwException(e);  
87.         \}  
88.     \}  
89. \}

上面的方法都比较简单，不用再详细的解释，需要注意的是addToVariablesToRemove、removeFromVariablesToRemove的实现：

**\[java\]** [view plain][] [copy][view plain]

1.  **private** **static** **final** **int** variablesToRemoveIndex = InternalThreadLocalMap.nextVariableIndex();  
2.  **private** **static** **void** addToVariablesToRemove(InternalThreadLocalMap threadLocalMap, FastThreadLocal<?> variable) \{  
3.      Object v = threadLocalMap.indexedVariable(variablesToRemoveIndex);  
4.      Set<FastThreadLocal<?>> variablesToRemove;  
5.      **if** (v == InternalThreadLocalMap.UNSET || v == **null**) \{  
6.          variablesToRemove = Collections.newSetFromMap(**new** IdentityHashMap<FastThreadLocal<?>, Boolean>());  
7.          threadLocalMap.setIndexedVariable(variablesToRemoveIndex, variablesToRemove);  
8.      \} **else** \{  
9.          variablesToRemove = (Set<FastThreadLocal<?>>) v;  
10.     \}  
11.   
12.     variablesToRemove.add(variable);  
13. \}  
14.   
15. **private** **static** **void** removeFromVariablesToRemove(  
16.         InternalThreadLocalMap threadLocalMap, FastThreadLocal<?> variable) \{  
17.   
18.     Object v = threadLocalMap.indexedVariable(variablesToRemoveIndex);  
19.   
20.     **if** (v == InternalThreadLocalMap.UNSET || v == **null**) \{  
21.         **return**;  
22.     \}  
23.   
24.     @SuppressWarnings("unchecked")  
25.     Set<FastThreadLocal<?>> variablesToRemove = (Set<FastThreadLocal<?>>) v;  
26.     variablesToRemove.remove(variable);  
27. \}

从上面的代码可以看到variablesToRemoveIndex占用了一个下标， 该下标的元素是一个包装了IndentityHashMap的Set， 每次FastThreadLocal中设置值的时候将自己加到该Set，移除值的时候将自己从该Set移除。也就是说初始值、设置值、删除值这几个功能的耗时比有值时的get更多。那么为什么要加这个功能？来看看该Set的使用场景：

**\[java\]** [view plain][] [copy][view plain]

1.  /\*\* 
2.   \* 移除绑定到该线程的所有变量. 当不希望保留你无法管理的本地变量时使用该方法做清理。 
3.   \*/  
4.  **public** **static** **void** removeAll() \{  
5.      InternalThreadLocalMap threadLocalMap = InternalThreadLocalMap.getIfSet();  
6.      **if** (threadLocalMap == **null**) \{  
7.          **return**;  
8.      \}  
9.    
10.     **try** \{  
11.         Object v = threadLocalMap.indexedVariable(variablesToRemoveIndex);  
12.         **if** (v != **null** && v != InternalThreadLocalMap.UNSET) \{  
13.             // 取出Set中的所有FastThreadLocal进行清理  
14.             Set<FastThreadLocal<?>> variablesToRemove = (Set<FastThreadLocal<?>>) v;  
15.             FastThreadLocal<?>\[\] variablesToRemoveArray =  
16.                     variablesToRemove.toArray(**new** FastThreadLocal\[variablesToRemove.size()\]);  
17.             **for** (FastThreadLocal<?> tlv: variablesToRemoveArray) \{  
18.                 tlv.remove(threadLocalMap);  
19.             \}  
20.         \}  
21.     \} **finally** \{  
22.         InternalThreadLocalMap.remove();  
23.     \}  
24. \}

比如在web server中，classloader可能会开启热加载功能，在重新加载一个类的时候老的数据必须做清理，否则会造成内存泄露。 removeAll提供了这样一个清理数据的方式。 为了这个功能，FastThreadLocal可能会变得不是那么Fast，按我简单的想法，应该提供一个参数来打开它，这样只有需要的应用才打开，而默认情况下FastThreadLocal就真正的Fast了。然而其实这点性能的损失很小，是可以接受的。另一方面，如果你看了jdk的ThreadLocal.remove()，你会发现也! 很! 慢!

[view plain]: https://blog.csdn.net/youaremoon/article/details/50345651#