JDK1.8源码阅读笔记之java.lang.ThreadLocal

ThreadLocal类简介

官方介绍

 * This class provides thread-local variables.  These variables differ from

* their normal counterparts in that each thread that accesses one (via its

* {@code get} or {@code set} method) has its own, independently initialized

* copy of the variable.  {@code ThreadLocal} instances are typically private

* static fields in classes that wish to associate state with a thread (e.g.,

* a user ID or Transaction ID).

根据《Java并发编程实战》书中的解释：

//官方例子
public class ThreadId {
// Atomic integer containing the next thread ID to be assigned
private static final AtomicInteger nextId = new AtomicInteger(0);
// Thread local variable containing each thread's ID
private static final ThreadLocal<Integer> threadId =
new ThreadLocal<Integer>() {
@Override
protected Integer initialValue() {
return nextId.getAndIncrement();
}
};
// Returns the current thread's unique ID, assigning it if necessary
public static int get() {
return threadId.get();
}

}

线程局部变量的创建

这个方法采用Lambda方式传入实现了 Supplier 函数接口的参数。

/**
* Creates a thread local variable. The initial value of the variable is
* determined by invoking the {@code get} method on the {@code Supplier}.
*
* @param <S> the type of the thread local's value
* @param supplier the supplier to be used to determine the initial value
* @return a new thread local variable
* @throws NullPointerException if the specified supplier is null
* @since 1.8
*/
public static <S> ThreadLocal<S> withInitial(Supplier<? extends S> supplier) {
return new SuppliedThreadLocal<>(supplier);
}// Supplier接口源码
@FunctionalInterface
public interface Supplier<T> {

/**
* Gets a result.
*
* @return a result
*/
T get();
}

java.util.function.Supplier 接口仅包含一个无参的方法: T get() 。用来获取一个泛型参数指定类型的对象数据。由于这是一个函数式接口，这也就意味着对应的Lambda表达式需要“对外提供”一个符合泛型类型的对象数据

get方法

public T get() {
//获取当前线程
Thread t = Thread.currentThread();
//根据当前线程获取ThreadLocalMap ①
ThreadLocalMap map = getMap(t);
if (map != null) {
// 如果map不为空,则根据当前TL获取存取的V,根据泛型T强转返回。 ②
ThreadLocalMap.Entry e = map.getEntry(this);
if (e != null) {
@SuppressWarnings("unchecked")
T result = (T)e.value;
return result;
}
}
//如果e为null，返回初始化值 ③
return setInitialValue();
}

①ThreadLocalMap是ThreadLocal的静态内部类，实现类似map的功能，下面是TLM的核心源码

//Entry是TLM的静态内部类,实现KV存储功能
//使用弱引用
static class Entry extends WeakReference<ThreadLocal<?>> {
/** The value associated with this ThreadLocal. */
Object value;

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


//TLM初始大小16
private static final int INITIAL_CAPACITY = 16;


//Entry数组
private Entry[] table;


//记录table大小
private int size = 0;

//扩容阈值
private int threshold;


//负载因子是2/3 阈值是len*2/3
private void setThreshold(int len) {
threshold = len * 2 / 3;
}

//构造方法
ThreadLocalMap(ThreadLocal<?> firstKey, Object firstValue) {
//新建初始容量为16的数组
table = new Entry[INITIAL_CAPACITY];
//hash碰撞计算存放下标,初始化容量需是2的倍数
int i = firstKey.threadLocalHashCode & (INITIAL_CAPACITY - 1);
table[i] = new Entry(firstKey, firstValue);
size = 1;
//设置阈值
setThreshold(INITIAL_CAPACITY);
}

注意:Entry 使用弱引用是为了可以在垃圾回收时将无用的TL回收掉,如果使用强引用的话,如果TLM未进行回收,则一直会有强引用指向TL,导致TL无法被回收,虽然弱引用优于强引用,但是仍然无法解决内存泄露的问题。

• 解决TL内存泄露只有两种方式
• a.调用TL的remove方法(使用TL,必须在使用完成后调用remove)
• b.结束线程，TL声明周期是同线程声明周期的，线程结束，TL自然可以被回收（该方法不可控,无法确定线程结束时间）

②getEntity方法源码

​方法比较简单,根据key计算存在table中的下标,如果对应entry!=null并且entry.key==传入的key，直接返回，否则调用getEntryAfterMiss获取，接下来看一下getEntryAfterMiss方法

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

通过线性探测法查找k相同的entry，如果遇到k为null的则调用expungeStaleEntry清理

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

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;
//计算下标
a. int i = key.threadLocalHashCode & (len-1);
//遍历tab,给K V赋值
for (Entry e = tab[i];e != null;e = tab[i = nextIndex(i, len)]) {
//获取tab[i]对应的key
ThreadLocal<?> k = e.get();
//如果已有K,则将V更新
if (k == key) {
e.value = value;
return;
}
//如果key为null,调用replaceStaleEntry替换旧值
if (k == null) {
c. replaceStaleEntry(key, value, i);
return;
}
}

tab[i] = new Entry(key, value);
int sz = ++size;
//不能进行回收且size大于阈值的情况下扩容，扩容是在先放入元素后扩容
d. if (!cleanSomeSlots(i, sz) && sz >= threshold)
rehash();
}

