FastThreadLocal
前面介绍过 JDK 的 ThreadLocal , 使用不当的话容易造成内存泄漏最终导致OOM, 并且也有一些地方设计的不够好(相对于接下来要介绍的 FastThreadLocal), 接下来我们就介绍一下 Netty 改进的 FastThreadLocal, 看它到底 Fast 在哪里.
(JDK 的 ThreadLocal 的地址: )
同样的, 这回我们根据 FastThreadLocal 的源码对其进行分析.
FastThreadLocal#构造方法
FastThreadLocal 有一个标记自己下标的 index , 表明当前 FastThreadLocal 在 InternalThreadLocalMap 存储数据的数组中(Object[] indexedVariables)所处的下标.
// 位于 map 中的下标 private final int index; public FastThreadLocal() { index = InternalThreadLocalMap.nextVariableIndex(); } 跟踪 InternalThreadLocalMap.nextVariableIndex(); 的实现可以看到:
public static int nextVariableIndex() { // nextIndex 见下面 int index = nextIndex.getAndIncrement(); // 整数的最大值+1就变成了负数, 不过一般也不会用这么多的 ThreadLocal if (index < 0) { nextIndex.decrementAndGet(); throw new IllegalStateException("too many thread-local indexed variables"); } return index; } // 这是个原子变量, 可以根据这个变量获取当前 FastThreadLocal 下标, 因为这是递增的(nextIndex.getAndIncrement()), 所以不会出现多个 FastThreadLocal 下标相同, 即 FastThreadLocal 的下标唯一. static final AtomicInteger nextIndex = new AtomicInteger(); //todo: 扩容的时候, ThreadLocal 根据 hash 值取余长度计算下标, 可能会导致下标冲突, 需要循环往后查找空的位置放置. FastThreadLocal 直接复制以前的部分, 扩容出来的直接设置初始值, 不用加多一层循环去判断是否为空(可以设置进去), 这就是 唯一的 index 的好处, 不会导致冲突.
FastThreadLocal#set()
public final void set(V value) { if (value != InternalThreadLocalMap.UNSET) { // 获取当前线程的 threadLocalMap InternalThreadLocalMap threadLocalMap = InternalThreadLocalMap.get(); // 如果是新添加进来的话,则需要注册一个清理器 if (setKnownNotUnset(threadLocalMap, value)) { // 注册清理器 registerCleaner(threadLocalMap); } } else { remove(); } } private boolean setKnownNotUnset(InternalThreadLocalMap threadLocalMap, V value) { // 返回true的话表示是新添加的 ThreadLocal if (threadLocalMap.setIndexedVariable(index, value)) { // 则添加进需要 remove 的 集合(set)中 addToVariablesToRemove(threadLocalMap, this); return true; } return false; } // 根据下标设置值, index 为 FastThreadLocal 的 唯一index public boolean setIndexedVariable(int index, Object value) { // 获取到所有存储的 FastThreadLocal Object[] lookup = indexedVariables; // 下标越界判断 if (index < lookup.length) { Object oldValue = lookup[index]; lookup[index] = value; // 只有添加了新的 ThreadLocal 才会返回 true return oldValue == UNSET; } else { // 超过了 map 的大小则进行扩容,扩容见后面的代码 expandIndexedVariableTableAndSet(index, value); return true; } } 可以看到, FastThreadLocal#set 的时候直接根据原子变量获取最新的 index , 然后直接设置进去.
FastThreadLocal#get()
get的过程比较简单,就不多赘述了
public final V get() { // 当前 thread 的 threadLocalMap InternalThreadLocalMap threadLocalMap = InternalThreadLocalMap.get(); // value 值根据 new 的时候的 index 来获取 Object v = threadLocalMap.indexedVariable(index); if (v != InternalThreadLocalMap.UNSET) { return (V) v; } V value = initialize(threadLocalMap); registerCleaner(threadLocalMap); return value; } InternalThreadLocalMap#expandIndexedVariableTableAndSet
这是 ThreadLocalMap 中的一个扩容方法,一共有3步操作:
1.申请一个新数组,大小为原来的两倍
2.copy数据到新数组(浅拷贝)并且初始化新增部分
3.设置map中新的数组
// 对 Object[] 进行扩容 private void expandIndexedVariableTableAndSet(int index, Object value) { // 旧的 Object[] Object[] oldArray = indexedVariables; final int oldCapacity = oldArray.length; // index * 2 int newCapacity = index; newCapacity |= newCapacity >>> 1; newCapacity |= newCapacity >>> 2; newCapacity |= newCapacity >>> 4; newCapacity |= newCapacity >>> 8; newCapacity |= newCapacity >>> 16; newCapacity ++; // 进行浅拷贝 Object[] newArray = Arrays.copyOf(oldArray, newCapacity); // 初始化后面新申请的元素 newArray[ oldCapacity , newArray.length ) Arrays.fill(newArray, oldCapacity, newArray.length, UNSET); // 设置刚加进来的值 newArray[index] = value; // 设置新数组 indexedVariables = newArray; } 接下来看一下 JDK 中的 ThreadLocal 中的扩容方法, 也把它当成三步来看吧
1.申请新数组,大小为原来的两倍
2.将数据放入新的数组( hash % (newLen -1))
3.设置map中新的数组
几个步骤看起来是差不多, 主要的不同就是在第二步, JDK 中计算下标的位置是 hash % (newLen -1), 用的是 hash值取余, 会出现冲突, 就像 HashMap 从头节点一直找到链表的最后一个节点(如果是树的话就找到相应大小的地方), 冲突后就循环查找, 这里就是 JDK 的 ThreadLocal 耗时的地方.
private void resize() { Entry[] oldTab = table; int oldLen = oldTab.length; // double size int newLen = oldLen * 2; Entry[] newTab = new Entry[newLen]; int count = 0; for (int j = 0; j < oldLen; ++j) { Entry e = oldTab[j]; if (e != null) { ThreadLocal k = e.get(); if (k == null) { e.value = null; // Help the GC } else { // 这里是数据位于数组中的下标 int h = k.threadLocalHashCode & (newLen - 1); // 直到找到空的Entry, 才设置进去, 如果原来的 Entry 已经有了, 需要一直循环往后查找空的位置 while (newTab[h] != null) h = nextIndex(h, newLen); newTab[h] = e; count++; } } } setThreshold(newLen); size = count; table = newTab; } Netty 的 FastThreadLocal 并不会像 JDK 的 ThreadLocal 那样会出现下标冲突和循环里查找, 这是 FastThreadLocal --> Fast 的其中重要原因.