ThreadLocal數(shù)據(jù)存儲結(jié)構(gòu)原理解析

更新時(shí)間：2022年10月09日 15:57:22 作者：沉迷學(xué)習(xí)的小伙

這篇文章主要為大家介紹了ThreadLocal數(shù)據(jù)存儲結(jié)構(gòu)原理解析，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進(jìn)步，早日升職加薪

一：簡述

我們很多時(shí)候?yàn)榱藢?shí)現(xiàn)數(shù)據(jù)在線程級別下的隔離，會使用到ThreadLocal，那么TheadLocal是如何實(shí)現(xiàn)數(shù)據(jù)隔離的呢？今天就和大家一起分析一下ThreadLocal的實(shí)現(xiàn)原理。

二：TheadLocal的原理分析

1.ThreadLocal的存儲結(jié)構(gòu)

每個(gè)Thread對象中都有一個(gè)threadLocals成員變量，threadLocals是一個(gè)類型為ThreadLocalMap的map，而ThreadLocal正是基于這個(gè)map來實(shí)現(xiàn)線程級別的數(shù)據(jù)隔離的。

我們先看ThreadLocalMap的成員變量

        //默認(rèn)的初始化容量大小
        private static final int INITIAL_CAPACITY = 16;
        //Entry數(shù)組 真正存儲的數(shù)據(jù)結(jié)構(gòu)
        private Entry[] table;
        //記錄當(dāng)前元素的數(shù)量
        private int size = 0;
        //擴(kuò)容的閾值
        private int threshold;

Entry數(shù)組是真正存儲數(shù)據(jù)的地方，可以看出Entry是一個(gè)key-value的存儲結(jié)構(gòu)，以當(dāng)前ThreadLocal對象的引用作為key，存儲的值為value。Entry繼承了WeakReference，并且在構(gòu)造函數(shù)的時(shí)候，調(diào)用super(k)（也就是WeakReference的構(gòu)造函數(shù)）來對key進(jìn)行初始化，所以Entry的key是一個(gè)弱引用。

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

根據(jù)上面的分析，我們可以知道ThreadLocal的存儲結(jié)構(gòu)大概是這樣的：

2.源碼分析

接下來我們從ThreadLocal的set()，get(),remove()方法為入口對ThreadLocal的源碼進(jìn)行分析。

set()方法

首先判斷當(dāng)前線程的threadLocals是否初始化，如果沒有初始化，那么調(diào)用createMap()方法進(jìn)行初始化并設(shè)置值，否則調(diào)用ThreadLocalMap的set()方法設(shè)置值。

流程圖：

三：源碼分析

public void set(T value) {
         //利用當(dāng)前線程獲取它的threadLocals（threadLocals是一個(gè)ThreadLocalMap）
        Thread t = Thread.currentThread();
        ThreadLocalMap map = getMap(t);
        //如果已經(jīng)初始化 那么就調(diào)用ThreadLocalMap的set()方法
        if (map != null)
            map.set(this, value);
        else
            // 沒有初始化 先進(jìn)行初始化
            createMap(t, value);
    }

    ThreadLocalMap getMap(Thread t) {
        //返回當(dāng)前線程的threadLocals
        return t.threadLocals;
    }

createMap()

createMap()會調(diào)用ThreadLocalMap的構(gòu)造函數(shù)對當(dāng)前線程的threadLocals初始化，并且初始化Entry數(shù)組，然后利用hash算法計(jì)算出數(shù)組下標(biāo)，將需要set的值存儲在Entry數(shù)組。

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

        ThreadLocalMap(ThreadLocal<?> firstKey, Object firstValue) {
            // 初始化Entry數(shù)組
            table = new Entry[INITIAL_CAPACITY];
            //計(jì)算數(shù)組下標(biāo)
            int i = firstKey.threadLocalHashCode & (INITIAL_CAPACITY - 1);
            table[i] = new Entry(firstKey, firstValue);
            size = 1;
            //設(shè)置默認(rèn)的擴(kuò)容閾值 和默認(rèn)容量一樣
            setThreshold(INITIAL_CAPACITY);
        }

如果threadLocals已經(jīng)初始化，直接調(diào)用ThreadLocalMap的set()，接下來看ThreadLocalMap的set()方法。

