0%

ThreadLocal源码剖析

Posted on Edited on In

顾名思义，ThreadLocal是为线程提供私有的局部变量。它不同于其他常规的变量，需要使用自身的get和set方法来获取和设置值。ThreadLocal的典型应用是在类中被申明为静态变量，用于关联用户ID、事务ID，亦或者其他需要线程独有的属性。

对于ThreadLocal，只要该线程处于活动状态并且ThreadLocal实例是可访问的，每个线程都保留对其本地线程副本的隐式引用。如果线程消失后，其所有副本线程本地实例便会受到垃圾回收（除非其他情况，即线程外存在对这些副本的引用）。

构建实例

首先ThreadLocal在构建方面可以用类自身的构造函数，但是除了构造函数外，还支持了一个withInitial方法，如下所示：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
public static <S> ThreadLocal<S> withInitial(Supplier<? extends S> supplier) {
    return new SuppliedThreadLocal<>(supplier);
}
static final class SuppliedThreadLocal<T> extends ThreadLocal<T> {

    private final Supplier<? extends T> supplier;

    SuppliedThreadLocal(Supplier<? extends T> supplier) {
        // 判空并赋值
        this.supplier = Objects.requireNonNull(supplier);
    }

    @Override
    protected T initialValue() {
        return supplier.get();
    }
}

该方法利用了java.util.function中的Supplier接口，该接口用于提供获取值，接口代码如下所示：

1
2
3
public interface Supplier<T> {
    T get();
}

当构建了ThreadLocal实例之后，在没有通过set方法设置值的前提下，调用get方法后会调用该接口的方法获取值。

设置值

在ThreadLocal中为线程的私有局部变量设置值是通过其共有的set方法，首先是要获取到当前线程的ThreadLocalMap，如果该映射存在则直接插入值，否则要创建并且插入值。源码如下所示：

1
2
3
4
5
6
7
8
9
10
public void set(T value) {
    // 获取当前线程实例
    Thread t = Thread.currentThread();
    // 获取当前线程的ThreadLocalMap
    ThreadLocalMap map = getMap(t);
    if (map != null)
        map.set(this, value);
    else
        createMap(t, value);
}

这里涉及一个ThreadLocalMap，会在本文的后半部分进行剖析，这里先直接理解为一个map映射，且map的key为当前ThreadLocal实例。

获取值

在ThreadLocal中获取线程的私有局部变量是通过其共有的get方法来实现的，同set方法一样，首先要获取当前线程的map映射，然后获取值。

1
2
3
4
5
6
7
8
9
10
11
12
13
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();
}

这里有要说明的一点是setInitialValue方法，其是在从map映射中获取不到值的时候，便会invoke此方法，以此来获取值。而该方法首先是调用initialValue方法获取初始值，随后在向线程的私有map映射中设置此值。源码如下：

1
2
3
4
5
6
7
8
9
10
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;
}

到此再回头看ThreadLocal的构造方式的源码，SuppliedThreadLocal就重写了initialValue方法，当在这里调用此方法的时候，便会执行Supplier接口的get方法以便获取值。当然，如果你使用的时候默认的构造函数构造实例，那么也可以自己实现initialValue，用于设置初始值。默认情况下，initialValue的实现如下所示：

1
2
3
protected T initialValue() {
    return null;
}

变量的移除

要移除线程私有的变量，只需拿到线程的map映射，在移动当前的ThreadLocal即可，源码如下：

1
2
3
4
5
public void remove() {
    ThreadLocalMap m = getMap(Thread.currentThread());
    if (m != null)
        m.remove(this);
}

在删除此线程本地变量值后，如果其随后由当前线程get方法进行读取，其值将通过调用其initialValue方法进行重新初始化，除非当前线程的值在进行set， 这可能导致多次调用当前线程中的initialValue方法。

线程的资源空间有限，所我们在使用完本地局部变量后需要把变量remove掉，以防出现内存泄露。当然线程在退出的时候，也会进行资源相关的清理，其是直接吧整个线程的threadLocals映射表置null（方便垃圾回收更快速的清理）。源码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
private void exit() {
    if (group != null) {
        group.threadTerminated(this);
        group = null;
    }
    /* Aggressively null out all reference fields: see bug 4006245 */
    target = null;
    /* Speed the release of some of these resources */
    threadLocals = null;
    inheritableThreadLocals = null;
    inheritedAccessControlContext = null;
    blocker = null;
    uncaughtExceptionHandler = null;
}

ThreadLocalMap

键值对定义

ThreadLocalMap是特意为ThreadLocal适配定制的一个哈希表。为了帮助处理非常大且长期存在的用法，哈希表条目使用弱引用WeakReferences作为键。但是由于未使用引用队列，所以其仅在表的可利用空间不足时，才会去删除过时的数据。哈希表的键值对定义如下：

当key为null的时候，代码当前key不再被引用，可以被清除

1
2
3
4
5
6
7
8
9
static class Entry extends WeakReference<ThreadLocal<?>> {
    /** The value associated with this ThreadLocal. */
    Object value;

