1 性能優化工具類

1.1 FastThreadLocal

1.1.1 傳統的ThreadLocal

ThreadLocal最常用的兩個接口是set和get
最常見的應用場景為在線程上下文之間傳遞信息，使得用戶不受復雜代碼邏輯的影響

public void set(T value) {Thread t = Thread.currentThread();ThreadLocalMap map = getMap(t); t.threadLocals;

我們使用set的時候實際上是獲取Thread對象的threadLocals屬性，把當前ThreadLocal當做參數然后調用其set(ThreadLocal，Object)方法來設值
threadLocals是ThreadLocal.ThreadLocalMap類型的

每個線程對象關聯著一個ThreadLocalMap實例,主要是維護著一個Entry數組
Entry是擴展了WeakReference，提供了一個存儲value的地方
一個線程對象可以對應多個ThreadLocal實例，一個ThreadLocal也可以對應多個Thread對象，當一個Thread對象和每一個ThreadLocal發生關系的時候會生成一個Entry，并將需要存儲的值存儲在Entry的value內

• 一個ThreadLocal對于一個Thread對象來說只能存儲一個值，為Object型
• 多個ThreadLocal對于一個Thread對象，這些ThreadLocal和線程相關的值存儲在Thread對象關聯的ThreadLocalMap中
• 使用擴展WeakReference的Entry作為數據節點在一定程度上防止了內存泄露
• 多個Thread線程對象和一個ThreadLocal發生關系的時候其實真實數據的存儲是跟著線程對象走的，因此這種情況不討論

我們在看看ThreadLocalMap#set：

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)]) {ThreadLocal k = e.get();if (k == key) {e.value = value;return;}if (k == null) {replaceStaleEntry(key, value, i);return;} } tab[i] = new Entry(key, value); int sz = ++size; if (!cleanSomeSlots(i, sz) && sz >= threshold)rehash();

每個ThreadLocal實例都有一個唯一的threadLocalHashCode初始值
上面首先根據threadLocalHashCode值計算出i，有下面兩種情況會進入for循環：

• 由于threadLocalHashCode &（len-1）對應的槽有內容，因此滿足tab[i]!=null條件，進入for循環，如果滿足條件且當前key不是當前threadlocal只能說明hash沖突了
• ThreadLocal實例之前被設置過，因此滿足tab[i]!=null條件，進入for循環

進入for循環會遍歷tab數組，如果遇到以當前threadLocal為key的槽，即上面第（2）種情況，有則直接將值替換
如果找到了一個已經被回收的ThreadLocal對應的槽，也就是當key==null的時候表示之前的threadlocal已經被回收了，但是value值還存在，這也是ThreadLocal內存泄露的地方。碰到這種情況，則會引發替換這個位置的動作
如果上面兩種情況都沒發生，即上面的第（1）種情況，則新創建一個Entry對象放入槽中

private Entry getEntry(ThreadLocal key) {int i = key.threadLocalHashCode & (table.length - 1);Entry e = table[i];if (e != null && e.get() == key)return e;elsereturn getEntryAfterMiss(key, i, e); }

當命中的時候，也就是根據當前ThreadLocal計算出來的i恰好是當前ThreadLocal設置的值的時候，可以直接根據hashcode來計算出位置，當沒有命中的時候，這里沒有命中分為三種情況：

• 當前ThreadLocal之前沒有設值過，并且當前槽位沒有值。
• 當前槽位有值，但是對于的不是當前threadlocal，且那個ThreadLocal沒有被回收。
• 當前槽位有值，但是對于的不是當前threadlocal，且那個ThreadLocal被回收了。

上面三種情況都會調用getEntryAfterMiss方法。調用getEntryAfterMiss方法會引發數組的遍歷。

總結一下ThreadLocal的性能，一個線程對應多個ThreadLocal實例的場景中
在沒有命中的情況下基本上一次hash就可以找到位置
如果發生沒有命中的情況，則會引發性能會急劇下降，當在讀寫操作頻繁的場景，這點將成為性能詬病。

1.1.2 實例

兩個線程操作同一object 對象,顯然非線程安全,但是由于使用了 FTL, 線程安全!

結果表明內存地址不同,并非操作同一個 object!

讓T1每1s 中新生成一個 object 對象
T2驗證當前 object 是否與之前狀態相同

顯然,每個線程拿到的對象都是線程獨享的!
某線程對變量的修改不影響其他線程!
通過對象隔離優化了程序性能!