首先利用hash算法計(jì)算數(shù)組下標(biāo)，如果計(jì)算出的位置沒有值，直接將值設(shè)置進(jìn)去，如果存在值（出現(xiàn)hash沖突），分為三種情況：

1.如果key相同，那么直接覆蓋值

2.如果計(jì)算出的位置的Entry的key為null，那么說明是無效的數(shù)據(jù)（key為null，entry不為null）,為了避免內(nèi)存泄漏需要清除這種數(shù)據(jù)。所以調(diào)用replaceStaleEntry()方法將無效數(shù)據(jù)清除并且將需要設(shè)置的值設(shè)置到Entry數(shù)組中。

3.如果key不相同，而且計(jì)算出的位置的Entry的key不為null，那么進(jìn)入到下一次for循環(huán)將計(jì)算出的下標(biāo)+1，（如果到達(dá)下標(biāo)最大值，則設(shè)置為0），利用新的位置重新進(jìn)行判斷，直到獲取到一個(gè)合法的位置（線性尋址法解決hash沖突的問題）。

注：這里大家可以評論區(qū)討論下為什么不和HashMap那樣利用鏈表法解決hash沖突。我個(gè)人的看法是因?yàn)門hreadLocal的數(shù)據(jù)量不會向HashMap那么多，所以不需要利用鏈表和紅黑樹來解決hash沖突，鏈表法解決代碼相對比較復(fù)雜而且擴(kuò)容遷移數(shù)據(jù)的數(shù)據(jù)會比較麻煩。

源碼：

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;
            // 計(jì)算數(shù)組下標(biāo)
            int i = key.threadLocalHashCode & (len-1);
            //如果出現(xiàn)hash沖突會進(jìn)入for循環(huán)
            for (Entry e = tab[i];
                 e != null;
                 e = tab[i = nextIndex(i, len)]) {
                ThreadLocal<?> k = e.get();
                //如果key相同 那么直接將值覆蓋
                if (k == key) {
                    e.value = value;
                    return;
                }
                //如果key為null 那么說明是無效的數(shù)據(jù) 需要進(jìn)行清除
                if (k == null) {
                    //調(diào)用replaceStaleEntry()方法進(jìn)行清除數(shù)據(jù) 并設(shè)置值
                    replaceStaleEntry(key, value, i);
                    return;
                }
            }
            //如果沒有hash沖突 直接賦值到對應(yīng)下標(biāo)的位置
            tab[i] = new Entry(key, value);
            // 將當(dāng)前元素個(gè)數(shù)+1
            int sz = ++size;
            //如果沒有需要清除的元素，并且當(dāng)前元素個(gè)數(shù)已經(jīng)達(dá)到擴(kuò)容的閾值，那么進(jìn)行擴(kuò)容
            if (!cleanSomeSlots(i, sz) && sz >= threshold)
                rehash();
        }

接下來看replaceStaleEntry()，看ThreadLocal是如何清除無效的數(shù)據(jù)的。

當(dāng)前節(jié)點(diǎn)是無效的數(shù)據(jù)，那么周圍也可能存在無效的數(shù)據(jù)，所以ThreadLocal在清除無效的數(shù)據(jù)時(shí)，會順便清除周圍的連續(xù)的無效數(shù)據(jù)，先利用for循環(huán)從當(dāng)前節(jié)點(diǎn)向前遍歷，調(diào)整slotToExpunge的值（slotToExpunge 用于保存開始清除無效數(shù)據(jù)的下標(biāo)位置），然后向后遍歷，如果有entry的key和需要存放的數(shù)據(jù)的key相同，那么直接覆蓋值，并且交換當(dāng)前節(jié)點(diǎn)和新設(shè)置的entry的值。

流程圖：