    Entry(ThreadLocal<?> k, Object v) {
        super(k);
        value = v;
    }
}

哈希码定义

在剖析ThreadLocalMap的具体实现之前，先来看看其hash-code的定义（定义在ThreadLocal中），如下所示：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
private final int threadLocalHashCode = nextHashCode();

// 下一个哈希码，从0开始原子更新
private static AtomicInteger nextHashCode =
    new AtomicInteger();

// 哈希码。
// 其连续生成的哈希码之间的差异是：将隐式顺序线程本地ID转换为用于2的幂次方表的近似最佳分布的乘法哈希值。
private static final int HASH_INCREMENT = 0x61c88647;

// 获取下一个hashcode
private static int nextHashCode() {
    return nextHashCode.getAndAdd(HASH_INCREMENT);
}

ThreadLocals依赖于附加到每个线程的每线程线性探针哈希映射（Thread.threadLocals是InheritableThreadLocals）。ThreadLocal对象充当键，通过threadLocalHashCode搜索。 这是一个自定义哈希码（仅在ThreadLocalMaps中有用），在相同的线程使用连续构造的ThreadLocals的常见情况下，它消除了冲突。

构造实例

构造ThreadLocalMap实例的方法有两个，一个是通过当个值构造，一个是通过整个ThreadLocalMap进行构造。单值构造源码如下：

1
2
3
4
5
6
7
8
9
10
ThreadLocalMap(ThreadLocal<?> firstKey, Object firstValue) {
    // 创建哈希表， INITIAL_CAPACITY=16
    table = new Entry[INITIAL_CAPACITY];
    // 根据hash码计算下标
    int i = firstKey.threadLocalHashCode & (INITIAL_CAPACITY - 1);
    table[i] = new Entry(firstKey, firstValue);
    size = 1;
    // 设置resize的临界值 INITIAL_CAPACITY*2/3
    setThreshold(INITIAL_CAPACITY);
}

批量构建ThreadLocalMap是通过给定的继承的ThreadLocalMap来构建一个完成的映射表，源码如下所示：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
private ThreadLocalMap(ThreadLocalMap parentMap) {

    // 初始化哈希表
    Entry[] parentTable = parentMap.table;
    int len = parentTable.length;
    setThreshold(len);
    table = new Entry[len];

    // 设置值
    for (int j = 0; j < len; j++) {
        Entry e = parentTable[j];
        if (e != null) {
            @SuppressWarnings("unchecked")
            ThreadLocal<Object> key = (ThreadLocal<Object>) e.get();
            if (key != null) {
                Object value = key.childValue(e.value);
                Entry c = new Entry(key, value);
                int h = key.threadLocalHashCode & (len - 1);
                // 寻找下标定位
                while (table[h] != null)
                    h = nextIndex(h, len);
                table[h] = c;
                size++;
            }
        }
    }
}
private void setThreshold(int len) {
    threshold = len * 2 / 3;
}
private static int nextIndex(int i, int len) {
    return ((i + 1 < len) ? i + 1 : 0);
}

根据给定的映射表构建ThreadLocalMap会在ThreadLocal中被调用，调用方法如下，这也进一步说明了ThreadLocalMap类是ThreadLocal的私有的类，不能直接被外部访问。

1
2
3
static ThreadLocalMap createInheritedMap(ThreadLocalMap parentMap) {
    return new ThreadLocalMap(parentMap);
}

获取键值对

ThreadLocalMap在获取值方面，首先是根据给定的key计算出其在哈希表中的下标，然后直接命中哈希表中的此值，如果键值对存在且与给定的key一致，则直接返回，否则进行下一步操作。

1
2
3
4
5
6
7
8
private Entry getEntry(ThreadLocal<?> key) {
    int i = key.threadLocalHashCode & (table.length - 1);
    Entry e = table[i];
    if (e != null && e.get() == key)
        return e;
    else
        return getEntryAfterMiss(key, i, e);
}

在直接命中哈希槽没有获取到键值对的时候，会进行遍历整个哈希表来寻找值，源码如下所示：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
private Entry getEntryAfterMiss(ThreadLocal<?> key, int i, Entry e) {
    Entry[] tab = table;
    int len = tab.length;

    while (e != null) {
        ThreadLocal<?> k = e.get();
        if (k == key)
            return e;
        if (k == null)
            // 清楚过时的key
            expungeStaleEntry(i);
        else
            i = nextIndex(i, len);
        e = tab[i];
    }
    return null;
}

可以看见，在此次遍历哈希表的过程中会有异步特殊的操作，那就是清除不再引用（key=null）的键值对。

设置键值对

ThreadLocalMap在设置值的时候，如果key存在会直接覆盖，如果不存在则会替换一个过时的槽，最后再根据阙值情况决定是否进行rehash行为。源码如下所示：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
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)]) {
        ThreadLocal<?> k = e.get();

        // 如果key存在则直接覆盖
        if (k == key) {
            e.value = value;
            return;
        }
        // 如果找到为null的槽哈希槽则赋值
        if (k == null) {
            replaceStaleEntry(key, value, i);
            return;
        }
    }

    tab[i] = new Entry(key, value);
    int sz = ++size;
    if (!cleanSomeSlots(i, sz) && sz >= threshold)
        rehash();
}

