并发编程（十一）：ThreadLocal源码分析

亦凉 2023-06-04 10:59 5阅读 0赞

# 一，ThreadLocal概述 #

## 1，TheadLocal —— 线程隔离对象 ##

\* JDK内部提供了线程隔离对象 *ThreadLocal* ，注意该类与AQS或者JUC基本没有关系。线程对 *ThreadLocal* 操作，实际上对线程属性 *threadLocals* 进行操作，该属性是 *ThreadLocal* 内部类 *ThreadLocalMap* 的实例。*ThreadLocal* 底层通过 *ThreadLocalMap* 进行存储，具体在 *ThreadLocalMap* 内通过 *Entry* 数组进行数据保存。*Entry* 在 *ThreadLocal* 的内部实现，并通过 *WeakReference* 包装泛型 *ThreadLocal* 为弱引用（在每一次GC时，弱引用对象如果没有强引用对象引用，会被GC回收），*Entry* 以每一个 *ThreadLocal* 作为键值在 *Thread* 内进行不同的数据存储

\* 总而言之，一个 ThreadLocal 对象可以存储在多个线程的成员变量 threadLocals 中。每一个 *ThreadLocal* 在线程中的存在形式，通过包装为 Entry 对象，存储在 ThreadLocalMap 的成员变量 Entry\[\] 中。

## 2，类图 ##

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTE5NzYzODg_size_16_color_FFFFFF_t_70][]

## 3，常用API ##

// 构造
    public ThreadLocal();
    // 初始化, 初始化对象时, 可以重写此方法, 完成给每一个线程赋初值
    protected T initialValue();
    // 存储数据
    public void set(T value);
    // 获取数据
    public T get();
    // 移除数据
    public void remove();

## 4，功能DEMO ##

\* 从输出结果可以看出，各个线程保存的是操作时候的变量值，后续变量变化并没有产生影响；演示比较简单，感觉一个Map就能玩，不过在开发中，ThreadLocal 理论上是存在在整个线程周期的，所以不需要频繁传值

package com.gupao.concurrent;
    
    /**
     * @author pj_zhang
     * @create 2019-10-21 22:09
     **/
    public class ThreadLocalTest {
    
        private static volatile int count = 0;
    
        public static void main(String[] args) {
            ThreadLocal threadLocal = new ThreadLocal();
            for (int i = 0; i < 10; i++) {
                new Thread(() -> {
                    threadLocal.set(++count);
                    try {
                        Thread.sleep(100);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    System.out.println(Thread.currentThread().getName() + ": " + threadLocal.get());
                }, "THREAD_" + i).start();
            }
        }
    
    }

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTE5NzYzODg_size_16_color_FFFFFF_t_70 1][]

# 二，ThreadLocal源码分析 #

## 1，初始化 ##

\* 构造：ThreadLocal()

// 直接初始化，无特殊
    public ThreadLocal() {
    }

\* 初始化值：initialValue()

// 该方法用于重写，为每一个线程赋初值
    protected T initialValue() {
    	return null;
    }

## 2，set()源码分析 ##

\* set(T value)

public void set(T value) {
    	Thread t = Thread.currentThread();
    	// 获取当前线程的成员成员 threadLocals
    	ThreadLocalMap map = getMap(t);
    	// 成员变量不为空，说明已经初始化，给该线程的当前 ThreadLocal 赋值
    	if (map != null)
    		map.set(this, value);
    	// 成员变量为空，创建并赋值
    	else
    		createMap(t, value);
    }

\* getMap(Thread t)：获取线程成员变量 threadLocals

ThreadLocalMap getMap(Thread t) {
    	// 直接返回线程成员变量
    	return t.threadLocals;
    }

\* createMap(Thread t, T firstValue)：成员变量不存在，创建并赋值

void createMap(Thread t, T firstValue) {
    	// 直接初始化 ThreadLocalMap
    	t.threadLocals = new ThreadLocalMap(this, firstValue);
    }

\* ThreadLocalMap(ThreadLocal<?> firstKey, Object firstValue)：初始化加魔数及扩容

ThreadLocalMap(ThreadLocal<?> firstKey, Object firstValue) {
    	// 初始化table，默认数组长度为16
    	table = new Entry[INITIAL_CAPACITY];
    	// 根据斐波拉契散列获取下标值
    	int i = firstKey.threadLocalHashCode & (INITIAL_CAPACITY - 1);
    	// 构建Entry
    	table[i] = new Entry(firstKey, firstValue);
    	size = 1;
    	// 设置扩容因子，扩容因子由两方面构成
    	setThreshold(INITIAL_CAPACITY);
    }
    
    // 斐波拉契散列魔数
    private final int threadLocalHashCode = nextHashCode();
    
    private static AtomicInteger nextHashCode = new AtomicInteger();
    
    // 魔数
    private static final int HASH_INCREMENT = 0x61c88647;
    
    private static int nextHashCode() {
    	// 通过 AtomicInteger 类实现原子递增
    	return nextHashCode.getAndAdd(HASH_INCREMENT);
    }
    
    // 扩容因子，设置为2/3并取整
    private void setThreshold(int len) {
    	threshold = len * 2 / 3;
    }

