ThreadLocal企业中真实应用

公众号 IT老哥

修改于 2020-09-21 15:07:33

1.1K0

修改于 2020-09-21 15:07:33

文章被收录于专栏：用户7621540的专栏

本文源自公-众-号 IT老哥的分享

ThreadLocal解决多线程安全案例

项目中封装的日期工具类用在多线程环境下居然出了问题，来看看怎么回事吧

public class ThreadLocalTest {

    public static void main(String[] args) {

        // 创建线程池
        ThreadFactory threadFactory = new ThreadFactoryBuilder().setNameFormat("thread-%d").build();
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(20, 20, 0L,
                TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>(1024), threadFactory);

        for (int i = 0; i < 20; i++) {
            threadPoolExecutor.execute(
                ()-> System.out.println(DateUtilSafe.parse("2019-06-01 16:34:30"))
            );
        }
        threadPoolExecutor.shutdown();
    }
}

日期工具类（线程不安全）


public class DateUtilNotSafe {

    private static final SimpleDateFormat sdf =
            new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

    public static Date parse(String dateStr) {
        Date date = null;
        try {
            date = sdf.parse(dateStr);
        } catch (ParseException e) {
            e.printStackTrace();
        }
        return date;
    }
}

多线程下报错截图：

ThreadLocal解决方案：

public class DateUtilSafe {

    private static final ThreadLocal<DateFormat> THREAD_LOCAL = ThreadLocal.withInitial(
        () -> new SimpleDateFormat("yyyy-MM-dd HH:mm:ss")
    );

    public static Date parse(String dateStr) {
        Date date = null;
        try {
            date = THREAD_LOCAL.get().parse(dateStr);
        } catch (ParseException e) {
            e.printStackTrace();
        }
        return date;
    }
}

分析：

SimpleDateFormat(下面简称sdf)类内部有一个Calendar对象引用，它用来储存和这个sdf相关的日期信息，例如sdf.parse(dateStr), sdf.format(date) 诸如此类的方法参数传入的日期相关String、Date等等，都是交友Calendar引用来储存的，这样就会导致一个问题，如果你的sdf是个static的, 那么多个thread 之间就会共享这个sdf，同时也是共享这个Calendar引用，并且, 观察 sdf.parse() 方法，parse方法里没有保证原子性，所以存在线程安全问题：


Date parse() {

  calendar.clear(); // 清理calendar

  ... // 执行一些操作, 设置 calendar 的日期什么的

  calendar.getTime(); // 获取calendar的时间

}

既然是因为多个线程共享SimpleDateFormat造成的，那么我们就让它不共享，每个线程存一份自己的SimpleDateFormat对象。自己玩自己的对象，就不会出现线程问题了。ThreadLocal作用就是让线程自己独立保存一份自己的变量副本。每个线程独立的使用自己的变量副本，不会影响其他线程内的变量副本。

ThreadLocal简介

很多小伙伴认为ThreadLocal是多线程同步机制的一种，其实不然，他是为多线程环境下为变量线程安全提供的一种解决思路，他是解决多线程下成员变量的安全问题，不是解决多线程下共享变量的安全问题。

线程同步机制是多个线程共享一个变量，而ThreadLocal是每个线程创建一个自己的单独变量副本，所以每个线程都可以独立的改变自己的变量副本。并且不会影响其他线程的变量副本。

ThreadLocalMap

ThreadLocal内部有一个非常重要的内部类：ThreadLocalMap，该类才是真正实现线程隔离机制的关键，ThreadLocalMap内部结构类似于map，由键值对key和value组成一个Entry，key为ThreadLocal本身，value是对应的线程变量副本

注意：

1、ThreadLocal本身不存储值，他只是提供一个查找到值的key给你。

2、ThreadLocal包含在Thread中，不是Thread包含在ThreadLocal中。

ThreadLocalMap 和HashMap的功能类似，但是实现上却有很大的不同：

HashMap 的数据结构是数组+链表
ThreadLocalMap的数据结构仅仅是数组
HashMap 是通过链地址法解决hash 冲突的问题
ThreadLocalMap 是通过开放地址法来解决hash 冲突的问题
HashMap 里面的Entry 内部类的引用都是强引用
ThreadLocalMap里面的Entry 内部类中的key 是弱引用，value 是强引用

链地址法

