Java高进进阶之FastThreadLocal源码详解（修复ThreadLocal的缺陷）_Java教程

Java高进进阶之FastThreadLocal源码详解（修复ThreadLocal的缺陷）

前言

ThreadLocal被ThreadLocalMap中的entry的key弱引用，如果出现GC的情况时，

没有被其他对象引用，会被回收，但是ThreadLocal对应的value却不会回收，容易造成内存泄漏，这也间接导致了内存溢出以及数据假丢失;

那么问题来了，有没有更高效的ThreadLocal有;

今天我们就来分析一波FastThreadLocalThread

一、FastThreadLocalThread源码分析

Netty为了在某些场景下提高性能，改进了jdk ThreadLocal，Netty实现的FastThreadLocal 优化了Java 原生 ThreadLocal 的访问速度，存储速度。避免了检测弱引用带来的 value 回收难问题，和数组位置冲突带来的线性查找问题，解决这些问题并不是没有代价;

Netty实现的 FastThreadLocal 底层也是通过数组存储 value 对象，与Java原生ThreadLocal使用自身作为Entry的key不同，FastThreadLocal通过保存数组的全局唯一下标，实现了对value的快速访问。同时FastThreadLocal 也实现了清理对象的方法;

1、FastThreadLocalThread

在Netty中，要使用 FastThreadLocal 实现线程本地变量需要将线程包装成 FastThreadLocalThread ，如果不是 FastThreadLocalThread ，会使用 slowThreadLocalMap的 ThreadLocal 来存储变量副本;

io.netty.util.concurrent.DefaultThreadFactory
@Override
public Thread newThread(Runnable r) {
Thread t = newThread(FastThreadLocalRunnable.wrap(r), prefix + nextId.incrementAndGet());
// 一般daemon为false，意思是不设置为守护线程
if (t.isDaemon() != daemon) {
t.setDaemon(daemon);
}
// 优先级默认为5
if (t.getPriority() != priority) {
t.setPriority(priority);
}
return t;
}
protected Thread newThread(Runnable r, String name) {
return new FastThreadLocalThread(threadGroup, r, name);
}

FastThreadLocalThread 继承自Thread类，有如下成员变量:

io.netty.util.concurrent.FastThreadLocalThread
// 任务执行完，是否清除FastThreadLocal的标记
private final boolean cleanupFastThreadLocals;
// 类似于Thread类中ThreadLocalMap，为了实现FastThreadLocal
private InternalThreadLocalMap threadLocalMap;

2、 InternalThreadLocalMap

Java高进进阶之FastThreadLocal源码详解（修复ThreadLocal的缺陷）

FastThreadLocalThread.threadLocalMap 是 InternalThreadLocalMap 对象实例。在第一次获取FTL数据时，会初始化FastThreadLocalThread.threadLocalMap，调用的构造函数如下：

private InternalThreadLocalMap() {
//为了简便，InternalThreadLocalMap父类
//UnpaddedInternalThreadLocalMap不展开介绍
super(newIndexedVariableTable());
}
//默认的数组大小为32，且使用UNSET对象填充数组
//如果下标处数据为UNSET，则表示没有数据
private static Object[] newIndexedVariableTable() {
Object[] array = new Object[32];
Arrays.fill(array, UNSET);
return array;
}

为了避免写时候影响同一cpu缓冲行的其他数据并发访问，其使用了缓存行填充技术 (cpu 缓冲行填充)，在类定义中声明了如下long字段进行填充;

//InternalThreadLocalMap
// Cache line padding (must be public)
// With CompressedOops enabled, an instance of this class should occupy at least 128 bytes.
public long rp1, rp2, rp3, rp4, rp5, rp6, rp7, rp8, rp9;

FTL使用的数组下标是InternalThreadLocalMap中的静态变量nextIndex统一递增生成的：

static final AtomicInteger nextIndex = new AtomicInteger();
public static int nextVariableIndex() {
//Netty中所有FTL数组下标都是通过递增这个静态变量实现的
//采用静态变量生成所有FTL元素在数组中的下标会造成一个问题，
//会造成InternalThreadLocalMap中数组不必要的自动扩容
int index = nextIndex.getAndIncrement();
if (index < 0) {
nextIndex.decrementAndGet();
throw new IllegalStateException("too many thread-local indexed variables");
}
return index;
}

InternalThreadLocalMap.nextVariableIndex()方法获取FTL在该FastThreadLocalThread.threadLocalMap数组下标，因为InternalThreadLocalMap.nextVariableIndex() 使用静态域 nextIndex 递增维护所有FTL的下标，会造成后面实例化的 FTL 下标过大，如果FTL下标大于其对应 FastThreadLocalThread.threadLocalMap 数组的长度，会进行数组的自动扩容，如下：

