线程隔离,ThreadLocal之诞生

Posted 毛奇志

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了线程隔离,ThreadLocal之诞生相关的知识,希望对你有一定的参考价值。

一、前言

在Java多线程模块中,ThreadLocal是比较重要的知识点,虽然ThreadLocal类位于java.lang包,但是这个类基本上仅用于多线程。

二、ThreadLocal类

2.1 ThreadLocal类

由来:一般的变量是多个线程共享,如果想一个线程独享一个变量,就需要用到ThreadLocal类。

含义:ThreadLocal直译为线程局部变量,意思是ThreadLocal在每个线程中都创建了一个变量的副本,不同线程拥有的副本互不影响。

使用场景:
① 在进行对象跨层传递的时候,可以避免多次传递,打破层次间的约束;
② 线程间数据隔离;
③ 进行事务操作,用于存储线程事务信息;
④ 数据库连接,Session会话管理。

2.2 LocalThread类的结构

LocalThread类结构,如下:

(1) 一个线程一个ThreadLocalMap:每个Thread维护着一个ThreadLocalMap的引用,变量副本存储在线程自己的ThreadLocalMap中;

(2) ThreadLocalMap结构:ThreadLocalMap是ThreadLocal的内部类,用Entry来存储key-value,键值为ThreadLocal对象,value为线程变量;

(3) ThreadLocal仅仅作为key:ThreadLocal本身并不存储值,它只是作为一个key来让线程从ThreadLocalMap获取value。

2.3 ThreadLocal变量的线程隔离性

既然ThreadLocal的作用是每一个线程创建一个副本,我们使用一个例子来验证一下。创建两个线程,线程t1设置var值为20,线程t2设置var值为15,分别输出var值,运行结果如下:

public class Demo 
    private static ThreadLocal<Integer> var = new ThreadLocal<>();  //构造函数
    public static void main(String[] args) 
        Thread t1 = new Thread(()->
            var.set(20);   // 开发者调用set()方法
            System.out.println(Thread.currentThread().getName() + ":设置var值为20");
            for (int i = 0; i < 3; i++) 
                System.out.println(Thread.currentThread().getName() + ":获取var值为" + var.get());   // 开发者调用get()方法
               try 
                    Thread.sleep(1000);
                 catch (InterruptedException e) 
                    e.printStackTrace();
                
           
        , "Thread1");

        Thread t2 = new Thread(()->
            var.set(15);  //  开发者调用set()方法
            for (int i = 0; i < 3; i++) 
               System.out.println(Thread.currentThread().getName() + ":获取var值为" + var.get());  // // 开发者调用get()方法
                try 
                    Thread.sleep(1000);
                 catch (InterruptedException e) 
                    e.printStackTrace();
                
            
        , "Thread2");

        t1.start();
        t2.start();
    

运行结果:

Thread1:设置var值为20
Thread1:获取var值为20
Thread2:获取var值为15
Thread2:获取var值为15
Thread1:获取var值为20
Thread2:获取var值为15
Thread1:获取var值为20

根据结果可以看出,ThreadLocal类变量在不同线程保存的变量副本是互不影响的,是相互隔离的。

三、元素值操作

3.1 设置元素值

3.1.1 set()方法

public void set(T value) 
    Thread t = Thread.currentThread();    // 当前线程
    ThreadLocalMap map = getMap(t);     // 获取Map
    if (map != null)      // map不为空,直接设置值
        map.set(this, value);
    else                  // map为空,创建map
        createMap(t, value);

对于set()方法的解释:
(1) 获取当前线程,并获取当前线程的ThreadLocalMap实例(从getMap(Thread t)中很容易看出来).
(2) 如果获取到的map实例不为空,调用map.set()方法,否则调用createMap(t, value)实例化map.

3.1.2 set(ThreadLocal<?> key, Object value)方法

我们来看下map.set(this, value)方法的具体实现:

private void set(ThreadLocal<?> key, Object value) 
    Entry[] tab = table;
    int len = tab.length;
    // 计算key的索引值 
    int i = key.threadLocalHashCode & (len-1);
    // 根据获取到的索引进行循环,如果当前索引上的table[i]不为空,在没有return的情况下,就使用nextIndex()获取下一个(线性探测法)
    // 下一个不为null,且没有return,继续下一个,知道e==null跳出循环
    for (Entry e = tab[i];
         e != null;
         e = tab[i = nextIndex(i, len)]) 
        ThreadLocal<?> k = e.get();  // 确定当前的k,下面两个if判断
        // table[i]上key不为空,并且和方法参数key相同,就找到了,更新value,结束函数
        if (k == key) 
            e.value = value;
            return;
        
        // table[i]上的key为空,说明被回收了(弱引用)。
        // 这个时候说明该table[i]可以重新使用,用新的key-value将其替换,并删除其他无效的entry
        if (k == null) 
            replaceStaleEntry(key, value, i);   //分支方法里面详细解释
            return;
        
    
    // e==null,跳出循环,线性探测结束
    // 找到为空的插入位置,插入值,在为空的位置插入需要对size进行加1操作
    tab[i] = new Entry(key, value);  // 方法参数中的key,value新建Entry,放到i
    int sz = ++size;  
    // cleanSomeSlots用于清除那些e.get()==null,也就是table[index] != null && table[index].get()==null
    // 这种数据key关联的对象已经被回收,所以这个Entry(table[index])可以被置null。
    // 如果没有清除任何entry,并且当前使用量达到了负载因子所定义(长度的2/3),那么进行rehash()
    if (!cleanSomeSlots(i, sz) && sz >= threshold)
        rehash();

在插入过程中,根据 ThreadLocal 对象的索引值,定位到 table 中的位置 i,过程如下:

(1) 如果当前索引上的table[i]为空(如果e==null则跳出循环),那么正好,就初始化一个 Entry 对象放在位置 i 上;

(2) 循环内的两个if判断:不巧,位置 i 已经有 Entry 对象了,如果这个 Entry 对象的 key 正好是即将设置的 key,则更新 Entry 中的 value值。如果Entry对象的key为null,则说明该table[i]可以重新使用,用新的key-value将其替换,并删除其他无效的entry;

(3) 不断next循环:很不巧,位置 i 的 Entry 对象,和即将设置的 key 没关系,那么只能找下一个空位置。

3.1.3 threadLocalHashCode变量

每个 ThreadLocal 对象都有一个 hash 值 threadLocalHashCode,每初始化一个 ThreadLocal对象,hash 值就增加一个固定的大小 0x61c88647,其主要目的就是为了让哈希码能均匀的分布在2的n次方的数组里, 也就是Entry[] table中。

private static final int HASH_INCREMENT = 0x61c88647;
private final int threadLocalHashCode = nextHashCode();

private static int nextHashCode() 
    return nextHashCode.getAndAdd(HASH_INCREMENT);

ThreadLocalMap使用线性探测法来解决哈希冲突,线性探测法的地址增量di = 1, 2, ... , m-1,其中,i为探测次数。该方法一次探测一个地址,直到有空的地址后插入,若整个空间都找不到空余的地址,则产生溢出。

假设当前table长度为16,也就是说如果计算出来key的hash值为14,如果table[14]上已经有值,并且其key与当前key不一致,那么就发生了hash冲突,这个时候将14加1得到15,取table[15]进行判断,这个时候如果还是冲突会回到0,取table[0],以此类推,直到可以插入。

按照上面的描述,可以把table看成一个环形数组。先看一下线性探测相关的代码,从中也可以看出来table实际是一个环:

private static int nextIndex(int i, int len) 
    // 如果当前i不是最后一个,返回i+1,表示下一个,
    // 值最后一个(就是i==len-1,最后一个),返回0,第一个,所以是环状
    return ((i + 1 < len) ? i + 1 : 0);   

可以发现,set()方法如果冲突严重的话,效率会很低。

3.2 读取元素值

3.2.1 get()方法

public T get() 
    // 跟set方法类似,获取对应线程中的ThreadLocalMap实例
    Thread t = Thread.currentThread();
    ThreadLocalMap map = getMap(t);   // 根据线程得到ThreadLocalMap实例 good 
    if (map != null) 
        ThreadLocalMap.Entry e = map.getEntry(this);  // 根据ThreadLocalMap实例得到ThreadLocalMap.Entry
        if (e != null)   // 获得的Entry不为null,返回value(即 ThreadLocalMap.Entry)
            @SuppressWarnings("unchecked")
            T result = (T)e.value;    
            return result;
        
    
    // 为空返回初始化值
    return setInitialValue();

对于get()方法解释:
(1) 获取当前线程,并获取当前线程的ThreadLocalMap实例(从getMap(Thread t)中很容易看出来);
(2) 如果获取到的map实例不为空,调用map.getEntry(this)获取Entry对象,否则调用setInitialValue()实例化map。

3.2.2 getEntry()方法