这种方法的基本思想是将所有哈希地址为i的元素构成一个称为同义词链的单链表，并将单链表的头指针存在哈希表的第i个单元中，因而查找、插入和删除主要在同义词链中进行。

开放地址法

这种方法的基本思想是一旦发生了冲突，就去寻找下一个空的散列地址(这非常重要，源码都是根据这个特性，必须理解这里才能往下走)，只要散列表足够大，空的散列地址总能找到，并将记录存入。

链地址法和开放地址法的优缺点

开放地址法：

容易产生堆积问题，不适于大规模的数据存储。
散列函数的设计对冲突会有很大的影响，插入时可能会出现多次冲突的现象。
删除的元素是多个冲突元素中的一个，需要对后面的元素作处理，实现较复杂。

链地址法：

处理冲突简单，且无堆积现象，平均查找长度短。
链表中的结点是动态申请的，适合构造表不能确定长度的情况。
删除结点的操作易于实现。只要简单地删去链表上相应的结点即可。
指针需要额外的空间，故当结点规模较小时，开放定址法较为节省空间。

ThreadLocalMap 采用开放地址法原因

ThreadLocal 中看到一个属性 HASH_INCREMENT = 0x61c88647 ，0x61c88647 是一个神奇的数字，让哈希码能均匀的分布在2的N次方的数组里, 即 Entry[] table，关于这个神奇的数字google 有很多解析，这里就不重复说了
ThreadLocal 往往存放的数据量不会特别大（而且key 是弱引用又会被垃圾回收，及时让数据量更小），这个时候开放地址法简单的结构会显得更省空间，同时数组的查询效率也是非常高，加上第一点的保障，冲突概率也低

Thread、ThreadLocal、ThreadLocalMap之间的关系

从上面的结构图，我们已经窥见ThreadLocal的核心机制：

每个Thread线程内部都有一个Map。Map里面存储线程本地对象（key）和线程的变量副本（value）Thread内部的Map是由ThreadLocal维护的，由ThreadLocal负责向map获取和设置线程的变量值。所以对于不同的线程，每次获取副本值时，别的线程并不能获取到当前线程的副本值，形成了副本的隔离，彼此之间互不干扰。

源码解读

先了解一下ThreadLocal类提供的几个方法：


public T get() { }
public void set(T value) { }
public void remove() { }
protected T initialValue() { }

get()方法是用来获取ThreadLocal在当前线程中保存的变量副本。 set()用来设置当前线程中变量的副本。 remove()用来移除当前线程中变量的副本。 initialValue()是一个protected方法，一般是用来在使用时进行重写的

get方法

// 通过key拿value值
     public T get() {

        // 获取当前线程
        Thread t = Thread.currentThread();

        // 获取当前线程的ThreadLocalMap
        ThreadLocalMap map = getMap(t);

        if (map != null) {

            // this是当前的ThreadLocalMap（key），getEntry通过key拿到value：e
            ThreadLocalMap.Entry e = map.getEntry(this);

            if (e != null) {
                @SuppressWarnings("unchecked")
                T result = (T)e.value;
                // 返回获取到的value
                return result;
            }
        }
        return setInitialValue();
    }

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

set方法

public void set(T value) {
        // 获取当前线程
        Thread t = Thread.currentThread();
        
        // 获取当前线程的ThreadLocalMap
        ThreadLocalMap map = getMap(t);
        if (map != null)
            // 重新将ThreadLocal和新的value副本放入到map中。
            map.set(this, value);
        else
            // 创建
            createMap(t, value);
    }

    // 创建ThreadLocalMap，将ThreadLocalMap和Thread绑定关系
    void createMap(Thread t, T firstValue) {
        t.threadLocals = new ThreadLocalMap(this, firstValue);
    }

remove方法

public void remove() {
         ThreadLocalMap m = getMap(Thread.currentThread());
         if (m != null)
             // 调用的ThreadLocalMap里的remove方法，之后统一分析
             m.remove(this);
     }

这里罗列了 ThreadLocal 的几个public方法，其实所有工作最终都落到了 ThreadLocalMap 的头上，ThreadLocal 仅仅是从当前线程取到 ThreadLocalMap 而已，具体执行，请看下面对 ThreadLocalMap 的分析。

ThreadLocalMap数据结构源码：

public class ThreadLocal<T> {
    
    // 数据结构采用 数组 + 开放地址法
    static class ThreadLocalMap {
        