移除条目

ThreadLocalMap移除条目是在寻找到key之后，直接置空引用并且对当前过时槽进行一次清理。源码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
private void remove(ThreadLocal<?> key) {
    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)]) {
        if (e.get() == key) {
            // 引用置null
            e.clear();
            // 清除
            expungeStaleEntry(i);
            return;
        }
    }
}

清除过时槽

清除过时的条目操作中，会rehash任何一个可能冲突的建，直到遇到空槽为止，并且此过程中遇到的所有的过时槽均会被清除。源码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
private int expungeStaleEntry(int staleSlot) {
    // 获取哈希表
    Entry[] tab = table;
    int len = tab.length;
    // 清楚过时的槽，设置为空槽
    tab[staleSlot].value = null;
    tab[staleSlot] = null;
    // 当前哈希表中的元素个数减一
    size--  
    // 执行Rehash操作直到遇到空槽为止（null）
    Entry e;
    int i;
    for (i = nextIndex(staleSlot, len);
         (e = tab[i]) != null;
         i = nextIndex(i, len)) {
        ThreadLocal<?> k = e.get();
        // 如果key（ThreadLocal）为null，即为过时槽，进行清楚置空
        if (k == null) {
            e.value = null;
            tab[i] = null;
            size--;
        } else {
            // 根据当前的哈希码计算下标
            int h = k.threadLocalHashCode & (len - 1);
            if (h != i) {
                tab[i] = null;
                // Unlike Knuth 6.4 Algorithm R, we must scan until
                // null because multiple entries could have been stale.
                while (tab[h] != null)
                    h = nextIndex(h, len);
                tab[h] = e;
            }
        }
    }
    return i;
}

替代过时的槽

替换过时槽的行为中，首先会备份找到第一个需要清理的过时槽；然后再寻找key，如果找到了赋值且会执行清理过时槽的操作；最后在没有寻找到key的情况下结束循环，直接在过时槽位置设置新的值，随后根据情况（备份过时槽与过时槽是否一致）决定是否清理过时槽。源码如下所示：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
private void replaceStaleEntry(ThreadLocal<?> key, Object value,
                               int staleSlot) {
    Entry[] tab = table;
    int len = tab.length;
    // 备份过时槽
    int slotToExpunge = staleSlot;
    //扫描需要被清理的槽
    for (int i = prevIndex(staleSlot, len);
         (e = tab[i]) != null;
         i = prevIndex(i, len))
        if (e.get() == null)
            slotToExpunge = i;
    // 寻找key或者空槽
    for (int i = nextIndex(staleSlot, len);
         (e = tab[i]) != null;
         i = nextIndex(i, len)) {
        ThreadLocal<?> k = e.get();
        // 如果寻找到了key，则交换值以此保持哈希表的顺序
        if (k == key) {
            e.value = value;
            tab[i] = tab[staleSlot];
            tab[staleSlot] = e;
            // Start expunge at preceding stale entry if it exists
            if (slotToExpunge == staleSlot)
                slotToExpunge = i;
            //清理槽
            cleanSomeSlots(expungeStaleEntry(slotToExpunge), len);
            return;
        }
        // 如果在向后扫描中未找到过时的条目，且看到的第一个过时的条目
        // 仍然是运行中的第一个过时的条目，则赋值。
        if (k == null && slotToExpunge == staleSlot)
            slotToExpunge = i;
    }
    // 跳出循环后即代表没有找到key，则直接在过时槽上赋值
    tab[staleSlot].value = null;
    tab[staleSlot] = new Entry(key, value);
    // 如果还存在其他的过时槽，则进行清理
    if (slotToExpunge != staleSlot)
        cleanSomeSlots(expungeStaleEntry(slotToExpunge), len);
}

cleanSomeSlots行为也是去执行清理过时槽操作，它是进行对数O(log2n)时间复杂度的扫描来清理过时的条目，作为无扫描（快速但保留垃圾）和与元素数量成比例的扫描数量之间的平衡。这个过程会发现所有垃圾，但也会导致某些插入花费O（n）时间。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
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;
}

rehash操作

在这个rehash操作中，是先清理掉所有过时的槽，然后在较低的阙值下进行resize操作。resize操作第一步便是直接翻倍哈希表的空间，然后将原始的哈希表中的元素置入新的哈希表，并且在置入元素时进行过时槽检测，一旦发现即可置空槽以便GC。 rehash相关源码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
private void rehash() {
    // 清除所有的过时槽
    expungeStaleEntries();
    // 使用较低的阈值加倍以避免滞后
    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) {
                // 如果是过时的槽则置空，方便垃圾回收（GC）
                e.value = null;
            } else {
                int h = k.threadLocalHashCode & (newLen - 1);
                while (newTab[h] != null)
                    h = nextIndex(h, newLen);
                newTab[h] = e;
                count++;
            }
        }
    }
    setThreshold(newLen);
    size = count;
    table = newTab;
}