1.1.3 Netty FastThreadLocal源碼解析

1.1.3.1 創建

創建時重寫一下初始值方法

實際上在構造FastThreadLocal實例的時候就決定了這個實例的索引

index 為 private 且非 static, 說明每個實例都有該值

再看看索引的生成相關代碼

index 從0開始計數


nextIndex是InternalThreadLocalMap父類的一個全局靜態的AtomicInteger類型的對象，這意味著所有的FastThreadLocal實例將共同依賴這個指針來生成唯一的索引，而且是線程安全的
Netty重新設計了更快的FastThreadLocal，主要實現涉及

• FastThreadLocalThread
• FastThreadLocal
• InternalThreadLocalMap

FastThreadLocalThread是Thread類的簡單擴展，主要是為了擴展threadLocalMap屬性

FastThreadLocal提供的接口和傳統的ThreadLocal一致，主要是set和get方法，用法也一致
不同地方在于FastThreadLocal的值是存儲在InternalThreadLocalMap這個結構里面的，傳統的ThreadLocal性能槽點主要是在讀寫的時候hash計算和當hash沒有命中的時候發生的遍歷，我們來看看FastThreadLocal的核心實現

InternalThreadLocalMap實例和Thread對象一一對應
UnpaddedInternalThreadLocalMap維護著一個數組：

這個數組用來存儲跟同一個線程關聯的多個FastThreadLocal的值，由于FastThreadLocal對應indexedVariables的索引是確定的，因此在讀寫的時候將會發生隨機存取，非常快。

另外這里有一個問題，nextIndex是靜態唯一的，而indexedVariables數組是實例對象的，因此我認為隨著FastThreadLocal數量的遞增，這會造成空間的浪費

1.1.3.2 get方法實現

獲取 ThreadLocalMap

首先拿到當前線程,再判斷是否為 FTL 線程快速獲取否則慢速獲取

• 讓我們先分析一下 slowGet方法
  
  首先會獲取一個 ThreadLocal 變量
  
  拿到 JDK 的 ThreadLocal 變量,用于給每個線程拿到InternalThreadLocalMap變量,所以過程較慢,該方法稱為 slowGet 可想而知!
  
  由于在創建 ThreadLocal 時,并沒有重寫 initValue 方法,所以可能為 null
• 接下啦看 fastGet 方法
  
  
  
  

直接通過索引取出對象

通過每個線程獨享的 ThreadLocalMap 對象借助在 JVM 中每個 FTL 的唯一索引

1.2 輕量級對象池 Recycler

1.2.1 Recycler的使用

所以不使用 new 而是直接復用

Netty使用

1.2.2 Recycler的創建

• 創建方式為直接new 一個 Recycler 對象,然后重寫 newObject 方法
  
• 轉到構造方法
  

再看看每個Recycler 的結構是如何的

• 每個Recycler 中對應每條線程都持有一個 Stack 對象
  
• 下面圖示說明

看下 Stack對象 的 element 參數,一些默認處理器的數組,該數組實際存放對象池,每個處理器都包裝了一個對象,handler 可被外部對象引用.,從而回收該對象

• 參數列表

• 其中,radiomask 控制對象回收的比率,所以并非每次調用recycler 都會發生回收

• maxcapacity 池子最大元素容量

• 線程1殘留的對象會緩存到線程2中繼續釋放
  所以 maxdelayedqueues 也就是可以緩存對象的線程數.如若再有個線程3,而隊列結構在線程2,那3會直接拋棄1的殘留對象.

• availablesharedcapacity:線程1中創建的對象能夠在其他線程中緩存的對象的最大個數.
  
  以上即為 Stack 對象所有成員變量.

下面回到Recycler的構造方法,看其入參.

該數值即為Stack 數組元素能有多少個

• 再看看如下構造方法的參數.
  
• 默認值為2
• 看看新的構造器的 radio 參數
• 默認值8
  
• 兩倍CPU核心數
  
• 自然該值為7
  

1.2.3 回收對象到 Recycler

1.2.3.1 同線程回收

• 客戶端開始調用
  
  
• Recycler抽象類
• 將當前對象壓棧
• 如下,首先判斷當前線程,thread 即為S tack 對象中保存的成員變量,若是創建該 stack 的線程,則直接壓棧Stack 中,若不是再 pushlater.先分析 pushnow.

• 首先驗證兩個 id,由于默認初始值為0,所以通過判斷.
  