        private Entry[] table;

        // Entry 继承弱引用WeakReference，
        // 这块会存在内存泄露问题，之后详细说明
        static class Entry extends WeakReference<ThreadLocal<?>> {
            /** ThreadLocal key对应的值value */
            Object value;

            // 内部类Entry是类似于map结构的key、value结构
            // key就是ThreadLocal，value是变量副本值
            Entry(ThreadLocal<?> k, Object v) {
                super(k);
                value = v;
            }
        }
    }
}

set方法


// ThreadLocalMap设置key、value
        private void set(ThreadLocal<?> key, Object value) {

            Entry[] tab = table;
            int len = tab.length;
            
            // 计算key的索引值
            int i = key.threadLocalHashCode & (len-1);

            for (Entry e = tab[i];
                 e != null;
                 e = tab[i = nextIndex(i, len)]) {

                // 拿到此次循环的key
                ThreadLocal<?> k = e.get();
                
                // 根据key计算的索引值
                // 进行线性搜索后找到的第一个Key为空的Entry
                if (k == key) {
                    e.value = value;
                    return;
                }
                
                // 如果k == null && e != null，说明k被回收了，
                // 因为Entry 继承 WeakReference弱引用，GC的时候会把key回收调
                if (k == null) {
                    // k被回收后，这个位置已经没人用了，就可以将新的key和value放到这个位置
                    replaceStaleEntry(key, value, i);
                    return;
                }
            }
            // 如果方法没有在上面的方法中return
            // 说明此时位置i的Entry是空的，可以设置key和value
            tab[i] = new Entry(key, value);
            int sz = ++size;
            
            // cleanSomeSlots方法返回false表示数组中已经不存在key为空需要清除的Entry了
            // 此时数组装满了，而 sz 表示此时数组中元素的数量大于临界值了时
            // 需要调用rehash进行扩容
            if (!cleanSomeSlots(i, sz) && sz >= threshold)
                // 扩容
                rehash();
        }

replaceStaleEntry替换方法

private void replaceStaleEntry(ThreadLocal<?> key, Object value,
                               int staleSlot) {
    Entry[] tab = table;
    int len = tab.length;
    Entry e;
 
    // 清除元素的开始位置（记录索引位置最前面的）
    int slotToExpunge = staleSlot;

    // 向前遍历，直到遇到Entry为空
    for (int i = prevIndex(staleSlot, len);
         (e = tab[i]) != null;
         i = prevIndex(i, len))
        if (e.get() == null)
            // 记录最后一个key为null的索引位置
            slotToExpunge = i;
 
    // Find either the key or trailing null slot of run, whichever
    // occurs first
    // 向后遍历，直到遇到Entry为空
    for (int i = nextIndex(staleSlot, len);
         (e = tab[i]) != null;
         i = nextIndex(i, len)) {
        ThreadLocal<?> k = e.get();
 
        // 该Entry的key和传入的key相等, 则将传入的value替换掉该Entry的value
        if (k == key) {
            e.value = value;
 
            // 将i位置和staleSlot位置的元素对换（staleSlot位置较前，是要清除的元素）
            tab[i] = tab[staleSlot];
            tab[staleSlot] = e;
 
            // 如果相等, 则代表上面的向前寻找key为null的遍历没有找到，
            // 即staleSlot位置前面的元素没有需要清除的，此时将slotToExpunge设置为i,
            // 因为原staleSlot的元素已经被放到i位置了，这时位置i前面的元素都不需要清除
            if (slotToExpunge == staleSlot)
                slotToExpunge = i;

            // 从slotToExpunge位置开始清除key为空的Entry
            cleanSomeSlots(expungeStaleEntry(slotToExpunge), len);
            return;
        }
 
        // 如果第一次遍历到key为null的元素，并且上面的向前寻找key为null的遍历没有找到，
        // 则将slotToExpunge设置为当前的位置
        if (k == null && slotToExpunge == staleSlot)
            slotToExpunge = i;
    }
 
    // 如果key没有找到，则新建一个Entry，放在staleSlot位置
    tab[staleSlot].value = null;
    tab[staleSlot] = new Entry(key, value);
 
    // 如果slotToExpunge!=staleSlot，代表除了staleSlot位置还有其他位置的元素需要清除
    // 需要清除的定义：key为null的Entry，调用cleanSomeSlots方法清除key为null的Entry
    if (slotToExpunge != staleSlot)
        cleanSomeSlots(expungeStaleEntry(slotToExpunge), len);
}

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

        // 遍历到key为null的元素
        if (e != null && e.get() == null) {

            // 重置n的值
            n = len;

            // 标志有移除元素
            removed = true;

            // 移除i位置及之后的key为null的元素
            i = expungeStaleEntry(i);
        }
    } while ( (n >>>= 1) != 0);
    return removed;
}