\* set(ThreadLocal<?> key, Object value)：成员变量已经初始化，直接重赋值

private void set(ThreadLocal<?> key, Object value) {
    	// 获取到存储容器：Entry{]
    	Entry[] tab = table;
    	// 获取容器容量，并通过斐波拉契散列获取线程应该所在的索引下标
    	// key.threadLocalHashCode：ThreadLocal成员变量，通过Atomic原子类实现递增(CAS)
    	int len = tab.length;
    	int i = key.threadLocalHashCode & (len-1);
    
    	// 获取数组中该下标的值，判断是否已经存在元素
    	// nextIndex：获取下一个下标，通过三元表达式实现首尾相连
    	
    	// 如果Entry为空，说明当前 TheadLocal 实例不存在，同时也不可能存在在其他索引位置，分析如下：
    	// 首先GC回收失效只对ThreadLocal进行清理，不会对Entry进行清理，所以Entry只会为空，不会为null
    	// 此时如果当前 ThreadLocal 在此Entry有效时插入，则会顺序寻找下一个槽位。
    	// 此时需要做的就是从其他槽位找到当前ThreadLocal，并把当前 ThreadLocal 拉回到正确槽位
    	
    	// 其次，如果当前线程在插入 ThreadLocal 时所在槽位已经被占用，此时该 ThreadLocal 占用其他槽位，
    	// 之后如果占用该槽位的 ThreadLocal 被程序清理，注意程序清理会直接将Entry置null
    	// 清理完成后，程序会重新向下整理一次元素的下标位置，尽量将元素放在hash算出的hash位置
    	for (Entry e = tab[i];
    		 e != null;
    		 e = tab[i = nextIndex(i, len)]) {
    		ThreadLocal<?> k = e.get();
    		// 存在元素，且是当前 ThreadLocal 元素，则直接覆盖
    		if (k == key) {
    			e.value = value;
    			return;
    		}
    
    		// 存在元素，但是元素key为null
    		// 因为Entry的类定义为 static class Entry extends WeakReference<ThreadLocal<?>>
    		// 即Entry继承自包装了ThreadLocal对象的弱引用
    		// 弱引用对象在GC时如果没有强引用对象指向，则会直接被回收，所以ThreadLocal可能为null
    		// 但是在回收时候，Entry对象不会被回收，所以会出现下列情况
    		if (k == null) {
    			// 清理并存储当前节点
    			replaceStaleEntry(key, value, i);
    			return;
    		}
    	}
    	
    	// 如果当前下标未被占用，则直接初始化为当前 ThreadLocal
    	tab[i] = new Entry(key, value);
    	int sz = ++size;
    	// 清理节点，并且判断扩容因子
    	if (!cleanSomeSlots(i, sz) && sz >= threshold)
    		// 重新进行hash分配，是否进行扩容内部存在二次判断s
    		rehash();
    }

\* replaceStaleEntry(ThreadLocal<?> key, Object value, int staleSlot)：

