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

概述

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

Recyler我們應(yīng)該也不會太陌生, 因為在之前章節(jié)中, 有好多地方使用了Recyler

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

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

這一章就對FastThreadLocal和Recyler兩個并發(fā)工具類進行分析

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

首先我們看一個最簡單的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();
    }
}

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

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

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

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

然后啟動兩個線程, 每個線程通過fastThreadLocalDemo.fastThreadLocalTest.get()的方式拿到線程共享對象, 因為fastThreadLocalDemo是相同的, 所以fastThreadLocalTest對象也是同一個, 同一個對象在不同線程中進行g(shù)et()

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

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

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

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

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

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

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

這里的index, 代表FastThreadLocal對象的一個下標(biāo), 每創(chuàng)建一個FastThreadLocal, 都會有一個唯一的自增的下標(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()進行原子自增, 創(chuàng)建第一個FastThreadLocal對象時, nextIndex為0, 創(chuàng)建第二個FastThreadLocal對象時nextIndex為1, 以此類推, 第n次nextIndex為n-1, 如圖所示

8-1-1

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

Object obj = fastThreadLocalDemo.fastThreadLocalTest.get();

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

我們跟到get方法中:

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

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

我們跟到InternalThreadLocalMap的get方法中, 分析其實如何獲取InternalThreadLocalMap對象的

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線程, 所以這里會走到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拿到一個ThreadLocal對象slowThreadLocalMap, slowThreadLocalMap是UnpaddedInternalThreadLocalMap類的一個靜態(tài)屬性, 類型是ThreadLocal類型

這里的ThreadLocal是jdk的ThreadLocal

然后通過slowThreadLocalMap對象的get方法, 獲取一個InternalThreadLocalMap

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

因為netty實現(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í)行到這一步的時候會調(diào)用fastGet方法

我們跟進fastGet

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

首先FastThreadLocalThread對象直接通過threadLocalMap拿到threadLocalMap對象

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

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

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

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

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

我們跟到newIndexedVariableTable()中:

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

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

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

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

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

這里我們可以知道, 每個InternalThreadLocalMap對象中都維護了一個Object類型的數(shù)組, 那么這個數(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, 其實也就是拿到每個FastThreadLocal對象的綁定的線程共享對象

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

我們跟到indexedVariable方法中:

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

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

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

這里我們可以分析到, 其實每一個FastThreadLocal對象中所綁定的線程共享對象, 是存放在threadLocalMap對象中的一個對象數(shù)組的中的, 數(shù)組中的元素的下標(biāo)其實就是對應(yīng)著FastThreadLocal中的index屬性, 對應(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ù)以上邏輯, 我們知道, 第一次獲取對象v是只能獲取到UNSET對象, 因為該對象并沒有保存在threadLocalMap中的數(shù)組indexedVariables中, 所以第一次獲取在if判斷中為false, 會走到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實際上走的是FastThreadLocal子類的重寫initialValue方法

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

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

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

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

跟到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對象的index是否超過數(shù)組indexedVariables的長度, 如果沒有超過, 則直接通過下標(biāo)設(shè)置新創(chuàng)建的線程共享對象, 通過這個操作, 下次獲取該對象的時候就可以直接通過數(shù)組下標(biāo)進行取出

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

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