02源码分析-ThreadLocal详解-并发编程(Java)

文章目录

- 1 ThreadLocal内部结构
- 2 主要方法源码分析
- - 2.1 set()方法
  - - 2.1.1 getMap()
    - 2.1.2 createMap()
    - 2.1.3 ThreadLocalMap.set()
    - 2.1.4 replaceStaleEntry()
    - 2.1.5 expungeStaleEntry()
    - 2.1.6 cleanSomeSlots()
    - 2.1.7 rehash()
    - 2.1.8 expungeStaleEntries()
    - 2.1.9 resize()
  - 2.2 get()方法
  - - 2.2.1 setInitialValue()
    - 2.2.2 initialValue()
  - 2.3 remove()方法
  - - 2.3.1 map.remove()
- 3 问题与分析
- ## 4 后记

1 ThreadLocal内部结构

ThreadLocal继承关系如下：在这里插入图片描述

下面展示ThreadLocal的内存结构设计对比图如下1-1所示：在这里插入图片描述

优化之后的设计：每个Thread维护一个ThreadLocalMap，这个Map的key为ThreadLocal实例本身，value是要存储的值。

具体的过程如下：

Thread类内部有一个ThreadLocalMap类型的变量：ThreadLocal.ThreadLocalMap threadLocals = null;
Map里面存储Entry键值对，key为ThreadLocal对象，value为线程的变量副本
Thread内部的Map实际由ThreadLocal维护，由ThreadLocal负责向线程设置和获取线程的变量值。
- ThreadLocal对象根据需要由我们自己创建。
对于不同的线程，每次获取副本值时，别的线程不能获取到当前线程的副本值，形成副本隔离，互不干扰。

优化之后的设计方案的优点：

每个Map存储的Entry数量变少
- ThreadLocal对象数量一般少于线程数量
当Thread销毁的时候，ThreadLocalMap也随之销毁，及时释放内存，减少内存开销

2 主要方法源码分析

2.1 set()方法

源码如下1-1所示：

public void set(T value) {
    Thread t = Thread.currentThread();
    ThreadLocalMap map = getMap(t);
    if (map != null)
        map.set(this, value);
    else
        createMap(t, value);
}

执行流程:

获取当前线程赋值给t
获取当前线程维护的ThreadLocalMap对象map
map判断是否为空
- 为空调用createMap(t, value)创建新的map
- 不为空通过map.set(this,value)把键值对存入map中

2.1.1 getMap()

源码如下：

ThreadLocalMap getMap(Thread t) {
    return t.threadLocals;
}

就是获取线程内部的ThreadLocalMap变量。

2.1.2 createMap()

源码如下：

void createMap(Thread t, T firstValue) {
    t.threadLocals = new ThreadLocalMap(this, firstValue);
}
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);
}

新建ThreadLocalMap，初始化容器，以当前ThreadLocal实例为key，值为value新增Entry存入数组的适当位置。关于位置索引的计算，后面讲到hash冲突处理的时候详细说明。

2.1.3 ThreadLocalMap.set()

private void set(ThreadLocal<?> key, Object value) {

    // We don't use a fast path as with get() because it is at
    // least as common to use set() to create new entries as
    // it is to replace existing ones, in which case, a fast
    // path would fail more often than not.

    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();

        if (k == key) {
            e.value = value;
            return;
        }

        if (k == null) {
            replaceStaleEntry(key, value, i);
            return;
        }
    }

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

该方法才是set()的主要实现逻辑处理的方法，执行流程：

hash计算key的数组索引i
初始化Entry e为tab[i]
开始循环：判断条件就是e不为空，就是数组元素不为空，指向某一Entry
- 获取e的key，判断和参数key是否相等
  - 相等直接替换该键值对中的值，return
- 不相等判断key是否为空：为空意味着该Entry是“垃圾”，需要被回收
  - 调用replaceStaleEntry()方法，构建一个新的Entry替换e，return
- 继续获取下一个索引处的Entry
  - nextIndex()用来获取下一个索引，这个方法实现闭环获取，即到达数组末尾会从0开始。