private void replaceStaleEntry(ThreadLocal<?> key, Object value,
                                       int staleSlot) {
            Entry[] tab = table;
            int len = tab.length;
            Entry e;
	    // slotToExpunge 用于保存開始清除無效數(shù)據(jù)的下標(biāo)位置
            int slotToExpunge = staleSlot;
	    //從當(dāng)前位置向前遍歷，直到找到一個(gè)有效數(shù)據(jù)的下標(biāo)
            for (int i = prevIndex(staleSlot, len);
                 (e = tab[i]) != null;
                 i = prevIndex(i, len))
		//e.get()返回Entry的key，威null代表是無效數(shù)據(jù) 
                //（因?yàn)橹挥衑ntry不為null才會進(jìn)入for循環(huán)） 所以key為null，就是無效數(shù)據(jù)
                //for循環(huán)將清除無效數(shù)據(jù)的下標(biāo)往前挪
                if (e.get() == null)
                    slotToExpunge = i;
            //從當(dāng)前位置往后遍歷
            for (int i = nextIndex(staleSlot, len);
                 (e = tab[i]) != null;
                 i = nextIndex(i, len)) {
                ThreadLocal<?> k = e.get();
                //如果遍歷的是否發(fā)現(xiàn)有和當(dāng)前Entry相同的key的entry，那么交換兩者的位置
                if (k == key) {
                    e.value = value;
                    tab[i] = tab[staleSlot];
                    tab[staleSlot] = e;
                    //如果slotToExpunge和staleSlot相等 
                    //證明當(dāng)前節(jié)點(diǎn)的前面沒有和當(dāng)前節(jié)點(diǎn)連續(xù)的無效數(shù)據(jù) 
                    //所以從交換完的位置開始清除無效數(shù)據(jù) 調(diào)用cleanSomeSlots()方法和expungeStaleEntry()方法清除無效數(shù)據(jù) 清除完返回。
                    if (slotToExpunge == staleSlot)
                        slotToExpunge = i;
                    cleanSomeSlots(expungeStaleEntry(slotToExpunge), len);
                    return;
                }
                //如果key為null 而且當(dāng)前entry之前沒有與當(dāng)前節(jié)點(diǎn)連續(xù)的無效數(shù)據(jù)
                //刷新開始清除無效數(shù)據(jù)的下標(biāo)
                if (k == null && slotToExpunge == staleSlot)
                    slotToExpunge = i;
            }
            // If key not found, put new entry in stale slot
            //如果沒有找到連續(xù)的無效數(shù)據(jù) 把當(dāng)前的節(jié)點(diǎn)的value重置為null 并且將新的值賦值到當(dāng)前位置
            //因?yàn)楫?dāng)前的entry是無效的數(shù)據(jù) 
            tab[staleSlot].value = null;
            tab[staleSlot] = new Entry(key, value);
            // If there are any other stale entries in run, expunge them
            //如果slotToExpunge 和 staleSlot不相等 說明有連續(xù)的無效數(shù)據(jù)需要順便清除
            if (slotToExpunge != staleSlot)
                cleanSomeSlots(expungeStaleEntry(slotToExpunge), len);
        }

注：大家可以在評論區(qū)討論一下，這里為什么要交換一下數(shù)據(jù)，我個(gè)人認(rèn)為，第一是為了保證數(shù)據(jù)存儲的位置盡可能的在hash計(jì)算出位置，有利于后續(xù)的get()方法，第二：交換位置之后有利于讓無效的數(shù)據(jù)連續(xù)起來，提高清除無效數(shù)據(jù)的效率。

真正清除無效數(shù)據(jù)的方法是expungeStaleEntry()方法和cleanSomeSlots()方法

我們先看expungeStaleEntry()方法

expungeStaleEntry()

expungeStaleEntry()方法從當(dāng)前節(jié)點(diǎn)開始向后遍歷（直到遇到enrty為null的節(jié)點(diǎn)），將無效數(shù)據(jù)清除，并且重新計(jì)算有效的entry的數(shù)組下標(biāo)，如果計(jì)算出的下標(biāo)和entry的下標(biāo)不相同（這是因?yàn)椴捎昧司€性尋址法，所以hash計(jì)算出下標(biāo)可能和實(shí)際的下標(biāo)不一樣），重新找到合適的位置。

private int expungeStaleEntry(int staleSlot) {
            Entry[] tab = table;
            int len = tab.length;
            // expunge entry at staleSlot
            //先將當(dāng)前節(jié)點(diǎn)清除
            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();
                if (k == null) {
                    //key為null 證明是無效數(shù)據(jù) 清除
                    e.value = null;
                    tab[i] = null;
                    size--;
                } else {
                    //重新計(jì)算數(shù)組下標(biāo) 如果數(shù)組下標(biāo)發(fā)生變化 那么將數(shù)據(jù)遷移到新的位置上
                    int h = k.threadLocalHashCode & (len - 1);
                    if (h != i) {
                        tab[i] = null;
                        //重新利用線性尋址法尋找合適的下標(biāo)位置
                        while (tab[h] != null)
                            h = nextIndex(h, len);
                        tab[h] = e;
                    }
                }
            }
            return i;
        }

