Java 并发编程之 ThreadLocal

ThreadLocal 通过在各线程内部创建一个变量的副本,相比于使用各类锁机制访问变量保证线程安全性,ThreadLocal 的思想就是用空间换时间,使各线程都能访问属于自己这一份的变量副本,各线程之间变量值不互相干扰。

ThreadLocalMap

Java 中每个 Thread 都会维护一个 ThreadLocalMap 变量,存储在 ThreadLocalMap 中的就是一个以 Entry 为元素的 table 数组,Entry 是一个 key-value 结构,key 为 ThreadLocal,value 为存储的值。

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

Entry 是继承了 WeakReference 实现的,当 ThreadLocal Ref 销毁时,指向堆中 ThreadLocal 实例的强引用消失了,只有 Entry 有一条指向 ThreadLocal 实例的弱引用,这时 ThreadLocal 对象是可以被 GC 的。这时 Entry 里的 key 为 null 了,那么直到线程结束前,Entry 中的 value 都是无法回收的,所以这里可能会产生内存泄露,下面会说如何解决。

ThreadLocalMap 的主要数据结构就是一个 Entry 的数组 table,初始容量为 16,size 用于记录 map 中实际存在的 entry 个数,当 size 到 达threashold 时,需要 resize 整个 map,threshold 的初始值为len * 2 / 3

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
42
43
44
45
46
47
48
49
50
51
52
static class ThreadLocalMap {
static class Entry extends WeakReference<ThreadLocal<?>> {
/** The value associated with this ThreadLocal. */
Object value;

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

/**
* The initial capacity -- MUST be a power of two.
*/
private static final int INITIAL_CAPACITY = 16;

/**
* The table, resized as necessary.
* table.length MUST always be a power of two.
*/
private Entry[] table;

/**
* The number of entries in the table.
*/
private int size = 0;

/**
* The next size value at which to resize.
*/
private int threshold; // Default to 0

/**
* Set the resize threshold to maintain at worst a 2/3 load factor.
*/
private void setThreshold(int len) {
threshold = len * 2 / 3;
}

/**
* Increment i modulo len.
*/
private static int nextIndex(int i, int len) {
return ((i + 1 < len) ? i + 1 : 0);
}

/**
* Decrement i modulo len.
*/
private static int prevIndex(int i, int len) {
return ((i - 1 >= 0) ? i - 1 : len - 1);
}

getEntry

首先是计算索引位置 i,通过计算 key 的 hash%(table.length-1) 得出,根据获取 Entry,如果 Entry 存在且 Entry 的 key 恰巧等于 ThreadLocal,那么直接返回 Entry 对象,找不到就调用 getEntryAfterMiss。

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

getEntryAfterMiss

当出现 hash 冲突时会触发 getEntryAfterMiss,如果 k==key 那么代表找到了这个所需要的Entry,直接返回。如果 k==null,那么证明这个 Entry 中 key 已经为 null 那么这个 Entry 就是一个过期对象,这里调用 expungeStaleEntry 清理该 Entry。 上面提到的 key 被 gc 但是 value 没有被 gc,就是在这里解决的。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
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)
expungeStaleEntry(i);
else
i = nextIndex(i, len);
e = tab[i];
}
return null;
}

expungeStaleEntry

首先将 i 位置上的 Entry 的 value 设为 null,Entry 的引用也设为 null,那么系统 GC 的时候会自动回收。随后在一个循环中扫描 staleSlo t之后,null 之前的 Entry 数组,清除每一个 key 为 null 的 Entry,同时若是 key 不为空,做 rehash 调整位置。做 rehash 的目的是因为在清理的过程中会把某个值设为 null,那么这个值后面的区域如果之前是连着前面的,那么下次循环查找时,就会只查到 null 为止。

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

Get

首先获取当前的 Thread 对象,通过 getMap 获取 Thread 内的 ThreadLocalMap。如果map已经存在,以当前的 ThreadLocal 为键,获取 Entry 对象,并从 Entry 中取出值。否则调用 setInitialValue 方法进行初始化。

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

setInititialValue 在 map 不存在的时候调用,如果不存在则会调用 createMap 创建 ThreadLocalMap。

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

void createMap(Thread t, T firstValue) {
t.threadLocals = new ThreadLocalMap(this, firstValue);
}

set

1
2
3
4
5
6
7
8
public void set(T value) {
Thread t = Thread.currentThread();
ThreadLocalMap map = getMap(t);
if (map != null)
map.set(this, value);
else
createMap(t, value);
}

首先还是根据 key 计算出位置 i,如果 Entry 已经存在同时 key 等于传入的 key,那么这时候直接给这个 Entry 更新为新值。如果 Entry 存在同时 key 为null,则调用 replaceStaleEntry 来更换这个 key 为空的 Entry。不断循环直到遇到为 null 的地方。最后调用 cleanSomeSlots 清理 key 为 null 的 Entry 就行了,最后返回是否清理了 Entry,接下来再判断 sz>thresgold 看是否达到了 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
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();

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

清理完空 key 的 Entry 后,如果 size 大于 3/4 的 threshold,则调用 resize 方法,每次扩容大小扩展为原来的 2 倍,然后在 for 循环里清除空 key 的 Entry,同时重新计算 key 不为空的 Entry 的 hash 值把它们放到正确的位置上。

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

// Use lower threshold for doubling to avoid hysteresis
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);
while (newTab[h] != null)
h = nextIndex(h, newLen);
newTab[h] = e;
count++;
}
}
}

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

remove

在 map 中移除一个不用的 Entry。也是先计算出 hash 值,若是第一次没有命中,就循环直到 null,在此过程中也会调用 expungeStaleEntry 清除空 key 节点。

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

为了避免内存泄露,在 get、set、remove 方法中都会将 key 为 null 的 entry 清除,entry 的 value 没有强引用了也会被 gc 回收。但前提是调用了这几个方法,所以在实际中使用完 threadlocal,为了避免内存泄露最好手动调用 remove 方法。