private Entry getEntry(ThreadLocal<?> key) 
    // 根据key计算索引,获取entry
    int i = key.threadLocalHashCode & (table.length - 1);
    Entry e = table[i];    // 根据索引i得到entry
    if (e != null && e.get() == key)   // 这里两个条件要同时满足
        return e;   //返回entry
    else    // 三种情况 e==null || e.get()!=key
        // 因为用的是线性探测,所以往后找还是有可能能够找到目标Entry的
        return getEntryAfterMiss(key, i, e);   // 将方法参数中key传递过来,key得到的i传递过来,i对应的e传递过来,总结,就是key i entry 

3.2.3 getEntryAfterMiss()方法

// 通过计算出来的key找不到对应的value时使用这个方法,
// 结束getEntryAfterMiss()方法两种方式,不断后面找,直到找到或者都找不到e==null,返回null

private Entry getEntryAfterMiss(ThreadLocal<?> key, int i, Entry e) 
    Entry[] tab = table;
    int len = tab.length;
    while (e != null)     // 方法参数的e不为null,表示是通过e.get()!=key进来的
        ThreadLocal<?> k = e.get();   // e.get() 得到对应的k  
        if (k == key)   
            return e;    //  找到了,直接返回(第一个e不会走这一条路,后面 n-1 个才可能走这一条路)
        if (k == null)
          // 如果当前e中的k为null,清除无效的entry   
            expungeStaleEntry(i);   // 传入索引,清除指定的entry
        else
          // 基于线性探测法向后扫描,找到下一个i,上面说了,是环状
            i = nextIndex(i, len);    
        e = tab[i];   // 更新i后更新e
    
    return null;   //如果是因为e==null进来的,这个方法也无法处理,只能返回null

3.3 删除元素值

ThreadLocal类中通过remove()方法来删除指定的key,删除指定的键值对,remove()方法源码如下:

public void remove() 
   // 根据线程得到ThreadLocalMap,这是可以理解的,因为一个线程对应一个ThreadLocal,而一个ThreadLocal对应一个ThreadLocalMap
    ThreadLocalMap m = getMap(Thread.currentThread());
    if (m != null)
        m.remove(this);  // 然后根据ThreadLocalMap.remove(),将this传入

private void remove(ThreadLocal<?> key) 
    Entry[] tab = table;
    int len = tab.length;
     // 计算索引
    int i = key.threadLocalHashCode & (len-1);
     // 进行线性探测,查找正确的key
    for (Entry e = tab[i];e != null;e = tab[i = nextIndex(i, len)]) 
        if (e.get() == key)   // 找到了
            // 调用weakrefrence的clear()清除引用
            e.clear();
             // 连续段清除
            expungeStaleEntry(i);
            return;
        
    

public void clear() 
    this.referent = null;   // 代码层面的清空就是设置为null,设置为null后,后面Java内存回收一定会处理的

remove()在有上面了解后可以说极为简单了,就是找到对应的table[],调用weakrefrence的clear()清除引用,然后再调用expungeStaleEntry()进行清除。

四、键值对初始化和清理

4.1 ThreadLocalMap初始化

问题:ThreadLocalMap是何时初始化的?
回答:调用createMap(t, value)的时候初始化的,在set()方法和get()方法中都有调用createMap(t, value)方法,所以可以初始化ThreadLocalMap。
(1) 第一次调用set()方法时,如果ThreadLocalMap为null,则会调用createMap(t, value)方法对ThreadLocalMap进行初始化.
(2) 调用get()方法时,如果ThreadLocalMap为null,则会调用setInitialValue()方法对ThreadLocalMap进行初始化,最终其实也是调用了createMap(t, value)方法.

4.1.1 set()方法中的初始化

第一次调用set()方法时,如果ThreadLocalMap为null,则会调用createMap(t, value) 方法对ThreadLocalMap进行初始化,如下:

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

 
// 1、新建INITIAL_CAPACITY大小的Entry数组为table
// 2、在里面插入firstKey, firstValue
// 3、设置扩容阈值为初始容量的三分之二
ThreadLocalMap(ThreadLocal<?> firstKey, Object firstValue) 
    // 初始化一个大小 16 的 Entry 数组
    // 表的大小始终为 2 的幂次
   table = new Entry[INITIAL_CAPACITY];
   // 计算 key 的 的 hash 
   int i = firstKey.threadLocalHashCode & (INITIAL_CAPACITY - 1);
   table[i] = new Entry(firstKey, firstValue);
   size = 1;
   // 设定扩容阈值
   setThreshold(INITIAL_CAPACITY);