// 该方法实现两个功能
    // 移除：移除已经失效的节点
    // 构建：构建当前线程的新节点
    private void replaceStaleEntry(ThreadLocal<?> key, Object value, int staleSlot) {
    	Entry[] tab = table;
    	int len = tab.length;
    	Entry e;
    
    	// 每一次运行都会处理一次失效数据，减少数据扩容和数据冲突
    	// 循环到上一个非null的Entry对象，并保留失效的索引下标到slotToExpunge，以备后续清理
    	int slotToExpunge = staleSlot;
    	for (int i = prevIndex(staleSlot, len);
    		 (e = tab[i]) != null;
    		 i = prevIndex(i, len))
    		if (e.get() == null)
    			slotToExpunge = i;
    
    	// 从当前节点向下找，判断每一个槽位，是否存在当前 ThreadLocal
    	// 因为当前槽位已经被占用，所以如果当前 ThreadLocal 在占用之后曾访问一次，则会重新分配槽位
    	// 此时需要遍历寻找，判断当前 ThreadLocal 是否已经初始化为 Entry
    	for (int i = nextIndex(staleSlot, len);
    		 (e = tab[i]) != null;
    		 i = nextIndex(i, len)) {
    		ThreadLocal<?> k = e.get();
    
    		// key相同，说明当前 ThreadLocal 已经初始化为Entry，并占用了其他Hash槽位
    		if (k == key) {
    			// 修改线程值
    			e.value = value;
    			// 将原槽位已经作废的slot迁移到该槽位
    			// 将当前 ThreadLocal 迁移到正确槽位
    			tab[i] = tab[staleSlot];
    			tab[staleSlot] = e;
    
    			// 如果失效的下标相等，因为当前槽位已经与下一个槽位互转
    			// 所以修改失效槽位索引
    			if (slotToExpunge == staleSlot)
    				slotToExpunge = i;
    			// expungeStaleEntry: 消除当前槽位信息，此处为彻底消除，
    			// cleanSomeSlots: 清除部分失效节点，此处没有全遍历清除
    			cleanSomeSlots(expungeStaleEntry(slotToExpunge), len);
    			return;
    		}
    
    		// 槽位失效，并且相等，转换为下一个槽位索引
    		if (k == null && slotToExpunge == staleSlot)
    			slotToExpunge = i;
    	}
    
    	// 如果没有找到key，说明当前 ThreadLocal 并没有初始化，直接初始化
    	tab[staleSlot].value = null;
    	tab[staleSlot] = new Entry(key, value);
    
    	// 存在失效节点，清理掉！
    	if (slotToExpunge != staleSlot)
    		cleanSomeSlots(expungeStaleEntry(slotToExpunge), len);
    }

\* expungeStaleEntry(int staleSlot)：移除失效节点，并返回下一个有效节点索引，该方法为底层方法

// staleSlot：参数表示当前要清理的失效节点
    private int expungeStaleEntry(int staleSlot) {
    	Entry[] tab = table;
    	int len = tab.length;
    
    	// 注意此处清理节点是对Entry直接置空
    	tab[staleSlot].value = null;
    	tab[staleSlot] = null;
    	size--;
    
    	// 从当前节点向下reHash，直到遇到空，
    	// Entry元素插入，如果遇到Hash冲突，会顺序向下寻找节点，此处向下reHash，目的是为了把顺序向下存储的节点拉回到合适位置
    	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 {
    			// 节点未失效，重新计算节点hash值，并判断下标位置是否与当前位置一致
    			// 位置不一致时，将元素拉到对应的合适位置
    			// 此处也就是为什么当位置元素为空时，直接插入，不需要再循环判断是否已经存在
    			int h = k.threadLocalHashCode & (len - 1);
    			if (h != i) {
    				tab[i] = null;
    				while (tab[h] != null)
    					h = nextIndex(h, len);
    				tab[h] = e;
    			}
    		}
    	}
    	// 此处返回的i表示向下遇到的第一个null节点的下表
    	return i;
    }

\* cleanSomeSlots(int i, int n)：从指定下标索引开始向下位移一位找失效节点并移除，该方法为底层方法

// i表示开始清理的索引下标，len表示当前ThreadLocal容器长度
    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];
    		// 如果节点已经失效，对n重新赋初值
    		if (e != null && e.get() == null) {
    			n = len;
    			removed = true;
    			// 移除当前节点，返回的i为当前节点的下一个null节点索引
    			i = expungeStaleEntry(i);
    		}
    	// 节点全部有效的情况下，最多跑 n = 2 ^ m 的 m 次
    	} while ( (n >>>= 1) != 0);
    	return removed;
    }

\* rehash()：扩容

private void rehash() {
    	// 清理所有失效数据
    	expungeStaleEntries();
    
    	// 判断是否需要扩容
    	// threshold = len * 2 / 3
    	// threshold - threshold / 4 = size * 1 / 2
    	// 所以一半扩容，扩容一倍，用较低的阈值来保证内部判断逻辑更快，减少迟滞
    	if (size >= threshold - threshold / 4)
    		resize();
    }

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

\* resize()：扩容，代码中并没有看到移除后减小容量的部分，应该是不支持