• 接下來將其都賦值為第三個 id,該值在整個Recycler 中都是唯一確定值
  

緊接著判斷當前 size是否已到 maxcapacity,若達到上限,直接丟棄該對象即直接 return;否則繼續判斷 drop 處理器

首先判斷,若該對象之前未被回收過,繼續判斷;
至今已經回收了多少個對象,其中 rm 為7,即111(二進制表示),即每隔8個對象,就會來此判斷一次,將其與7進行與運算后,若不為0,則返回 true,表示只回收八分之一的對象.

繼續回到 pushnow 的流程,接下來判斷 size 是否等于數組的容量.
因為 element 是一個數組,并不是一開始就創建maxcapacity 容量大小,若容量不夠了,則進行兩倍大小擴容,再將其加入數組.

1.2.3.2 異線程回收對象

• 本節食用指南

1.2.3.2.1 獲取 WeakOrderQueue(以下簡稱WOQ)

由前面 pushnow 進入同線程回收, pushlater 即進入異線程回收過程.

• 先看看這么個東西是啥
  其類型就很神奇了,首先是個FTL,即每個線程都有一個 map,map 的key=stack 表示對于不同線程,對不同 stack 來說對應于不同的WOQ.
  那么它為何要定義成一個 map 結構呢,假設有3個線程T1/2/3;
  T1創建的對象肯可能跑到T3中回收,T2中創建的對象也可能到了T3回收.
  那么元素就是T1以及T2的WOQ.

假設當前在T2中,接下呢就通過get(this)拿到T1的WOQ,其中的 this 指的就是T1的 Stack.
然后若 queue==null,即表示T2從未回收過T1的對象,接下來開始判斷
當前的即T2已經回收過的線程數 size,若不小于 mDQ,說明已經不能再回收其他線程的對象了!
給WOQ設置 dummy 標志,即對應下面的若下次看到一個線程標志了 dummy 直接return;什么也不做.

以上即為第一個過程,從FTL中拿一個Stack 對應的WOQ.

1.2.3.2.2 創建 WeakOrderQueue

若之前沒拿到呢,那就直接創建一個WOQ吧!

• 接下來讓我們看看一個線程創建WOQ是如何與待回收對象的Stack 進行綁定的.
  
  其中的 this 即為 stack,是在T1中維護的,thread 即表示當前線程T2.
  allocat就是為了給當前線程T2分配一個在T1中的Stack 對應的一個WOQ.
  
  首先判斷,T1中的 Stack還能否再分配LINK_capacity 個內存,若不能直接返回 null;
  若可以,就直接 new 一個WOQ.
  讓我們具體看看其實現.
  
  此函數意義為:該 Stack 允許外部線程給它緩存多少個對象
  經過CAS操作設置該值為Stack 可為其他線程共享的回收對象的個數.

容量足夠,則直接創建一個WOQ,下面來看看其數據結構.

一個鏈表結構.將其 handle與Link 進行分離,極大地提升了性能,
因為不必判斷當前T2能否回收T1的對象,而只需判斷當前的L ink 中是否有空的,則可直接將 hande 塞進去.因為在前面一次性的判斷過,從T1中是否能批量分配這么多對象(以減少很多操作的頻率).

使用同步,將WOQ插到Stack 的頭部.

1.2.3.2.3 將對象追加到 WeakOrderQueue

• 一開始呢,就是這么創建一個WOQ,默認有16個 handle
  

• T2已經拿到queue,接著就是添加元素.
  

首先設置 上次回收 id.

• 該 id 為WOQ的 id,所以是以WOQ為基礎的
  

然后拿到尾指針,獲取Link 的長度,若已經等于 link_capacity,說明已經不可寫了;
繼續判斷 想辦法看看T1是否還能再分配一個Link來保存待回收的對象.
不允許,則直接丟棄;
允許,則直接創建Link并重新賦值 tail 節點.

創建完后,拿到其寫指針,即 tail 的長度(0).所以 tail 節點也已經又有了足夠的存儲空間,將 handle 追加進去.再將該 handle 的 stack 指針重置為 null,因為已經不屬于原來的 stack 了.
最后,寫指針+1.

1.2.4 從 Recycler 獲取對象

本節分析若當前 stack 為空

若當前線程T1去獲取對象,若 stack 中有對象,則直接拿出.T1所擁有的對象即為T1擁有的 stack 中的對象,若發現其中為空,會嘗試與 和T1的 stack 關聯的WOQ中的 T1創建的,但是在其他線程中去回收的對象.那么,T1中對象不足,就需要在其他線程中去回收.