a 、threadLocalHashCode 的赋值过程如下

private final int threadLocalHashCode = nextHashCode();

private static AtomicInteger nextHashCode = new AtomicInteger();

//0x61c88647是斐波那契散列乘数,保证hash可以均匀分布
private static final int HASH_INCREMENT = 0x61c88647;

private static int nextHashCode() {//初始化时自增,增量为0x61c88647
return nextHashCode.getAndAdd(HASH_INCREMENT);
}

b、可能会出现hash冲突，需要重新计算下标

c、replaceStaleEntry源码

private void replaceStaleEntry(ThreadLocal<?> key, Object value,
int staleSlot) {//staleSlot是set时K为null的下标
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).
//TL使用的是线性探测法解决冲突，通过前序查找是否还有K为null的脏数据
int slotToExpunge = staleSlot;//slotToExpunge 指向key为null的脏数据 需要进行清理
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.
//找到k相同的entry，value重新赋值，交换位置
if (k == key) {
e.value = value;

tab[i] = tab[staleSlot];
tab[staleSlot] = e;

// Start expunge at preceding stale entry if it exists
//slotToExpunge == staleSlot说明没有别的K为null的数据，由于上一步已经将i和staleSlot进行了数据交换，i下标对应的数据就是set时key为null的数据
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.
//如果前序查找没找到其他K为null的且后序查找有k为null的就将对应i清除
if (k == null && slotToExpunge == staleSlot)
slotToExpunge = i;
}

// If key not found, put new entry in stale slot
//没找到k，则将传入的k为null的下标的value改为null，相当于去除其引用，便于垃圾回收
tab[staleSlot].value = null;
//重新new entry放入要set的kv
tab[staleSlot] = new Entry(key, value);

// If there are any other stale entries in run, expunge them
//不相等说明还有其他k为null的数据，需要清除
if (slotToExpunge != staleSlot)
e. cleanSomeSlots(expungeStaleEntry(slotToExpunge), len); e
}

e核心方法有两个expungeStaleEntry和cleanSomeSlots，首先来看expungeStaleEntry //传入的staleSlot即k为null的下标

private int expungeStaleEntry(int staleSlot) {
Entry[] tab = table;
int len = tab.length;
//将staleSlot对应kv从数组移除
// 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();
//如果有k为null的移除
if (k == null) {
e.value = null;
tab[i] = null;
size--;
} else {
//如果k重新计算hash值后和i不通，则将i置为null，并将entry移至新下标
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;
}

接下来看cleanSomeSlots代码

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];
//找到e不为null但k为null的数据调用expungeStaleEntry进行清理
if (e != null && e.get() == null) {
n = len;
removed = true;
i = expungeStaleEntry(i);
}
} while ( (n >>>= 1) != 0);
return removed;
}

接下来看下rehash源码

private void rehash() {
//将整个table进行回收清理，
expungeStaleEntries();
//如果回收后的size仍大于3/4的阈值 进行扩容
// Use lower threshold for doubling to avoid hysteresis
if (size >= threshold - threshold / 4)
resize();
}

方法比较简单，先进行整个table的清理回收，如果size仍大于3/4的阈值 进行扩容，如何进行扩容呢，可以看下源码

private void resize() {
Entry[] oldTab = table;
int oldLen = oldTab.length;
//直接扩容2倍
int newLen = oldLen * 2;
Entry[] newTab = new Entry[newLen];
int count = 0;
//遍历旧数组，重新计算hash值放入newTable,遍历过程中如果发现k为null的
//将其value置为null便于gc
for (Entry e : oldTab) {
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方法，切记使用TL时，必须在使用结束后进行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();
//清理table[i]
expungeStaleEntry(i);
return;
}
}
} 【本文转自：韩国服务器 http://www.yidunidc.com处的文章,转载请说明出处】