private void resize() {
    	Entry[] oldTab = table;
    	int oldLen = oldTab.length;
    	// 两倍扩容
    	int newLen = oldLen * 2;
    	Entry[] newTab = new Entry[newLen];
    	int count = 0;
    
    	// 将旧容器中的元素全部拿出来，重新根据新容器长度计算下标位置，并构建
    	// 如果hash冲突，同样向下寻找节点
    	for (int j = 0; j < oldLen; ++j) {
    		Entry e = oldTab[j];
    		if (e != null) {
    			ThreadLocal<?> k = e.get();
    			if (k == null) {
    				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;
    }

## 3，get()源码分析 ##

\* get()

public T get() {
    	Thread t = Thread.currentThread();
    	// 获取线程属性 threadLocals
    	ThreadLocalMap map = getMap(t);
    	if (map != null) {
    		// 从 ThreadLocalMap 内部 Entry[] 中，根据当前请求的Threadlocal实例，获取对应Entry
    		ThreadLocalMap.Entry e = map.getEntry(this);
    		if (e != null) {
    			// Entry 不为空，直接获取对应值
    			@SuppressWarnings("unchecked")
    			T result = (T)e.value;
    			return result;
    		}
    	}
    	// map为空或者entry为空，构建并填充初值
    	return setInitialValue();
    }

\* getEntry(ThreadLocal<?> key)：从 *ThreadLocalMap* 中获取对应 *Entry*

private Entry getEntry(ThreadLocal<?> key) {
    	// 获取索引位置
    	int i = key.threadLocalHashCode & (table.length - 1);
    	// Entry存在且key为当前 ThreadLocal，则匹配成功
    	Entry e = table[i];
    	if (e != null && e.get() == key)
    		return e;
    	else
    		// key值不为当前 ThreadLocal，则说明已经被占用，向后继续寻找
    		return getEntryAfterMiss(key, i, e);
    }

\* getEntryAfterMiss(ThreadLocal<?> key, int i, Entry e)

private Entry getEntryAfterMiss(ThreadLocal<?> key, int i, Entry e) {
    	Entry[] tab = table;
    	int len = tab.length;
    
    	// e不为空，说明槽点被占用，继续向后寻找
    	// 找到下一个为空，说明当前段不存在，则不存在
    	while (e != null) {
    		ThreadLocal<?> k = e.get();
    		if (k == key)
    			return e;
    		// 失效，直接移除，则后续重新赋值后e为null
    		if (k == null)
    			expungeStaleEntry(i);
    		// 未失效，说明存在，继续下一个索引寻找
    		else
    			i = nextIndex(i, len);
    		e = tab[i];
    	}
    	// e为空，说明线程对应key确实不存在，直接返回null
    	return null;
    }

\* setInitialValue()：获取 Entry 失败，ThreadLocalMap 或者 Entry 为空，赋初值

private T setInitialValue() {
    	// 获取初值
    	T value = initialValue();
    	Thread t = Thread.currentThread();
    	ThreadLocalMap map = getMap(t);
    	// map存在，直接添加，map不存在，构建map后添加
    	if (map != null)
    		map.set(this, value);
    	else
    		createMap(t, value);
    	// 返回结果值
    	return value;
    }

## 4，remove()源码分析 ##

\* remove()

public void remove() {
    	// 获取线程对应的 ThreadLocalMap
    	ThreadLocalMap m = getMap(Thread.currentThread());
    	if (m != null)
    		// map存在，直接移除当前 ThreadLocal
    		m.remove(this);
    }

\* remove(ThreadLocal<?> key)

private void remove(ThreadLocal<?> key) {
    	Entry[] tab = table;
    	int len = tab.length;
    	// 获取下标位置
    	int i = key.threadLocalHashCode & (len-1);
    	// 从下标位置开始，向下寻找，获取指定的 ThreadLocal后直接移除
    	for (Entry e = tab[i];
    		 e != null;
    		 e = tab[i = nextIndex(i, len)]) {
    		if (e.get() == key) {
    			e.clear();
    			expungeStaleEntry(i);
    			return;
    		}
    	}
    }

# 三，ThreadLocal工具类 #

package com.gupao.concurrent;
    
    import java.util.*;
    
    public final class ThreadLocalUtil {
        
        private static final ThreadLocal<Map<String, Object>> threadLocal = new ThreadLocal() {
            protected Map<String, Object> initialValue() {
                return new HashMap(16);
            }
        };
    
        public static Map<String, Object> getThreadLocal() {
            return threadLocal.get();
        }
    
        public static <T> T get(String key) {
            Map map = (Map) threadLocal.get();
            return (T) map.get(key);
        }
    
        public static <T> T get(String key, T defaultValue) {
            Map map = (Map) threadLocal.get();
            return (T) map.get(key) == null ? defaultValue : (T) map.get(key);
        }
    
        public static void set(String key, Object value) {
            Map map = (Map) threadLocal.get();
            map.put(key, value);
        }
    
        public static void set(Map<String, Object> keyValueMap) {
            Map map = (Map) threadLocal.get();
            map.putAll(keyValueMap);
        }
    
        public static void remove() {
            threadLocal.remove();
        }
    
        public static <T> T remove(String key) {
            Map map = (Map) threadLocal.get();
            return (T) map.remove(key);
        }
    
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTE5NzYzODg_size_16_color_FFFFFF_t_70]: /images/20230601/13a3de333a72408b880892a671b58a2f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTE5NzYzODg_size_16_color_FFFFFF_t_70 1]: /images/20230601/d5fda262a10f4b3da7d03dc40df5bfb5.png