jdk源码解析三之ThreadLocal

柔情只为你懂 2023-02-20 12:08 44阅读 0赞

### 文章目录 ###

*  ThreadLocal
 *   *  set
     *   *  初始化ThreadLocalMap
         *  set赋值
         *  重新使用失效节点
         *  清空无效节点
         *  扩容
     *  get
     *  remove
     *  总结
     *  为何存在内存泄漏问题?

# ThreadLocal #

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L051YW5fRmVuZw_size_16_color_FFFFFF_t_70]

## set ##

public void set(T value) {
        
            //当前线程的.ThreadLocalMap绑定了当前ThreadLocal对象和value
            //获取当前线程
            Thread t = Thread.currentThread();
            //获取与当前线程绑定的map,这里主线程的map应该是jni初始化的
            ThreadLocalMap map = getMap(t);
            if (map != null)
                //将当前ThreadLocal与value绑定
                map.set(this, value);
            else
                //创建map
                createMap(t, value);
        }

### 初始化ThreadLocalMap ###

void createMap(Thread t, T firstValue) {
        
            t.threadLocals = new ThreadLocalMap(this, firstValue);
        }
      ThreadLocalMap(ThreadLocal<?> firstKey, Object firstValue) {
        
                //初始化一个16容量数组
                table = new Entry[INITIAL_CAPACITY];
                //根据hash算出索引
                int i = firstKey.threadLocalHashCode & (INITIAL_CAPACITY - 1);
                //Entry继承WeakReference
                //也就是说当一个对象仅仅被weak reference指向,
                // 而没有任何其他strong reference指向的时候, 如果GC运行, 那么这个对象就会被回收.
                table[i] = new Entry(firstKey, firstValue);
                //初始化容量值大小
                size = 1;
                //设置扩充容量值,容量*2/3
                setThreshold(INITIAL_CAPACITY);
            }
    
       static class Entry extends WeakReference<ThreadLocal<?>> {
        
                /** The value associated with this ThreadLocal. */
                Object value;
    
                Entry(ThreadLocal<?> k, Object v) {
        
                    super(k);
                    value = v;
                }
            }

### 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;
                int i = key.threadLocalHashCode & (len-1);
    
                /*
                遍历Entry
                    如果找到了则赋值返回
                    找到了回收的节点,则重新使用
                 */
                for (Entry e = tab[i];
                     e != null;
                     e = tab[i = nextIndex(i, len)]) {
        
                    ThreadLocal<?> k = e.get();
    
                    //当前ThreadLocal与k一致,则赋值,并返回
                    if (k == key) {
        
                        e.value = value;
                        return;
                    }
    
                    //说明被回收,则重新使用
                    if (k == null) {
        
                        replaceStaleEntry(key, value, i);
                        return;
                    }
                }
    
                //执行到这里,说明key不存在Entry中,被回收了
                //tab[i]此时为空,则赋值
                tab[i] = new Entry(key, value);
                int sz = ++size;
                //清空失效对象
                if (!cleanSomeSlots(i, sz) && sz >= threshold)
                    //扩容
                    rehash();
            }

### 重新使用失效节点 ###

private void replaceStaleEntry(ThreadLocal<?> key, Object value,
                                           int staleSlot) {
        
                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).
                //记录失效节点索引
                int slotToExpunge = staleSlot;
                //根据传入无效的索引向前遍历,找到一段连续的无效的索引
                for (int i = prevIndex(staleSlot, len);
                     (e = tab[i]) != null;
                     i = prevIndex(i, len))
                    if (e.get() == null)
                        slotToExpunge = i;
                //slotToExpunge此时只有2种状态
                //slotToExpunge=staleSlot,说明没有找到前驱节点有无效的
                //slotToExpunge!=staleSlot,说明靠近tab[i]=null的后驱节点有无效节点
    
                // 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.
                    //找到了key,则与无效索引互换位置,也就是当前无效索引对应的是存储的key-value
                    if (k == key) {
        
                        e.value = value;
    
                        tab[i] = tab[staleSlot];
                        tab[staleSlot] = e;
    
                        // Start expunge at preceding stale entry if it exists
                        //如果向前没有找到无效索引,则将slotToExpunge设置为i
                        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说明被回收,且如果没有找到上一个无效索引,则更改slotToExpunge为i
                    if (k == null && slotToExpunge == staleSlot)
                        slotToExpunge = i;
                }
    
                // If key not found, put new entry in stale slot
                //没有找到key,在无效的staleSlot索引上,新建一个Entry对象
                tab[staleSlot].value = null;
                tab[staleSlot] = new Entry(key, value);
    
                // If there are any other stale entries in run, expunge them
                //2者不相等,则清除其他无效的entry
                if (slotToExpunge != staleSlot)
                    cleanSomeSlots(expungeStaleEntry(slotToExpunge), len);
            }

### 清空无效节点 ###

