前文傳送門：Netty分布式Future與Promise執(zhí)行回調(diào)相關(guān)邏輯剖析

概述

FastThreadLocal我們?cè)谄饰龆淹鈨?nèi)存分配的時(shí)候簡單介紹過, 它類似于JDK的ThreadLocal, 也是用于在多線程條件下, 保證統(tǒng)一線程的對(duì)象共享, 只是netty中定義的FastThreadLocal, 性能要高于jdk的ThreadLocal, 具體原因會(huì)在之后的小節(jié)進(jìn)行剖析

Recyler我們應(yīng)該也不會(huì)太陌生, 因?yàn)樵谥罢鹿?jié)中, 有好多地方使用了Recyler

Recyler是netty實(shí)現(xiàn)的一個(gè)輕量級(jí)對(duì)象回收站, 很多對(duì)象在使用完畢之后, 并沒有直接交給gc去處理, 而是通過對(duì)象回收站將對(duì)象回收, 目的是為了對(duì)象重用和減少gc壓力

比如ByteBuf對(duì)象的回收, 因?yàn)锽yteBuf對(duì)象在netty中會(huì)頻繁創(chuàng)建, 并且會(huì)占用比較大的內(nèi)存空間, 所以使用完畢后會(huì)通過對(duì)象回收站的方式進(jìn)行回收, 已達(dá)到資源重用的目的

這一章就對(duì)FastThreadLocal和Recyler兩個(gè)并發(fā)工具類進(jìn)行分析

第一節(jié):FastThreadLocal的使用和創(chuàng)建

首先我們看一個(gè)最簡單的demo

public class FastThreadLocalDemo {
    final class FastThreadLocalTest extends FastThreadLocal<Object>{
        @Override
        protected Object initialValue() throws Exception {
            return new Object();
        }
    }
    private final FastThreadLocalTest fastThreadLocalTest;
    public FastThreadLocalDemo(){
        fastThreadLocalTest = new FastThreadLocalTest();
    }
    public static void main(String[] args){
        FastThreadLocalDemo fastThreadLocalDemo = new FastThreadLocalDemo();
        new Thread(new Runnable() {
            @Override
            public void run() {
                Object obj  = fastThreadLocalDemo.fastThreadLocalTest.get();
                try {
                    for (int i=0;i<10;i++){
                        fastThreadLocalDemo.fastThreadLocalTest.set(new Object());
                        Thread.sleep(1000);
                    }
                }catch (Exception e){
                    e.printStackTrace();
                }
            }
        }).start();
        new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    Object obj  = fastThreadLocalDemo.fastThreadLocalTest.get();
                    for (int i=0;i<10;i++){
                        System.out.println(obj == fastThreadLocalDemo.fastThreadLocalTest.get());
                        Thread.sleep(1000);
                    }
                }catch (Exception e){
                }
            }
        }).start();
    }
}

這里首先聲明一個(gè)內(nèi)部類FastThreadLocalTest繼承FastThreadLocal, 并重寫initialValue方法, initialValue方法就是用來初始化線程共享對(duì)象的

然后聲明一個(gè)成員變量fastThreadLocalTest, 類型就是內(nèi)部類FastThreadLocalTest

在構(gòu)造方法中初始化fastThreadLocalTest

main方法中創(chuàng)建當(dāng)前類FastThreadLocalDemo的對(duì)象fastThreadLocalDemo

然后啟動(dòng)兩個(gè)線程, 每個(gè)線程通過fastThreadLocalDemo.fastThreadLocalTest.get()的方式拿到線程共享對(duì)象, 因?yàn)閒astThreadLocalDemo是相同的, 所以fastThreadLocalTest對(duì)象也是同一個(gè), 同一個(gè)對(duì)象在不同線程中進(jìn)行g(shù)et()

第一個(gè)線程循環(huán)通過set方法修改共享對(duì)象的值

第二個(gè)線程則循環(huán)判斷并輸出fastThreadLocalTest.get()出來的對(duì)象和第一次get出來的對(duì)象是否相等