循环结束，没有return的情况就是在数组遍历到位空的元素之前没有找到key相等的Entry
- 此时索引i出元素就为空
直接创建新的Entry，放入数组索引i处，调整数组大小+1
调用cleanSomeSlots()方法来清理数组中有元素Entry但是键为空的元素
- 如果没有要清理的元素，继续判断数组元素个数是否大于等于阈值
  - 是的话就调用rehash()进行全数组清理，重新计算hash。并且判断如果size超过阈值的 $\frac{3}{4}$ ,进行扩容。

方法执行流程图如下2.1.3-1所示：在这里插入图片描述

2.1.4 replaceStaleEntry()

源代码如下：

private void replaceStaleEntry(ThreadLocal<?> key, Object value,
                               int staleSlot) {
    Entry[] tab = table;
    int len = tab.length;
    Entry e;

    // Back up to check for prior stale entry in current run.
    // We clean out whole runs at a time to avoid continual
    // incremental rehashing due to garbage collector freeing
    // up refs in bunches (i.e., whenever the collector runs).
    int slotToExpunge = staleSlot;
    for (int i = prevIndex(staleSlot, len);
         (e = tab[i]) != null;
         i = prevIndex(i, len))
        if (e.get() == null)
            slotToExpunge = i;

    // Find either the key or trailing null slot of run, whichever
    // occurs first
    for (int i = nextIndex(staleSlot, len);
         (e = tab[i]) != null;
         i = nextIndex(i, len)) {
        ThreadLocal<?> k = e.get();

        // If we find key, then we need to swap it
        // with the stale entry to maintain hash table order.
        // The newly stale slot, or any other stale slot
        // encountered above it, can then be sent to expungeStaleEntry
        // to remove or rehash all of the other entries in run.
        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 we didn't find stale entry on backward scan, the
        // first stale entry seen while scanning for key is the
        // first still present in the run.
        if (k == null && slotToExpunge == staleSlot)
            slotToExpunge = i;
    }

    // If key not found, put new entry in stale slot
    tab[staleSlot].value = null;
    tab[staleSlot] = new Entry(key, value);

    // If there are any other stale entries in run, expunge them
    if (slotToExpunge != staleSlot)
        cleanSomeSlots(expungeStaleEntry(slotToExpunge), len);
}

方法目标就是替换数组中的无用Entry，具体执行步骤如下：

设置变量
- tab：数组
- len：数组长度
- e：Entry变量
- slotToExpunge：标记要清除的Entry索引
循环前向遍历数组：循环条件当前Entry e不为空，从传递的索引staleSlot开始
- 判断如果e的key为空，e的索引标记为回收索引slotToExpunge
循环后向遍历数组：循环条件当前Entry e不为空，从传递的索引staleSlot开始
- 判断如果e的key=形参key，替换value
- 索引i与形参staleSlot指向的Entry互换
- 判断如果staleSlot=标记回收索引，此时互换后索引i标记为回收索引
  - slotToExpunge == staleSlot什么情况呢？就是之前前向遍历了一圈，staleSlot索引处为唯一的回收Entry或者在遍历到元素为空为止，没有找到可回收的Entry，此时slotToExpunge 的初始赋值为staleSlot。
- 调用cleanSomeSlots(expungeStaleEntry(slotToExpunge), len)执行清理回收Entries，return
- k == null && slotToExpunge == staleSlot，i标记为回收索引
此时还是没有找到key与形参key相等的Entry，那么直接释放原先回收索引staleSlot Entry的value，新建Entry(key,value)放在staleSlot 处。
slotToExpunge != staleSlot 说明有可以回收的Entry，执行清理

2.1.5 expungeStaleEntry()

源码如下：

private int expungeStaleEntry(int staleSlot) {
    Entry[] tab = table;
    int len = tab.length;

    // expunge entry at staleSlot
    tab[staleSlot].value = null;
    tab[staleSlot] = null;
    size--;

    // Rehash until we encounter null
    Entry e;
    int i;
    for (i = nextIndex(staleSlot, len);
         (e = tab[i]) != null;
         i = nextIndex(i, len)) {
        ThreadLocal<?> k = e.get();
        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;
}

执行回收Entry，步骤如下：

回收参数索引staleSlot处的Entry
循环后向遍历数组：判断条件Entry不为空
- entry的key为空直接回收
- entry的key不为空，重新计算hash
- 如果重新计算的hash和索引不相等
  - 后向遍历找到第一个为空的数组元素，把entry e插入
返回数组为空元素的索引

2.1.6 cleanSomeSlots()

cleanSomeSlots()方法包裹在expungeStaleEntry()之外，在执行回收后，继续执行清理，源代码如下：

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;
}

