Java8源码阅读 - ThreadLocal
2020-07-01 本文已影响0人
Mhhhhhhy
有时候我们想要将某些数据和特定的线程进行关联，可以使用一个全局Map将线程id和数据做映射，当然我们还可以使用JDK提供现成的ThreadLocal来完成这个需求；
只要是启动的线程并且ThreadLocal实例是可访问的，该个线程都有一个对其线程局部变量副本的隐式引用，在一个线程退出后，它的线程本地实例的所有副本都将会被GC给回收（除非存在对这些副本的其他引用）。
ThreadLocal.ThreadLocalMap threadLocals = null;
在Thread类里面有一个threadLocalMap的引用；
ThreadLocalMap
static class ThreadLocalMap {
static class Entry extends WeakReference<ThreadLocal<?>> {
Object value;
Entry(ThreadLocal<?> k, Object v) {
super(k);
value = v;
}
}
private Entry[] table;
...
}
ThreadLocalMap内部使用一个Entry类型来储存key和value的映射，key为ThreadLocal对象，value为想要储存的对象，即同一个ThreadLocal对象只能对应一个value；
同时Entry使用弱引用持有ThreadLocal对象，如果ThreadLocal对象不再有任何强引用存在时，能够让GC释放Entry对象已经内部的value，避免内存泄露；
// 必须为2的幂次方
private static final int INITIAL_CAPACITY = 16;
// 扩/缩容阀值
private int threshold;
ThreadLocalMap(ThreadLocal<?> firstKey, Object firstValue) {
table = new Entry[INITIAL_CAPACITY];
int i = firstKey.threadLocalHashCode & (INITIAL_CAPACITY - 1);
table[i] = new Entry(firstKey, firstValue);
size = 1;
setThreshold(INITIAL_CAPACITY); // threshold为table大小的2/3
}
构造器方法初始化了table、size、threshold，有个需要注意的地方就是关于hash定位的方式；
private final int threadLocalHashCode = nextHashCode();
private static AtomicInteger nextHashCode = new AtomicInteger();
// 将隐式顺序线程本地id转换为接近最优扩散的倍增哈希值
//（为什么是这个值，网上有很多资料）
private static final int HASH_INCREMENT = 0x61c88647;
private static int nextHashCode() {
return nextHashCode.getAndAdd(HASH_INCREMENT);
}
在同一个ClassLoader中所有ThreadLocal都共享一个AtomicInteger#nextHashCode，所以每次调用都是在上一次的基础上自增0x61c88647，那么就意味着每个ThreadLocal对象的hashCode是不一样的；
private void set(ThreadLocal<?> key, Object value) {
Entry[] tab = table;
int len = tab.length;
int i = key.threadLocalHashCode & (len-1);
for (Entry e = tab[i]; e != null; e = tab[i = nextIndex(i, len)]) {
// 如果出现hash碰撞
ThreadLocal<?> k = e.get();
if (k == key) {
// 如果是同一个threadLocal对象，替换value
e.value = value;
return;
}
// 不是同一个对象，意味着两个threadLocal对象定位到同一个slot
if (k == null) {
// 如果threadLocal被GC掉了
replaceStaleEntry(key, value, i); // replace
return;
}
// 该槽被占用了，重新定位
}
// 该slot是空的，将entry放入槽中
tab[i] = new Entry(key, value);
int sz = ++size;
// 如果有必要则进行清除工作
if (!cleanSomeSlots(i, sz) && sz >= threshold)
rehash(); // 扩容
}
// 到最大长度后从0开始
private static int nextIndex(int i, int len) {
return ((i + 1 < len) ? i + 1 : 0);
}
采用开放地址法处理哈希碰撞，如果同一个槽被占用了在再次定位前，会首先判断占用槽的ThreadLocal是否被GC释放掉了，如果是会把该槽原本的entry替换成新的threadLocal；
如果没有出现hash冲突，直接将新的entry放入槽中，并且在最后都需要进行检查table是是否有陈旧的slot，如果有必要的话进行清除工作，如果都没有且当前table的大小已经到了扩容阀值，则会进行扩容操作；
替换陈旧的Entry
private void replaceStaleEntry(ThreadLocal<?> key, Object value, int staleSlot) {
// key：需要set的threadLocal
// value：value
// staleSlot：set中遇到的第一个可以被替换的槽空间
Entry[] tab = table;
int len = tab.length;
Entry e;
// 以下操作其目的还是为了尽可能的释放slot空间
int slotToExpunge = staleSlot;
// 从当前槽开始向前局部扫描，查询并记录需要擦除的slot，直到遇到第一个空槽
for (int i = prevIndex(staleSlot, len); (e = tab[i]) != null; i = prevIndex(i, len))
if (e.get() == null)
slotToExpunge = I;
// 从staleSlot开始向后扫描，第一个元素为staleSlot+1
for (int i = nextIndex(staleSlot, len); (e = tab[i]) != null; i = nextIndex(i, len)) {
ThreadLocal<?> k = e.get();
// 如果找到了key，那么我们需要将它与陈旧的entry交换，以维护哈希表的顺序
if (k == key) {
// 找到了该threadLocal对象
e.value = value; // 直接替换value
// 将当前threadLocal对象所在的槽的位置移到需要交换的位置
tab[i] = tab[staleSlot];
tab[staleSlot] = e;
if (slotToExpunge == staleSlot) // 如果前面没有找到需要擦除的slots
slotToExpunge = I;
// 从原来threadLocal的位置开始进行清除工作
cleanSomeSlots(expungeStaleEntry(slotToExpunge), len);
return;
}
// 该槽的threadLocal对象不是调用set方法的threadLocal
if (k == null && slotToExpunge == staleSlot)
// k == null：意味着发现另一个可以被清空的槽
// slotToExpunge == staleSlot：意味着在遍历过程中这是第一个被发现陈旧的槽
slotToExpunge = I;
}
// 在刚刚的布局扫描中，没有发现传进来的threadLocal对象
// 把陈旧的slot中的threadLocal对象替换成新的threadLocal
tab[staleSlot].value = null;
tab[staleSlot] = new Entry(key, value);
// 在刚刚的布局扫描中，发现其他陈旧的slot，进行清除工作
if (slotToExpunge != staleSlot)
cleanSomeSlots(expungeStaleEntry(slotToExpunge), len);
}
该方法把在set操作过程中遇到的可以被替换的陈旧的Entry对象，替换成新的ThreadLocal，主要是两个步骤：
- 进行局部扫描，查询并记录是否有需要擦除的slot；
- 从被替换槽的位置开始向后扫描，如果发现threadLocal对象已经存在table里面，则意味着当时在set操作添加该threadLocal时遇到了哈希冲突，于是向后寻找空间插入，现在原本的位置上的old_threadLocal对象已经被GC释放掉，那么可以把该threadLocal对象的位置移动到原本old_threadLocal的位置，两个对象hash定位到的位置是同一个，所以移动不会造成哈希表顺序混乱；
- 如果在table中找不到threadLocal对象，则直接将旧的Entry替换成老的Entry；
- 方法结束前，如果有需要都会进行清理工作；
清除工作
private int expungeStaleEntry(int staleSlot) {
Entry[] tab = table;
int len = tab.length;
// 清除陈旧的slot
tab[staleSlot].value = null;
tab[staleSlot] = null;
size--;
// 以下操作目的是将部分槽重新哈希
Entry e;
int I;
// 从当前的staleSlot开始向后遍历，直到遇到空槽
for (i = nextIndex(staleSlot, len); (e = tab[i]) != null; i = nextIndex(i, len)) {
ThreadLocal<?> k = e.get();
if (k == null) {
// 如果遇到其他threadLocal被gc的，一并释放该槽空间
e.value = null;
tab[i] = null;
size--;
} else {
int h = k.threadLocalHashCode & (len - 1); // 重新哈希entry
if (h != i) {
// 如果重哈希后是不同的位置
tab[i] = null;
while (tab[h] != null)
// 如果遇到hash冲突，再次hash
h = nextIndex(h, len);
tab[h] = e;
}
}
}
return I;
}
该方法清除table中指定槽的空间，并且还会顺带遍历部分槽空间，如果发现有被淘汰的threadLocal也会一并把这些槽给清空，如果不需要清除，那么会尝试将它们重新hash定位，最后返回从staleSlot向后遍历时遇到的第一个空槽；
private boolean cleanSomeSlots(int i, int n) {
boolean removed = false;
Entry[] tab = table;
int len = tab.length;
do {
// 从某个位置开始遍历扫描
i = nextIndex(i, len);
Entry e = tab[I];
if (e != null && e.get() == null) {
n = len;
removed = true;
i = expungeStaleEntry(i);
}
} while ( (n >>>= 1) != 0);
return removed;
}
扫描log2(n)个槽空间，如果发现有需要清除的slot，会执行expungeStaleEntry进行清扫，并且还会花费O(n)的时间复杂度来清理其余部分陈旧的slot，这么多操作其目的就是为了尽可能的将陈旧的slot及时清空，避免不必要的扩容操作；
扩容
private void rehash() {
expungeStaleEntries(); // 遍历table，如果发现可回收的slot会进行回收
if (size >= threshold - threshold / 4)
resize();
}
// 直接扩容成两倍大小
private void resize() {
Entry[] oldTab = table;
int oldLen = oldTab.length;
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); // 重新hash
while (newTab[h] != null)
h = nextIndex(h, newLen);
newTab[h] = e;
count++;
}
}
}
setThreshold(newLen);
size = count;
table = newTab;
}
在rehash操作里面再一次的检查是否有可以清理的槽，可以看到作者的意图，在非不得已的情况下才会进行扩容操作，并且扩容后是不能把容量缩小的，扩容后会对所有的元素进行重新哈希；
Get
private Entry getEntry(ThreadLocal<?> key) {
int i = key.threadLocalHashCode & (table.length - 1);
Entry e = table[I];
if (e != null && e.get() == key)
// 如果hash命中，直接返回
return e;
else
return getEntryAfterMiss(key, i, e);
}
private Entry getEntryAfterMiss(ThreadLocal<?> key, int i, Entry e) {
Entry[] tab = table;
int len = tab.length;
while (e != null) {
// 有hash冲突
ThreadLocal<?> k = e.get();
if (k == key)
return e;
if (k == null)
// threadlocal被释放了，开始清理工作
expungeStaleEntry(i);
else
// 向下继续寻找
i = nextIndex(i, len);
e = tab[I];
}
return null;
}
在通常情况下会直接哈希命中获取到所对应值，但是在命中失败的情况下，如果有发现陈旧的entry，会先进行清理工作，然后重新定位直到找到或找不到value；
ThreadLocal
public void set(T value) {
Thread t = Thread.currentThread();
ThreadLocalMap map = getMap(t);
if (map != null)
map.set(this, value);
else
createMap(t, value);
}
ThreadLocalMap getMap(Thread t) {
return t.threadLocals;
}
void createMap(Thread t, T firstValue) {
t.threadLocals = new ThreadLocalMap(this, firstValue);
}
set方法将自身ThreadLocal对象作为key，然后调用ThreadLocalMap#set处理；
public T get() {
Thread t = Thread.currentThread();
ThreadLocalMap map = getMap(t);
if (map != null) {
ThreadLocalMap.Entry e = map.getEntry(this);
if (e != null) {
@SuppressWarnings("unchecked")
T result = (T)e.value;
return result;
}
}
return setInitialValue();
}
private T setInitialValue() {
T value = initialValue();
Thread t = Thread.currentThread();
ThreadLocalMap map = getMap(t);
if (map != null)
map.set(this, value);
else
createMap(t, value);
return value;
}
protected T initialValue() {
return null;
}
同样get方法也是将自身作为key到ThreadLocalMap中查找；
主要的逻辑代码差不多都看过了，整个引用关系图也就明了了；
实线为强引用，虚线为弱引用
每个线程ThreadLocalMap对象的生命周期是伴随着整个线程的生命周期，在同一个线程中，所有ThreadLocal对象操作的map都是同一个，而ThreadLocal和Thread的生命周期却没有必然的联系，所以说ThreadLocalMap内部对ThreadLocal持弱引用；
下面来看两个实例
示例1
public class ThreadLocalTest {
public static void main(String[] args) {
new MyThread().start();
}
static class MyThread extends Thread {
ThreadLocal<Integer> threadLocal = new ThreadLocal<>();
@Override
public void run() {
for (int i = 0; i < 9; i++) {
ThreadLocal<Integer> tmp = new ThreadLocal<>();
tmp.set(i+100);
}
System.gc();
try {
Thread.sleep(10000);
} catch (InterruptedException e) {
e.printStackTrace();
}
threadLocal.set(1);
}
}
}
实例一模拟的是ThreadLocal的生命周期比线程的生命周期短的场景；
在该场景下等到代码走到threadLocal.set(1);这一行时，该线程的ThreadLocalMap会有大量的Entry里面的referent是为空，因为tmp被GC回收了，然而Entry里面的Value仍然是存在强引用，如下图所示
最终会触发cleanSomeSlots进行清扫工作，将陈旧的Entry全部释放，进而释放掉Entry对Value的引用；
示例2
public class ThreadLocalTest {
static final ThreadLocal<Integer> threadLocal = new ThreadLocal<>();
public static void main(String[] args) {
new MyThread().start();
new MyThread().start();
new MyThread().start();
new MyThread().start();
System.gc();
try {
Thread.sleep(5000);
} catch (InterruptedException e) {
e.printStackTrace();
}
new MyThread().start();
}
static class MyThread extends Thread {
@Override
public void run() {
threadLocal.set(1);
}
}
}
第二种示例就是模拟ThreadLocal生命周期比线程生命周期长的场景，这种场景应该是比较常见的，引用关系也比较简单；