這里輸出結(jié)果都true, 說明其他線程雖然不斷修改共享對(duì)象的值, 但都不影響當(dāng)前線程共享對(duì)象的值

這樣就實(shí)現(xiàn)了線程共享的對(duì)象的功能

根據(jù)上述示例, 我們剖析FastThreadLocal的創(chuàng)建

首先跟到FastThreadLocal的構(gòu)造方法中:

public FastThreadLocal() {
    index = InternalThreadLocalMap.nextVariableIndex();
}

這里的index, 代表FastThreadLocal對(duì)象的一個(gè)下標(biāo), 每創(chuàng)建一個(gè)FastThreadLocal, 都會(huì)有一個(gè)唯一的自增的下標(biāo)

跟到nextVariableIndex方法中

public static int nextVariableIndex() {
    int index = nextIndex.getAndIncrement();
    if (index < 0) {
        nextIndex.decrementAndGet();
        throw new IllegalStateException("too many thread-local indexed variables");
    }
    return index;
}

這里只是獲取nextIndex通過getAndIncrement()進(jìn)行原子自增, 創(chuàng)建第一個(gè)FastThreadLocal對(duì)象時(shí), nextIndex為0, 創(chuàng)建第二個(gè)FastThreadLocal對(duì)象時(shí)nextIndex為1, 以此類推, 第n次nextIndex為n-1, 如圖所示

8-1-1

我們回到demo中, 我們看線程中的這一句:

Object obj = fastThreadLocalDemo.fastThreadLocalTest.get();

這里調(diào)用了FastThreadLocal對(duì)象的get方法, 作用是創(chuàng)建一個(gè)線程共享對(duì)象

我們跟到get方法中:

public final V get() {
    return get(InternalThreadLocalMap.get());
}

這里調(diào)用了一個(gè)重載的get方法, 參數(shù)中通過InternalThreadLocalMap的get方法獲取了一個(gè)InternalThreadLocalMap對(duì)象

我們跟到InternalThreadLocalMap的get方法中, 分析其實(shí)如何獲取InternalThreadLocalMap對(duì)象的

public static InternalThreadLocalMap get() {
    Thread thread = Thread.currentThread();
    if (thread instanceof FastThreadLocalThread) {
        return fastGet((FastThreadLocalThread) thread);
    } else {
        return slowGet();
    }
}

這里首先拿到當(dāng)前線程, 然后判斷當(dāng)前線程是否為FastThreadLocalThread線程, 通常NioEventLoop線程都是FastThreadLocalThread, 用于線程則不是FastThreadLocalThread

在這里, 如果FastThreadLocalThread線程, 則調(diào)用fastGet方法獲取InternalThreadLocalMap, 從名字上我們能知道, 這是一種效率極高的獲取方式

如果不是FastThreadLocalThread線程, 則調(diào)用slowGet方式獲取InternalThreadLocalMap, 同樣根據(jù)名字, 我們知道這是一種效率不太高的獲取方式

我們的demo并不是eventLoop線程, 所以這里會(huì)走到slowGet()方法中

我們首先剖析slowGet()方法

private static InternalThreadLocalMap slowGet() {
    ThreadLocal<InternalThreadLocalMap> slowThreadLocalMap = UnpaddedInternalThreadLocalMap.slowThreadLocalMap;
    InternalThreadLocalMap ret = slowThreadLocalMap.get();
    if (ret == null) {
        ret = new InternalThreadLocalMap();
        slowThreadLocalMap.set(ret);
    }
    return ret;
}

首先通過UnpaddedInternalThreadLocalMap.slowThreadLocalMap拿到一個(gè)ThreadLocal對(duì)象slowThreadLocalMap, slowThreadLocalMap是UnpaddedInternalThreadLocalMap類的一個(gè)靜態(tài)屬性, 類型是ThreadLocal類型

這里的ThreadLocal是jdk的ThreadLocal

然后通過slowThreadLocalMap對(duì)象的get方法, 獲取一個(gè)InternalThreadLocalMap