参数i为expungeStaleEntry()返回的数组元素为空的索引，n为数组长度
执行循环
- 设置是否有清理entry标记removed，默认false
- 后向获取索引i的下一个索引
- 判断如果entry e不为空但是，e的key为空，改entry需要回收
- n重新置为数组长度，removed置为true表示有清理entry
- expungeStaleEntry(i)回收索引i处的entry
n无符号法右移一位，如果=0结束循环
返回removed

注：n只有在有entry回收的时候重新置为数组长度，没有回收的时候会变为原来的1/2

2.1.7 rehash()

执行rehash()的前提是没有可清理的entry且元素个数大于等于阈值。cleanSomeSlots()或者expungeStaleEntry()并不是执行的整个数组清理回收。该方法源码如下：

private void rehash() {
    expungeStaleEntries();

    // Use lower threshold for doubling to avoid hysteresis
    if (size >= threshold - threshold / 4)
        resize();
}

执行流程如下：

执行整个数组的清理
清理之后元素个数大于等于阈值的3/4，执行扩容

2.1.8 expungeStaleEntries()

源代码如下:

private void expungeStaleEntries() {
    Entry[] tab = table;
    int len = tab.length;
    for (int j = 0; j < len; j++) {
        Entry e = tab[j];
        if (e != null && e.get() == null)
            expungeStaleEntry(j);
    }
}

很明显，循环从数组开头至数组的末尾，执行回收清理。

2.1.9 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);
                while (newTab[h] != null)
                    h = nextIndex(h, newLen);
                newTab[h] = e;
                count++;
            }
        }
    }

    setThreshold(newLen);
    size = count;
    table = newTab;
}

执行流程如下：

创建新的原来数组2倍长度的新数组
循环遍历旧数组
- 取entry e 判断如果e不为空
  - 判断e的key如果为空，直接回收
  - 不为空，重新计算hash放入新数组
设置新阈值，size，table

注：扩容原来的2倍

2.2 get()方法

源代码如下：

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();
}

执行流程：

获取当前线程t
获取线程t绑定的 ThreadLocalMap map
判断如果map不为空
- 以当前ThreadLocal实例为key获取entry e
- 判断如果e不为空，获取e中的value，返回
如果map为空执行setInitialValue()初始一个新的map

2.2.1 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;
}

执行流程如下：

获取初始化value值
获取当前线程t，获取线程t绑定的ThreadLocalMap map
判断如果map不为空，map.set(this,value)设置值
如果map为空，createMap(t,value)执行新建map操作

2.2.2 initialValue()

源代码如下：

protected T initialValue() {
    return null;
}

默认为空，需要子类覆盖

2.3 remove()方法

源代码如下：

public void remove() {
    ThreadLocalMap m = getMap(Thread.currentThread());
    if (m != null)
        m.remove(this);
}

2.3.1 map.remove()

源代码如下：

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) {
            e.clear();
            expungeStaleEntry(i);
            return;
        }
    }
}

执行流程如下：

计算key的hash结果 i
for循环后向获取entry e，判断条件e不为空
- 如果e的key等于形参key
  - e.clear() 回收key涉及弱引用后面关于ThreadLocal内存泄露的时候讲解
  - expungeStaleEntry(i)回收e，返回

3 问题与分析

1 为啥在set方法中回收清理这么频繁？

浅析：ThreadLocalMap绑定到线程中的，线程作为一种重要的资源，ThreadLocal作为ThreadLocalMap的管理者，如果作用域结束（任务完成）应该及时回收清理相应的内存。

2 那么这种回收清理是实时的吗？即当堆中ThreadLocal的引用被回收，对应的当前线程ThreadLocalMap以该ThreadLocal实例的为key的entry是否立即被回收了呢？

结论：如果没有执行ThreadLocal.remove()方法，该entry是不会被立即回收的。ThreadLocal的引用被回收后，对应的entry的key由于是弱引用被回收，但是entry的value是强引用。在当前线程引用没有被回收之前，如果不执行ThreadLocal的set(),remove()等方法时，该entry是不会被回收的。

关于ThreadLocal的内存堆栈结构，我们接下来会在ThreadLocal内存泄露内容部分讲解。