```java
     private int expungeStaleEntry(int staleSlot) {
        
                Entry[] tab = table;
                int len = tab.length;
    
                // expunge entry at staleSlot
                //清空value和索引对象
                tab[staleSlot].value = null;
                tab[staleSlot] = null;
                //数组里面的值大小-1
                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();
                    //清除过期的key
                    if (k == null) {
        
                        e.value = null;
                        tab[i] = null;
                        size--;
                    } else {
        
                        int h = k.threadLocalHashCode & (len - 1);
                        //重新计算k的索引,判断是否和当前i冲突
                        if (h != i) {
        
                            tab[i] = null;
    
                            // Unlike Knuth 6.4 Algorithm R, we must scan until
                            // null because multiple entries could have been stale.
                            //获取下一个tab为null的位置,赋值e
                            //这里实际上是将e移动到靠近h的后驱节点上,这样get/set就方便查找
                            while (tab[h] != null)
                                h = nextIndex(h, len);
                            //存储e
                            tab[h] = e;
                        }
                    }
                }
                //返回下一个为空的entry的索引
                return i;
            }
            
     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];
                    if (e != null && e.get() == null) {
        
                        //重置n为tab长度
                        n = len;
                        removed = true;
                        i = expungeStaleEntry(i);
                    }
                } while ( (n >>>= 1) != 0);//无符号右移,控制扫描次数log2(N)
                return removed;
            }

### 扩容 ###

private void rehash() {
        
                //清除所有无效节点
                expungeStaleEntries();
    
                // Use lower threshold for doubling to avoid hysteresis
                //2/3*len-2/3*len/4=1/2*len
                //size >= 1/2*len
                if (size >= threshold - threshold / 4)
                    resize();
            }
    
     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);
                }
            }

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

## get ##

public T get() {
        
            //获取当前线程
            Thread t = Thread.currentThread();
            //获取当前线程map
            ThreadLocalMap map = getMap(t);
            if (map != null) {
        
                //根据当前ThreadLocal获取value
                ThreadLocalMap.Entry e = map.getEntry(this);
                if (e != null) {
        
                    @SuppressWarnings("unchecked")
                            //获取值,并返回
                    T result = (T)e.value;
                    return result;
                }
            }
            //map为空则初始化value
            return setInitialValue();
        }

private T setInitialValue() {
        
            //初始化值为null
            T value = initialValue();
            //获取当前线程
            Thread t = Thread.currentThread();
            //获取map
            ThreadLocalMap map = getMap(t);
            if (map != null)
                //否则设置值
                map.set(this, value);
            else
                //map为空则创建map,并赋值
                createMap(t, value);
            return value;
        }

private Entry getEntry(ThreadLocal<?> key) {
        
                //算出索引
                int i = key.threadLocalHashCode & (table.length - 1);
                Entry e = table[i];
                //根据key获取Entry
                if (e != null && e.get() == key)
                    return e;
                else
                    //没有获取,则说明要么失效了,要么存在在后续节点
                    return getEntryAfterMiss(key, i, e);
            }

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

## remove ##

public void remove() {
        
             //获取当前线程map
             ThreadLocalMap m = getMap(Thread.currentThread());
             if (m != null)
                 //删除当前值
                 m.remove(this);
         }
       private void remove(ThreadLocal<?> key) {
        
                Entry[] tab = table;
                int len = tab.length;
                int i = key.threadLocalHashCode & (len-1);
                //遍历Entry
                for (Entry e = tab[i];
                     e != null;
                     e = tab[i = nextIndex(i, len)]) {
        
                    //删除当前ThreadLocal对应的value
                    if (e.get() == key) {
        
                        //清空referent
                        e.clear();
                        expungeStaleEntry(i);
                        return;
                    }
                }
            }

## 总结 ##

扩容时机,当前集合个数超过len\*3/4  
Entry继承WeakReference,也就是说当一个对象仅仅被weak reference指向,而没有任何其他strong reference指向的时候, 如果GC运行, 那么这个对象就会被回收.  
底层数组存储ThreadLocal和val,采用hashcode获取索引

为什么采用弱引用?因为当采用强引用的时候,如果ThreadLocal为空的时候,GC依旧不会清空存储的值,会造成内存泄漏  
在threadLocal的生命周期里（set,getEntry,remove）里，都会针对key为null的脏entry进行处理。

## 为何存在内存泄漏问题? ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L051YW5fRmVuZw_size_16_color_FFFFFF_t_70 1]  
内存泄漏的根本原因是强引用和弱引用的生命周期不一致造成的.ThreadLocal作为一个虚引用的key,当被GC回收时,但是value,被当前线程的ThreadLocalMap持有,只要当前线程不销毁,或者没调用get/set方法(调用get/set时,会清除null为key的数据),就一直持有  
最严重的的情况是使用了线程池或者静态声明ThreadLocal,大大延长了二者的生命周期.没能及时回收null数据,所以内存泄漏咯  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L051YW5fRmVuZw_size_16_color_FFFFFF_t_70 2]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L051YW5fRmVuZw_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20200702100653115.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L051YW5fRmVuZw==,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L051YW5fRmVuZw_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20200702095442809.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L051YW5fRmVuZw==,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L051YW5fRmVuZw_size_16_color_FFFFFF_t_70 2]: https://img-blog.csdnimg.cn/20200702100027516.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L051YW5fRmVuZw==,size_16,color_FFFFFF,t_70