get()方法

public T get() {
    Thread t = Thread.currentThread();
    ThreadLocalMap map = getMap(t);
    if (map != null) {

        // 调用getEntry方法, 通过this（调用get()方法的ThreadLocal）获取对应的Entry
        ThreadLocalMap.Entry e = map.getEntry(this);

        // Entry不为空则代表找到目标Entry, 返回该Entry的value值
        if (e != null) {
            @SuppressWarnings("unchecked")
            T result = (T)e.value;
            return result;
        }
    }

    // 该线程的ThreadLocalMap为空，或者没有找到目标Entry，则调用setInitialValue方法
    return setInitialValue();
}

setInitialValue方法

private T setInitialValue() {
    
    // 默认null，需要用户自己重写该方法，
    T value = initialValue();

    // 当前线程
    Thread t = Thread.currentThread();

    // 拿到当前线程的threadLocals
    ThreadLocalMap map = getMap(t);

    // threadLocals不为空则将当前的ThreadLocal作为key
    // null作为value，插入到ThreadLocalMap
    if (map != null)
        map.set(this, value);

    // threadLocals为空则调用创建一个ThreadLocalMap
    // 并新建一个Entry放入该ThreadLocalMap
    // 调用set方法的ThreadLocal和value作为该Entry的key和value
    else
        createMap(t, value);
    return value;
}

getEntry方法

private Entry getEntry(ThreadLocal<?> key) {
  
    //根据hash code计算出索引位置
    int i = key.threadLocalHashCode & (table.length - 1);
    Entry e = table[i];

    // 如果该Entry的key和传入的key相等, 则为目标Entry, 直接返回
    if (e != null && e.get() == key)
        return e;

    // 否则，e不是目标Entry, 则从e之后继续寻找目标Entry
    else
        return getEntryAfterMiss(key, i, e);
}

getEntryAfterMiss方法


private Entry getEntryAfterMiss(ThreadLocal<?> key, int i, Entry e) {
    Entry[] tab = table;
    int len = tab.length;
 
    while (e != null) {
        ThreadLocal<?> k = e.get();

        // 找到目标Entry，直接返回
        if (k == key)
            return e;

        // 调用expungeStaleEntry清除key为null的元素
        if (k == null)
            expungeStaleEntry(i);
        else
            // 下一个索引位置
            i = nextIndex(i, len);
        // 下一个遍历的Entry
        e = tab[i];
    }
    // 找不到, 返回空
    return null;
}

remove()方法


public void remove() {
  // 获取当前线程的ThreadLocalMap
  ThreadLocalMap m = getMap(Thread.currentThread());

  if (m != null)
        // 调用此方法的ThreadLocal作为入参，调用remove方法
    m.remove(this);
 }

private void remove(ThreadLocal<?> key) {
    Entry[] tab = table;
    int len = tab.length;

    // 根据hashCode计算出当前ThreadLocal的索引位置
    int i = key.threadLocalHashCode & (len-1);

    // 从位置i开始遍历，直到Entry为null
    for (Entry e = tab[i];
         e != null;
         e = tab[i = nextIndex(i, len)]) {

        // 如果找到key相同的
        if (e.get() == key) {
            
            // 则调用clear方法, 该方法会把key的引用清空
            e.clear();

            //调用expungeStaleEntry方法清除key为null的Entry
            expungeStaleEntry(i);
            return;
        }
    }
}

expungeStaleEntry方法