private void expandIndexedVariableTableAndSet(int index, Object value) {
Object[] oldArray = indexedVariables;
final int oldCapacity = oldArray.length;
//下面复杂的实现是为了将newCapacity规范为最接近的一个2的指数,
//这段代码在早期的 jdk HashMap 中见过
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);
Arrays.fill(newArray, oldCapacity, newArray.length, UNSET);
newArray[index] = value;
indexedVariables = newArray;
}

3、FastThreadLocal

构造函数：

有两个重要的下标域，FTL不仅在FastThreadLocalThread.threadLocalMap中保存了用户实际使用的value(在数组中的下标为index)，还在数组中保存为了实现清理记录的相关数据，也即下标variablesToRemoveIndex,一般情况 variablesToRemoveIndex = 0;因为variablesToRemoveIndex 是静态变量，所以全局唯一;

//如果在该FTL中放入了数据，也就实际调用了其set或get函数，会在
//该FastThreadLocalThread.threadLocalMap数组的
// variablesToRemoveIndex下标处放置一个IdentityHashMap，
//并将该FTL放入IdentityHashMap中，在后续清理时会取出
//variablesToRemoveIndex下标处的IdentityHashMap进行清理
private static final int variablesToRemoveIndex = InternalThreadLocalMap.nextVariableIndex();
//在threadLocalMap数组中存放实际数据的下标
private final int index;
public FastThreadLocal() {
index = InternalThreadLocalMap.nextVariableIndex();
}

用户可扩展的函数：

//初始化 value 函数
protected V initialValue() throws Exception {
return null;
}
//让使用者在该FTL被移除时可以有机会做些操作。
protected void onRemoval(@SuppressWarnings("UnusedParameters") V value) throws Exception { }

FastThreadLocalThread

Java高进进阶之FastThreadLocal源码详解（修复ThreadLocal的缺陷）

cleanupFastThreadLocals 字段在 4.1 的最新版本中已经没有在用到了

/**
* true，表示FTL会在线程结束时被主动清理见 FastThreadLocalRunnable 类
* false，需要将FTL放入后台清理线程的队列中
*/
// This will be set to true if we have a chance to wrap the Runnable.
//这个字段则用于标识该线程在结束时是否会主动清理FTL
private final boolean cleanupFastThreadLocals;
//次对象将在第一次 FastThreadLocal.get 和 FastThreadLocal.set 时候创建
private InternalThreadLocalMap threadLocalMap;
public FastThreadLocalThread(Runnable target) {
super(FastThreadLocalRunnable.wrap(target));
cleanupFastThreadLocals = true;
}

4、 set 方法

public final void set(V value) {
//判断设置的 value 值是否是缺省值
if (value != InternalThreadLocalMap.UNSET) {
//获取当前线程的 InternalThreadLocalMap , 如果当前线程为FastThreadLocalThread，那么直接通过threadLocalMap引用获取
//否则通过 jdk 原生的 threadLocal 获取
InternalThreadLocalMap threadLocalMap = InternalThreadLocalMap.get();
//FastThreadLocal 对应的 index 下标的 value 替换成新的 value
setKnownNotUnset(threadLocalMap, value);
} else {
//如果放置的对象为UNSET，则表示清理，会对该FTL进行清理，类似毒丸对象
remove();
}
}

这里扩容方会调用 InternalThreadLocalMap.expandIndexedVariableTableAndSet

private void setKnownNotUnset(InternalThreadLocalMap threadLocalMap, V value) {
//在数组下标index处放置实际对象，如果index大于数组length，会进行数组扩容.
if (threadLocalMap.setIndexedVariable(index, value)) {
//放置成功之后，将该FTL加入到 variablesToRemoveIndex 下标的
//IdentityHashMap，等待后续清理
addToVariablesToRemove(threadLocalMap, this);
}
}

/**
* 该FTL加入到variablesToRemoveIndex下标的IdentityHashMap
* IdentityHashMap的特性可以保证同一个实例不会被多次加入到该位置
*/
@SuppressWarnings("unchecked")
private static void addToVariablesToRemove(InternalThreadLocalMap threadLocalMap, FastThreadLocal<?> variable) {
//获取 variablesToRemoveIndex下标处的 IdentityHashMap
Object v = threadLocalMap.indexedVariable(variablesToRemoveIndex);
Set<FastThreadLocal<?>> variablesToRemove;
//如果是第一次获取，则 variablesToRemoveIndex下标处的值为 UNSET
if (v == InternalThreadLocalMap.UNSET || v == null) {
//新建一个新的 IdentityHashMap 并
variablesToRemove = Collections.newSetFromMap(new IdentityHashMap<FastThreadLocal<?>, Boolean>());
//放入到下标variablesToRemoveIndex处
threadLocalMap.setIndexedVariable(variablesToRemoveIndex, variablesToRemove);
} else {
variablesToRemove = (Set<FastThreadLocal<?>>) v;
}
//将该FTL放入该IdentityHashMap中
variablesToRemove.add(variable);
}