private void setThreshold(int len) 
    threshold = len * 2 / 3;   // 阈值为三分之二

关于& (INITIAL_CAPACITY - 1),这是取模的一种方式,对2的幂作为模数取模,用此代替%(2^n),这也就是为啥Entry的容量必须为2的幂。

4.1.2 get()方法中的初始化

调用get()方法时,如果ThreadLocalMap为null,则会调用setInitialValue()方法对ThreadLocalMap进行初始化,最终其实也是调用了createMap(t, value)方法,如:

private T setInitialValue() 
    // 获取初始化值,默认为null(如果没有子类进行覆盖)
    T value = initialValue();
    Thread t = Thread.currentThread();
    ThreadLocalMap map = getMap(t);
    // 不为空不用再初始化,直接调用set操作设值
    if (map != null)
        map.set(this, value);
    else
        // 第一次初始化,createMap在上面有介绍过
        createMap(t, value);    // createMap(t,null); good  
    return value;


protected T initialValue() 
    return null;

4.2 替换无效Entry

replaceStaleEntry()方法用来替换无效Entry,该方法的唯一调用是在set()方法中被调用,如下:

replaceStaleEntry()方法源码解析如下:

private void replaceStaleEntry(ThreadLocal<?> key, Object value,int staleSlot) 
 //  key和value是要设置的新值
 // staleSlot 表示不新鲜的位置,该位置k==null,要替换swap 掉这个元素
    Entry[] tab = table;
    int len = tab.length;
    Entry e;
     /**
     * 根据传入的无效entry的位置(staleSlot),向前扫描
     * 一段连续的entry(这里的连续是指一段相邻的entry并且table[i] != null),
     * 直到找到一个无效entry(k==null),或者扫描完也没找到
     */
    int slotToExpunge = staleSlot;   //  之后用于清理的起点,传递过来的参数确定   slotToExpunge  要删除的位置 
    for (int i = prevIndex(staleSlot, len); (e = tab[i]) != null; i = prevIndex(i, len))
        if (e.get() == null)   // 找一个无效entry  e.get==null
            slotToExpunge = i;   //不断向前移动,更新slotToExpunge  

    // 向后扫描一段连续的entry
      
    for (int i = nextIndex(staleSlot, len);(e = tab[i]) != null;i = nextIndex(i, len)) 
        ThreadLocal<?> k = e.get();
        // 如果找到了key,将其与传入的无效entry替换,也就是与table[staleSlot]进行替换
         
        if (k == key)    //找到了  
            e.value = value;   //因为找到了,所以设置value,参数中的value(要设置的value)赋值给tab[i].value,这里设置e.value,下面要  tab[staleSlot] = e;

            tab[i] = tab[staleSlot];   //无效的赋值给tab[i]
            tab[staleSlot] = e;    //当前的赋值给无效的

            // 小结向前扫描结果 
            // slotToExpunge 如果向前查找没有找到无效entry,则更新slotToExpunge为当前值i
            if (slotToExpunge == staleSlot) //向前没找到
                slotToExpunge = i;
                        
            // return之前要清空,所以调用cleanSomeSlots()方法
            cleanSomeSlots(expungeStaleEntry(slotToExpunge), len);
            return;  // 这里是因为k==key 找到而结束
           
         // 如果向前查找没有找到无效entry,并且当前向后扫描的entry无效,则更新slotToExpunge为当前值i         
        if (k == null && slotToExpunge == staleSlot)
            slotToExpunge = i;   // 向前扫描和向后扫描都没有成果
    
    // 跳出了前进后退的循环,就是前面和后面都是e==null  e=null 
    // 如果没有找到key(向后扫描找到key就return了),也就是说key之前不存在table中
    // 就直接最开始的无效entry——tab[staleSlot]上直接新增即可
     
    tab[staleSlot].value = null;
    tab[staleSlot] = new Entry(key, value);
     // slotToExpunge != staleSlot,说明存在其他的无效entry需要进行清理 good
     // 小结向前扫描结果 slotToExpunge 
     //函数结束之前要清空,所以调用cleanSomeSlots()方法
    if (slotToExpunge != staleSlot)
        cleanSomeSlots(expungeStaleEntry(slotToExpunge), len);  //清理

4.3 启发式地清理slot

cleanSomeSlots()方法的作用是启发式地清理slot,源码中,这个方法在replaceStaleEntry()被调用了0两次,在set()中被调用了一次,该方法源码如下:

/**
  * 启发式的扫描清除,扫描次数由传入的参数n决定
  *
  * @param i 从i向后开始扫描(不包括i,因为索引为i的Slot肯定为null)
  * @param n 控制扫描次数,正常情况下为 log2(n) ,
  *          如果找到了无效entry,会将n重置为table的长度len,进行段清除。
  *          map.set()调用的时候传入的是元素个数,replaceStaleEntry()调用的时候传入的是table的长度len
  */
private boolean cleanSomeSlots(int i, int n) 
    boolean removed = false;
    Entry[] tab = table;
    int len = tab.length;
    do 
        i = nextIndex(i, len);  // 从i得到下一个i
        Entry e = tab[i];   // 从i得到e   Entry e = tab[nextIndex(i, len)];
        if (e != null && e.get() == null)    // e不为null key为null 就知道了无效entry 注意,无效entry的判断依据是key==null
           // 重置n为len
            n = len;
            removed = true;
            // 依然调用expungeStaleEntry来进行无效entry的清除
            i = expungeStaleEntry(i);   // expungeStaleEntry方法返回一个让e==null对应的i
        
     while ( (n >>>= 1) != 0);   // 无符号的右移动,可以用于控制扫描次数在log2(n)
    return removed;   //true 删除了  false 未删除

正常情况下如果 log n 次扫描没有发现无效 slot,函数就结束了,返回false。但是如果发现了无效的 slot,将 n 置为 table 的长度 len,做一次连续段的清理,再从下一个空的 slot 开始继续扫描。如果发现了无效的 slot,将 n 置为 table 的长度 len,做一次连续段的清理,再从下一个空的 slot 开始继续扫描。

cleanSomeSlots()方法有两处地方会被调用,一处是map.set()会被调用,另外个是在替换无效slot时replaceStaleEntry()会被调用,区别是前者传入的 n 为元素个数,后者为 table 的容量。

4.4 连续段清除

expungeStaleEntry()方法用于连续段清除,源码中,这个方法在cleanSomeSlots()、expungeStaleEntries()、 getEntryAfterMiss()、remove()、replaceStaleEntry()多个方法中被调用。

expungeStaleEntry()方法的命名比较清晰,expunge英文意思是“清除”,staleEntry英文意思是“无效entry”,合并起来就是“清理无效entry”。

expungeStaleEntry()方法,源码如下:

/**
 * 连续段清除
 * 1、根据传入的staleSlot,清理对应的无效entry——table[staleSlot],
 * 2、并且根据当前传入的staleSlot,向后扫描一段连续的entry(这里的连续是指一段相邻的entry并且table[i] != null),
 * 3、对可能存在hash冲突的entry进行rehash,并且清理遇到的无效entry.
 *
 * @param staleSlot key为null,需要无效entry所在的table中的索引
 * @return 返回下一个为空的solt的索引。
 */
private int expungeStaleEntry(int staleSlot) 
    Entry[] tab = table;
    int len = tab.length;

    // 1、清理staleSlot指定的无效entry,置空,交给GC处理 
    tab[staleSlot].value = null;  
    tab[staleSlot] = null;
    // 1、同时,size减1,置空后table的被使用量减1
    size--;

    // Rehash until we encounter null  // 不断rehash直到遇到null
    Entry e;
    int i;
    /**
     * 从staleSlot开始向后扫描一段连续的entry,这就是连续的含义
     */
    for (i = nextIndex(staleSlot, len);   // 第一次使用staleSlot得到下一个i
         (e = tab[i]) != null;    // e!=null  就不会跳出循环
         i = nextIndex(i, len))    // n-1次都是使用i得到下一个i
        ThreadLocal<?> k = e.get();  // 第一个k
        // 如果遇到key为null,表示无效entry,置空,交给GC处理
        if (k == null)   //判断无效entry的条件是entry中k==null
            e.value = null;
            tab[i] = null;
            size--;
         else   
            // 如果key不为null,计算索引
            int h = k.threadLocalHashCode & (len - 1);
            /**
             * 计算出来的索引h,与其现在所在位置的索引i不一致,置空当前的table[i什么是 ThreadLocal?

六十七:flask上下文之Local线程隔离对象

Session管理之ThreadLocal

线程隔离ThreadLocal

浅析ThreadLocal的底层实现线程隔离+内存泄漏

浅析ThreadLocal的底层实现线程隔离+内存泄漏