// 从staleSlot开始, 清除key为空的Entry,
// 并将不为空的元素放到合适的位置，最后返回Entry为空的位置
private int expungeStaleEntry(int staleSlot) {
    Entry[] tab = table;
    int len = tab.length;
 
    // 将tab上staleSlot位置的对象清空
    tab[staleSlot].value = null;
    tab[staleSlot] = null;
    size--;
 
    // Rehash until we encounter null
    Entry e;
    int i;
    // 遍历下一个元素, 即(i+1)%len位置的元素
    for (i = nextIndex(staleSlot, len);
          // 遍历到Entry为空时, 跳出循环并返回索引位置
         (e = tab[i]) != null;
         i = nextIndex(i, len)) {
        ThreadLocal<?> k = e.get();
        
        // 当前遍历Entry的key为空, 则将该位置的对象清空
        if (k == null) {
            e.value = null;
            tab[i] = null;
            size--;
        } else {
            // 重新计算该Entry的索引位置
            int h = k.threadLocalHashCode & (len - 1);
            
            // 如果索引位置不为当前索引位置i
            if (h != i) {
                // 则将i位置对象清空, 替当前Entry寻找正确的位置
                tab[i] = null;
 
                // 如果h位置不为null，则向后寻找当前Entry的位置
                while (tab[h] != null)
                    h = nextIndex(h, len);
                tab[h] = e;
            }
        }

rehash方法


private void rehash() {

    // 调用expungeStaleEntries方法清理key为空的Entry
    expungeStaleEntries();
 
    // 如果清理后size超过阈值的3/4, 则进行扩容
    if (size >= threshold - threshold / 4)
        resize();
}

/**
 * Double the capacity of the table.
 */
private void resize() {
    Entry[] oldTab = table;
    int oldLen = oldTab.length;

    // 新表长度为老表2倍
    int newLen = oldLen * 2;

    // 创建新表
    Entry[] newTab = new Entry[newLen];
    int count = 0;
 
    for (int j = 0; j < oldLen; ++j) {
        
        // 拿到对应位置的Entry
        Entry e = oldTab[j];
        if (e != null) {
            ThreadLocal<?> k = e.get();

            // 如果key为null，将value清空
            if (k == null) {
                e.value = null; // Help the GC
            } else {

              // 通过hash code计算新表的索引位置
                int h = k.threadLocalHashCode & (newLen - 1);

                // 如果新表的该位置已经有元素，则调用nextIndex方法直到寻找到空位置
                while (newTab[h] != null)
                    h = nextIndex(h, newLen);
                
                // 将元素放在对应位置
                newTab[h] = e;
                count++;
            }
        }
    }
 
    // 设置新表扩容的阈值
    setThreshold(newLen);
    // 更新size
    size = count;
    // table指向新表
    table = newTab;
}

内存泄露问题：

static class Entry extends WeakReference<ThreadLocal<?>> {
    /** The value associated with this ThreadLocal. */
    Object value;
 
    Entry(ThreadLocal<?> k, Object v) {
        super(k);
        value = v;
    }
}

从上面源码可以看出，ThreadLocalMap使用ThreadLocal的弱引用作为Entry的key，如果一个ThreadLocal没有外部强引用来引用它，下一次系统GC时，这个ThreadLocal必然会被回收，这样一来，ThreadLocalMap中就会出现key为null的Entry，就没有办法访问这些key为null的Entry的value。

我们上面介绍的get、set、remove等方法中，都会对key为null的Entry进行清除（expungeStaleEntry方法，将Entry的value清空，等下一次垃圾回收时，这些Entry将会被彻底回收）。

但是如果当前线程一直在运行，并且一直不执行get、set、remove方法，这些key为null的Entry的value就会一直存在一条强引用练：Thread Ref -> Thread -> ThreadLocalMap -> Entry -> value，导致这些key为null的Entry的value永远无法回收，造成内存泄漏。

如何避免内存泄漏？为了避免这种情况，我们可以在使用完ThreadLocal后，手动调用remove方法，以避免出现内存泄漏。

云服务器，云硬盘，数据库（包括MySQL、Redis、MongoDB、SQL Server），CDN流量包，短信流量包，cos资源包，消息队列ckafka，点播资源包，实时音视频套餐，网站管家（WAF），大禹BGP高防（包含高防包及高防IP），云解析，SSL证书，手游安全MTP，移动应用安全、云直播等等。

本文参与腾讯云自媒体同步曝光计划，分享自微信公众号。

原始发表：2020-06-15，如有侵权请联系 cloudcommunity@tencent.com 删除

java

本文分享自 IT老哥微信公众号，前往查看

如有侵权，请联系 cloudcommunity@tencent.com 删除。

本文参与腾讯云自媒体同步曝光计划，欢迎热爱写作的你一起参与！

登录后参与评论

0 条评论

热度