然后是cleanSomeSlots()方法

cleanSomeSlots()

調(diào)用log(n)次expungeStaleEntry()方法進(jìn)行清除無效數(shù)據(jù)。這個(gè)官方說不調(diào)用n次來清除，為了效率，而且經(jīng)過測試調(diào)用log(n)次清除無效的數(shù)據(jù)的效果已經(jīng)很好了。(n代表entry數(shù)組的長度)。

private boolean cleanSomeSlots(int i, int n) {
            //removed 是否清除了數(shù)據(jù)的標(biāo)記
            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 = len;
                    removed = true;
                    i = expungeStaleEntry(i);
                }
            } while ( (n >>>= 1) != 0);
            return removed;
        }

如果set()方法設(shè)置值之后，需要擴(kuò)容會調(diào)用rehash()方法進(jìn)行擴(kuò)容。

先調(diào)用expungeStaleEntries()清除一下數(shù)據(jù)，如果還是需要擴(kuò)容，那么調(diào)用resize()進(jìn)行擴(kuò)容。

rehash()

       private void rehash() {
            //再試清除一下數(shù)據(jù)
            expungeStaleEntries();
            // Use lower threshold for doubling to avoid hysteresis
            //如果還是需要擴(kuò)容 那么會調(diào)用 resize()進(jìn)行擴(kuò)容
            if (size >= threshold - threshold / 4)
                resize();
        }

resize()

resize()方法會創(chuàng)建一個(gè)容量為原來兩倍的數(shù)組，并且將數(shù)據(jù)遷移到新的數(shù)組上面，將新的數(shù)組賦值給table變量。（擴(kuò)容方法比較簡單）

        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);
                        //線性尋址法解決hash沖突
                        while (newTab[h] != null)
                            h = nextIndex(h, newLen);
                        newTab[h] = e;
                        count++;
                    }
                }
            }
            setThreshold(newLen);
            size = count;
            table = newTab;
        }

get()方法

獲取到當(dāng)前線程的threadLocals，如果threadLocals已經(jīng)初始化，那么調(diào)用getEntry()方法獲取值。否則調(diào)用setInitialValue()獲取我們在initialValue()設(shè)置的初始化的值。

public T get() {
        Thread t = Thread.currentThread();
        //利用當(dāng)前線程獲取它的threadLocals（threadLocals是一個(gè)ThreadLocalMap）
        ThreadLocalMap map = getMap(t);
        if (map != null) {
            ThreadLocalMap.Entry e = map.getEntry(this);
            if (e != null) {
                @SuppressWarnings("unchecked")
                T result = (T)e.value;
                return result;
            }
        }
        return setInitialValue();
    }

現(xiàn)在我們看getEntry()方法

如果找到key相同的Entry 直接返回，否則調(diào)用getEntryAfterMiss()方法

getEntry()

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

getEntryAfterMiss()

getEntryAfterMiss()從當(dāng)前節(jié)點(diǎn)往后遍歷查找，遍歷找到key相同的entry，找到就返回，否則返回null，如果有無效數(shù)據(jù)，順便清除一下。

private Entry getEntryAfterMiss(ThreadLocal<?> key, int i, Entry e) {
            Entry[] tab = table;
            int len = tab.length;
            while (e != null) {
                ThreadLocal<?> k = e.get();
                //找到key相同的entry 直接返回
                if (k == key)
                    return e;
                if (k == null)
                    //當(dāng)前數(shù)據(jù)為無效數(shù)據(jù) 清除一下
                    expungeStaleEntry(i);
                else
                    //否則向后繼續(xù)查找
                    i = nextIndex(i, len);
                e = tab[i];
            }
            return null;
}

最后是remove()方法

remove()

利用hash算法計(jì)算下標(biāo)，從下標(biāo)位置開始往后遍歷，找到key相同的entry，將entry刪除，順便調(diào)用expungeStaleEntry()方法清除一下無效的數(shù)據(jù)。

public void remove() {
         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);
            for (Entry e = tab[i];
                 e != null;
                 e = tab[i = nextIndex(i, len)]) {
                if (e.get() == key) {
                    e.clear();
                    expungeStaleEntry(i);
                    return;
                }
            }
        }