如果第一次獲取, InternalThreadLocalMap有可能是null, 所以在if塊中, new了一個(gè)InternalThreadLocalMap對(duì)象, 并設(shè)置在ThreadLocal對(duì)象中

因?yàn)閚etty實(shí)現(xiàn)的FastThreadLocal要比jdk的ThreadLocal要快, 所以這里的方法叫slowGet

回到InternalThreadLocalMap的get方法:

public static InternalThreadLocalMap get() {
    Thread thread = Thread.currentThread();
    if (thread instanceof FastThreadLocalThread) {
        return fastGet((FastThreadLocalThread) thread);
    } else {
        return slowGet();
    }
}

我們繼續(xù)剖析fastGet方法, 通常EventLoop線程FastThreadLocalThread線程, 所以EventLoop線程執(zhí)行到這一步的時(shí)候會(huì)調(diào)用fastGet方法

我們跟進(jìn)fastGet

private static InternalThreadLocalMap fastGet(FastThreadLocalThread thread) {
    InternalThreadLocalMap threadLocalMap = thread.threadLocalMap();
    if (threadLocalMap == null) {
        thread.setThreadLocalMap(threadLocalMap = new InternalThreadLocalMap());
    }
    return threadLocalMap;
}

首先FastThreadLocalThread對(duì)象直接通過threadLocalMap拿到threadLocalMap對(duì)象

如果threadLocalMap為null, 則創(chuàng)建一個(gè)InternalThreadLocalMap對(duì)象設(shè)置到FastThreadLocalThread的成員變量中

這里我們可以知道FastThreadLocalThread對(duì)象中維護(hù)了一個(gè)InternalThreadLocalMap類型的成員變量, 可以直接通過threadLocalMap()方法獲取該變量的值, 也就是InternalThreadLocalMap

我們跟到InternalThreadLocalMap的構(gòu)造方法中:

private InternalThreadLocalMap() {
    super(newIndexedVariableTable());
}

這里調(diào)用了父類的構(gòu)造方法, 傳入一個(gè)newIndexedVariableTable()

我們跟到newIndexedVariableTable()中:

private static Object[] newIndexedVariableTable() {
    Object[] array = new Object[32];
    Arrays.fill(array, UNSET);
    return array;
}

這里創(chuàng)建一個(gè)長度為32的數(shù)組, 并為數(shù)組中的每一個(gè)對(duì)象設(shè)置為UNSET, UNSET是一個(gè)Object的對(duì)象, 表示該下標(biāo)的值沒有被設(shè)置

回到InternalThreadLocalMap的構(gòu)造方法, 再看其父類的構(gòu)造方法:

UnpaddedInternalThreadLocalMap(Object[] indexedVariables) {
    this.indexedVariables = indexedVariables;
}

這里初始化了一個(gè)數(shù)組類型的成員變量indexedVariables, 就是newIndexedVariableTable返回object的數(shù)組

這里我們可以知道, 每個(gè)InternalThreadLocalMap對(duì)象中都維護(hù)了一個(gè)Object類型的數(shù)組, 那么這個(gè)數(shù)組有什么作用呢?我們繼續(xù)往下剖析

回到FastThreadLocal的get方法中

public final V get() {
    return get(InternalThreadLocalMap.get());
}

我們剖析完了InternalThreadLocalMap.get()的相關(guān)邏輯, 再繼續(xù)看重載的get方法:

public final V get(InternalThreadLocalMap threadLocalMap) {
    Object v = threadLocalMap.indexedVariable(index);
    if (v != InternalThreadLocalMap.UNSET) {
        return (V) v;
    }
    return initialize(threadLocalMap);
}

首先看這一步:

Object v = threadLocalMap.indexedVariable(index);

這一步是拿到當(dāng)前index下標(biāo)的object, 其實(shí)也就是拿到每個(gè)FastThreadLocal對(duì)象的綁定的線程共享對(duì)象

index是我們剛才分析過, 是每一個(gè)FastThreadLocal的唯一下標(biāo)