其中的 cusor 指針即當前所需要回收的對象

• 彈棧獲取元素
• 若 size 為0,則從其他線程回收
  若已經回收到則直接 return true.沒有則重置兩個指針,將 cusor 指向頭結點,意味著準備從頭開始回收.

接下來具體分析這段長代碼

boolean scavengeSome() {WeakOrderQueue prev;// 先拿到 cusorWeakOrderQueue cursor = this.cursor;// cusor 節點無對象if (cursor == null) {prev = null;// 指向頭結點cursor = head;// 頭結點依舊為空,已經沒有與之關聯的WOQ,直接返回 false.if (cursor == null) {return false;}} else {prev = this.prev;}boolean success = false;// 此處 do/while 循環只為去尋找與 stack 關聯的WOQ,看看到底能不能碰到一個對象.do {// transfer 即為了將WOQ中的對象傳輸到 stack 中.成功獲取則結束循環!if (cursor.transfer(this)) {success = true;break;}// 沒有回收成功,則看往 cusor 的下一個節點WeakOrderQueue next = cursor.next;// owner 為與當前WOQ關聯的一個線程(對應圖中的T4)// 為空,說明T4已經不存在!隨后即,做一些善后清理工作if (cursor.owner.get() == null) {// If the thread associated with the queue is gone, unlink it, after// performing a volatile read to confirm there is no data left to collect.// We never unlink the first queue, as we don't want to synchronize on updating the head.// 判斷節點中是否還有數據if (cursor.hasFinalData()) {// 就需要想辦法將數據傳輸到 stack 中for (;;) {if (cursor.transfer(this)) {success = true;} else {break;}}}// 處理完該節點后,即將其刪除,通過傳統的指針的刪除方法if (prev != null) {prev.setNext(next);}// T4還存活,繼續看后繼節點.} else {prev = cursor;}cursor = next;// cusor 為空時,誒就結束循環啦!} while (cursor != null && !success);this.prev = prev;this.cursor = cursor;return success;}

• 下面看傳輸方法

// transfer as many items as we can from this queue to the stack, returning true if any were transferred@SuppressWarnings("rawtypes")boolean transfer(Stack<?> dst) {// 先找到頭結點Link head = this.head;if (head == null) {// WOQ無數據,直接返回 falsereturn false;}// 說明當前Link 的所有數據已被取走.if (head.readIndex == LINK_CAPACITY) {if (head.next == null) {return false;}this.head = head = head.next;}// 從該索引開始取對象final int srcStart = head.readIndex;int srcEnd = head.get();// 當前Link 要傳輸到 Stack 的對象個數.final int srcSize = srcEnd - srcStart;if (srcSize == 0) {return false;}

// dst 為當前Stackfinal int dstSize = dst.size;// 將對象都傳輸到Stack 所需容量.final int expectedCapacity = dstSize + srcSize;// 所需容量大于Stack 的大小,if (expectedCapacity > dst.elements.length) {// 因此進行擴容final int actualCapacity = dst.increaseCapacity(expectedCapacity);// 可傳輸的最后一個對象srcEnd = min(srcStart + actualCapacity - dstSize, srcEnd);}if (srcStart != srcEnd) {final DefaultHandle[] srcElems = head.elements;final DefaultHandle[] dstElems = dst.elements;int newDstSize = dstSize;for (int i = srcStart; i < srcEnd; i++) {DefaultHandle element = srcElems[i];//為0說明未被回收過if (element.recycleId == 0) {element.recycleId = element.lastRecycledId;} else if (element.recycleId != element.lastRecycledId) {throw new IllegalStateException("recycled already");}srcElems[i] = null;if (dst.dropHandle(element)) {// Drop the object.continue;}element.stack = dst;dstElems[newDstSize ++] = element;}if (srcEnd == LINK_CAPACITY && head.next != null) {// Add capacity back as the Link is GCed.reclaimSpace(LINK_CAPACITY);this.head = head.next;}head.readIndex = srcEnd;if (dst.size == newDstSize) {return false;}dst.size = newDstSize;return true;} else {// The destination stack is full already.return false;}}

1.3 小結

參考

Java讀源碼之Netty深入剖析

總結

以上是生活随笔為你收集整理的Netty源码实战(十) - 性能优化的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。