我們跟到indexedVariable方法中:

public Object indexedVariable(int index) {
    Object[] lookup = indexedVariables;
    return index < lookup.length? lookup[index] : UNSET;
}

這里首先拿到indexedVariables, 我們剛才分析過, indexedVariables是InternalThreadLocalMap對(duì)象中維護(hù)的數(shù)組, 初始大小是32

然后再return中判斷當(dāng)前index是不是小于當(dāng)前數(shù)組的長度, 如果小于則獲取當(dāng)前下標(biāo)index的數(shù)組元素, 否則返回UNSET代表沒有設(shè)置的對(duì)象

這里我們可以分析到, 其實(shí)每一個(gè)FastThreadLocal對(duì)象中所綁定的線程共享對(duì)象, 是存放在threadLocalMap對(duì)象中的一個(gè)對(duì)象數(shù)組的中的, 數(shù)組中的元素的下標(biāo)其實(shí)就是對(duì)應(yīng)著FastThreadLocal中的index屬性, 對(duì)應(yīng)關(guān)系如圖所示

8-1-2

回到FastThreadLocal重載的get方法:

public final V get(InternalThreadLocalMap threadLocalMap) {
    Object v = threadLocalMap.indexedVariable(index);
    if (v != InternalThreadLocalMap.UNSET) {
        return (V) v;
    }
    return initialize(threadLocalMap);
}

根據(jù)以上邏輯, 我們知道, 第一次獲取對(duì)象v是只能獲取到UNSET對(duì)象, 因?yàn)樵搶?duì)象并沒有保存在threadLocalMap中的數(shù)組indexedVariables中, 所以第一次獲取在if判斷中為false, 會(huì)走到initialize方法中

跟到initialize方法中:

private V initialize(InternalThreadLocalMap threadLocalMap) {
    V v = null;
    try {
        v = initialValue();
    } catch (Exception e) {
        PlatformDependent.throwException(e);
    }
    threadLocalMap.setIndexedVariable(index, v);
    addToVariablesToRemove(threadLocalMap, this);
    return v;
}

這里首先調(diào)用的initialValue方法, 這里的initialValue實(shí)際上走的是FastThreadLocal子類的重寫initialValue方法

在我們的demo中對(duì)應(yīng)這個(gè)方法

@Override
protected Object initialValue() throws Exception {
    return new Object();
}

通過這個(gè)方法會(huì)創(chuàng)建一個(gè)線程共享對(duì)象

然后通過threadLocalMap對(duì)象的setIndexedVariable方法將創(chuàng)建的線程共享對(duì)象設(shè)置到threadLocalMap中維護(hù)的數(shù)組中, 參數(shù)為FastThreadLocal和創(chuàng)建的對(duì)象本身

跟到setIndexedVariable方法中:

public boolean setIndexedVariable(int index, Object value) {
    Object[] lookup = indexedVariables;
    if (index < lookup.length) {
        Object oldValue = lookup[index];
        lookup[index] = value;
        return oldValue == UNSET;
    } else {
        expandIndexedVariableTableAndSet(index, value);
        return true;
    }
}

這里首先判斷FastThreadLocal對(duì)象的index是否超過數(shù)組indexedVariables的長度, 如果沒有超過, 則直接通過下標(biāo)設(shè)置新創(chuàng)建的線程共享對(duì)象, 通過這個(gè)操作, 下次獲取該對(duì)象的時(shí)候就可以直接通過數(shù)組下標(biāo)進(jìn)行取出

如果index超過了數(shù)組indexedVariables的長度, 則通過expandIndexedVariableTableAndSet方法將數(shù)組擴(kuò)容, 并且根據(jù)index的通過數(shù)組下標(biāo)的方式將線程共享對(duì)象設(shè)置到數(shù)組indexedVariables中

以上就是線程共享對(duì)象的創(chuàng)建和獲取的過程，更多關(guān)于Netty分布式工具類FastThreadLocal和Recycler的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章！

最新評(píng)論