深度解析線程本地儲存(TLS)ThreadLocal實現原理

2024年2月6日 25点热度 0人点赞

線程本地變量(Thread Local Storage,TLS)是一種特殊的變量,它的值對於每個線程都是獨立存儲的。這意味著每個線程都可以擁有該變量的一份獨立副本,在多線程環境中,每個線程可以獨立操作自己的變量副本,而不會影響其他線程的副本。以下是線程本地變量的特點:

(1)、線程隔離:每個線程都有自己的變量副本,彼此獨立,互不幹擾。
(2)、線程安全:由於每個線程都操作自己的副本,無需額外的同步措施,因此在多線程環境中使用線程本地變量可以避免競態條件和線程安全問題。
(3)、高效性:線程本地變量的訪問速度通常比全局變量或共享變量更快,因為它們存儲在每個線程的棧上,訪問時無需進行線程間的同步。
(4)、上下文保存:線程本地變量可以用於保存線程的上下文信息,這些信息對於特定線程的執行過程非常重要。

.NET中線程本地存儲包含兩種實現方式:原生實現(調用操作系統提供的接口訪問原生線程對應的線程本地儲存)和托管實現(.NET提供的接口訪問托管線程對應的線程本地儲存)。對於.NET的托管實現方式有以下幾種:

(1)、ThreadLocal :允許你創建線程本地變量,每個線程都有自己的變量副本。
(2)、AsyncLocal :提供了異步代碼中線程本地存儲的支持。在異步代碼中正確地傳播線程本地存儲的值。
(3)、CallContext :提供線程間數據傳遞的機制,允許將數據附加到調用的線程,然後在線程的整個執行過程中進行傳遞。
(4)、TLS Slots:一種低級機制,允許開發人員為每個線程分配一個數組以存儲線程本地數據。

.NET雖然提供了多種線程本地存儲的機制和實現方式,但是最常用的還是ThreadLocal ,這次我們就看重點看一下ThreadLocal 的內部實現原理和應用建議,以便讓大傢在實際的項目中去更好的使用和掌握。首先我們還是以一個簡單的樣例開始這次的探索之旅。

using System;
using System.Threading;
class Program
{
    // 聲明一個線程本地變量
    static ThreadLocal<int> _threadLocalValue = new ThreadLocal<int>(() =>
    {
        // 線程本地變量的初始值
        return Thread.CurrentThread.ManagedThreadId;
    });
    static void Main(string[] args)
    {
        // 啟動多個線程
        Thread[] threads = new Thread[5];
        for (int i = 0; i < threads.Length; i  )
        {
            threads[i] = new Thread(ThreadMethod);
            threads[i].Start();
        }
        // 等待所有線程結束
        foreach (Thread thread in threads)
        {
            thread.Join();
        }
        Console.WriteLine("Main線程結束");
    }
    static void ThreadMethod()
    {
        // 獲取線程本地變量的值
        int localValue = _threadLocalValue.Value;
        Console.WriteLine($"線程 {Thread.CurrentThread.ManagedThreadId} 的線程本地變量值為: {localValue}");
        // 修改線程本地變量的值
        _threadLocalValue.Value = Thread.CurrentThread.ManagedThreadId * 100;
        // 再次獲取線程本地變量的值
        localValue = _threadLocalValue.Value;
        Console.WriteLine($"線程 {Thread.CurrentThread.ManagedThreadId} 的線程本地變量值修改為: {localValue}");
    }
}

上面的示例創建了一個線程本地變量 _threadLocalValue,每個線程都可以擁有該變量的一份獨立副本。在每個線程中,初始值為當前線程的ID。然後,每個線程修改自己的副本,而不會影響其他線程的副本。接下來我們逐步查看其內部的實現邏輯,首先看一下對象的初始化。

        private void Initialize(Func<T>? valueFactory, bool trackAllValues)
        {
            _valueFactory = valueFactory;
            _trackAllValues = trackAllValues;
             _idComplement = ~s_idManager.GetId(trackAllValues);
            _initialized = true;
        }

Initialize(Func? valueFactory, bool trackAllValues)初始化線程本地變量(ThreadLocal),valueFactory用於在需要時創建線程本地變量的值,trackAllValues 表示是否要跟蹤所有線程的值;~s_idManager.GetId(trackAllValues)對象獲取了一個唯一的ID,並將其按位取反;_initialized最後標記線程本地變量實例已經完全初始化。這個字段的目的是跟蹤實例是否已經成功初始化。如果在構造函數中發生了異常,此字段將保持為 false,以便後續操作可以識別並進行適當的處理。

接下來我們看看Value屬性的實現邏輯,該屬性用於獲取和設置線程本地變量的值。該屬於屬性的主體部分包含了 get 和 set 訪問器,用於獲取和設置線程本地變量的值。

        public T Value
        {
            get
            {
                LinkedSlotVolatile[]? slotArray = ts_slotArray;
                LinkedSlot? slot;
                int id = ~_idComplement;
                if (slotArray != null  && id >= 0 && id < slotArray.Length  
                    && (slot = slotArray[id].Value) != null && _initialized)
                {
                    return slot._value;
                }
                return GetValueSlow();
            }
            set
            {
                LinkedSlotVolatile[]? slotArray = ts_slotArray;
                LinkedSlot? slot;
                int id = ~_idComplement;
                if (slotArray != null && id >= 0 && id < slotArray.Length  
                    && (slot = slotArray[id].Value) != null && _initialized)
                {
                    slot._value = value;
                }
                else
                {
                    SetValueSlow(value, slotArray);
                }
            }
        }

我們重點來看一下"slotArray != null && id >= 0 && id < slotArray.Length && (slot = slotArray[id].Value) != null && _initialized"這一部分的邏輯,這裡用於檢查是否可以通過快速路徑來獲取或設置線程本地變量的值。

(1)、slotArray != null:檢查線程本地變量的數組 slotArray 是否已經被初始化。如果 slotArray  null,那麼說明線程本地變量還沒有被初始化,無法通過快速路徑來獲取或設置值。
(2)、id >= 0:檢查當前線程的 ID 是否為非負數。ID 的取反值在初始化時會被賦值給 _idComplement,因此取反後的值非負表示該線程本地變量實例未被銷毀。如果 ID 是負數,說明實例已被銷毀,無法再進行操作。
(3)、id < slotArray.Length:檢查當前線程的 ID 是否在合法范圍內,即是否小於 slotArray 的長度。如果超出了數組的索引范圍,那麼該線程本地變量實例也無法被訪問。
(4)、(slot = slotArray[id].Value) != null:嘗試從 slotArray 中獲取與當前線程 ID 對應的 LinkedSlot 對象,並將其賦給 slot 變量。然後它檢查 slot 是否為 null,如果不為 null,表示已經為當前線程分配了對應的 LinkedSlot 對象,可以繼續進行後續操作。
(5)、_initialized:用於確保線程本地變量實例仍然處於初始化狀態。如果該標志為 false,可能表示在構造函數中發生了異常,導致實例未能完全初始化。在這種情況下,需要通過慢速路徑來處理獲取或設置值的操作。

這些條件一起確保了隻有在線程本地變量實例已經被正確初始化,且對應的 LinkedSlot 對象已經被分配的情況下,才能通過快速路徑來獲取或設置值。我們從上面的代碼中可以發現兩種類型,分別是快速路徑和慢速路徑,至於為什麼會有這兩種方式,主要是考慮到多線程環境下的線程安全性和性能問題,因此可能會存在快速路徑和慢速路徑的區分。主要的原因包括:

(1)、慢速路徑用於處理復雜情況:在多線程環境下,線程之間的並發訪問可能會導致競態條件和數據不一致的問題。因此,ThreadLocal 需要在慢速路徑中處理諸如資源分配、線程同步、異常處理等復雜情況,以確保線程安全和數據一致性。
(2)、慢速路徑可能涉及額外的操作:在某些情況下,獲取或設置線程本地變量的值可能需要進行額外的操作,例如檢查線程的狀態、處理異常情況、執行資源回收等。這些額外的操作會增加代碼的復雜性和執行的開銷,因此可能會導致慢速路徑的執行。
(3)、慢速路徑用於處理初始化和清理:在初始化 ThreadLocal 實例或線程本地變量的值時,可能涉及到較為復雜的初始化過程,例如調用用戶提供的工廠方法、執行必要的線程同步操作等。同樣,當 ThreadLocal 實例或線程本地變量被銷毀時,可能需要執行一些清理操作,如釋放資源、處理異常等。
(4)、性能和安全的權衡:為了保證線程安全和數據一致性,有時必須犧牲一定的性能。因此,在一些情況下,為了處理復雜的線程同步和數據訪問問題,可能需要采用慢速路徑來保證程序的正確性。

我們來具體看一下該屬性內的兩個核心方法的實現SetValueSlow(value, slotArray)和GetValueSlow()。我們先來看一下SetValueSlow(value, slotArray)方法的實現邏輯。SetValueSlow負責在慢速路徑上對線程本地變量的值進行設置。它包括了創建線程本地變量數組、增加數組大小以及創建新的 LinkedSlot 等操作,以確保可以正確設置線程本地變量的值。

        private void SetValueSlow(T value, LinkedSlotVolatile[]? slotArray)
        {
            // 獲取當前線程的ID,ID的取反值在初始化時會被賦值給_idComplement,因此取反後的值就是線程的ID。
            int id = ~_idComplement;
            //檢查線程是否已被銷毀,如果線程的ID為負數,說明該線程已經被銷毀
            ObjectDisposedException.ThrowIf(id < 0, this);
            // 如果當前線程的線程本地變量數組為 null,說明尚未為該線程創建數組,
            // 因此需要在這裡創建線程本地變量數組,並進行相關的初始化工作。
            if (slotArray == null)
            {
                slotArray = new LinkedSlotVolatile[GetNewTableSize(id   1)];
                ts_finalizationHelper = new FinalizationHelper(slotArray);
                ts_slotArray = slotArray;
            }
            // 如果當前線程的ID超出了數組的長度,說明數組不夠大,需要增加數組的大小以容納新的線程本地變量值。
            if (id >= slotArray.Length)
            {
                GrowTable(ref slotArray!, id   1);
                ts_finalizationHelper.SlotArray = slotArray;
                ts_slotArray = slotArray;
            }
            // 如果在數組中對應位置的LinkedSlot為空,則需要創建一個新的LinkedSlot 對象,並將其添加到數組中。
            if (slotArray[id].Value == null)
            {
                CreateLinkedSlot(slotArray, id, value);
            }
            else
            {
                LinkedSlot? slot = slotArray[id].Value;
                slot!._value = value;
            }
        }

從上面的源碼中發現了兩個核心的實現方法, GrowTable(ref slotArray!, id 1)和CreateLinkedSlot(slotArray, id, value),分別是用於擴容和創建LinkedSlot,我們將從這兩個方法中來分析線程本地存儲對象的數據結構是如何的。

我們先看一下CreateLinkedSlot(LinkedSlotVolatile[] slotArray, int id, T value)方法,該方法用於創建一個 LinkedSlot 並將其插入到由 ThreadLocal 實例維護的鏈表中。LinkedSlotVolatile 結構體用作線程本地變量數組的元素類型,而 LinkedSlot 類用作線程本地變量的節點,在 ThreadLocal 實例中維護一個雙向鏈表。通過volatile 關鍵字的使用,確保了在多線程環境下對線程本地變量的讀取和寫入的一致性。

        private void CreateLinkedSlot(LinkedSlotVolatile[] slotArray, int id, T value)
        {
            // 創建一個新的LinkedSlot 對象,並傳入slotArray 數組作為參數
            var linkedSlot = new LinkedSlot(slotArray);
            // 使用 s_idManager 對象進行鎖定,確保在多線程環境中創建和插入過程的原子性。
            lock (s_idManager)
            {
                // 獲取鏈表中的第一個真實節點,這個節點不是啞頭結點。
                // 鏈表的結構是由一個啞頭結點和真實節點組成,啞頭結點位於鏈表頭部。
                LinkedSlot? firstRealNode = _linkedSlot._next;
                // 設置新節點的下一個節點為原鏈表中的第一個真實節點,
                // 上一個節點為啞頭結點,然後將新節點的值設置為傳入的value。
                linkedSlot._next = firstRealNode;
                linkedSlot._previous = _linkedSlot;
                linkedSlot._value = value;
                // 如果原鏈表中存在真實節點,則將原鏈表中的第一個真實節點的上一個節點設置為新節點;
                // 然後將新節點設置為啞頭結點的下一個節點,即新的鏈表頭部。
                if (firstRealNode != null)
                {
                    firstRealNode._previous = linkedSlot;
                }
                // 將新節點分配給slotArray 數組的指定位置,以便其他線程可以通過數組訪問到該節點。
                _linkedSlot._next = linkedSlot;
                slotArray[id].Value = linkedSlot;
            }
        }

我們發現.NET在內部管本地線程儲存時,使用多個鏈表,有著比較多的優勢,如:動態性、插入效率高、內存分配靈活、易於維護。但是也有這一些不足的地方,如果線程本地變量的數量相對固定且數量較大,且需要快速的隨機訪問操作,那麼使用數組可能更為合適。而如果需要高效的查找和插入操作,並且可以接受一定的空間開銷,那麼哈希表可能是更好的選擇。

GrowTable(ref LinkedSlotVolatile[] table, int minLength)實現了線程本地變量表的擴容操作,即在需要時將線程本地變量表的大小增加到一個更大的值。lock(s_idManager):在對舊表進行遷移時,使用鎖對其進行保護,以避免與 ThreadLocal.Dispose 方法的競爭條件。遷移舊表的內容到新表,遍歷舊表,將舊表中每個位置上的LinkedSlot 對象(如果存在)的 _slotArray 屬性指向新的表,並將該 LinkedSlot 對象復制到新表中相同的位置。

        private static void GrowTable(ref LinkedSlotVolatile[] table, int minLength)
        {
            int newLen = GetNewTableSize(minLength);
            LinkedSlotVolatile[] newTable = new LinkedSlotVolatile[newLen];
            lock (s_idManager)
            {
                for (int i = 0; i < table.Length; i  )
                {
                    LinkedSlot? linkedSlot = table[i].Value;
                    if (linkedSlot != null && linkedSlot._slotArray != null)
                    {
                        linkedSlot._slotArray = newTable;
                        newTable[i] = table[i];
                    }
                }
            }
            table = newTable;
        }

GetNewTableSize 方法用於確定下一個更大的表的大小。它首先將輸入值減一,然後通過位運算將 1 位向右傳播,直到找到比輸入值更大的下一個 2 的冪。最後,它將結果加一,以確保返回的值大於輸入值。

        private static int GetNewTableSize(int minSize)
        {
            if ((uint)minSize > Array.MaxLength){ return int.MaxValue; }
            int newSize = minSize;
            newSize--;
            newSize |= newSize >> 1;
            newSize |= newSize >> 2;
            newSize |= newSize >> 4;
            newSize |= newSize >> 8;
            newSize |= newSize >> 16;
            newSize  ;
            if ((uint)newSize > Array.MaxLength){ newSize = Array.MaxLength; }
            return newSize;
        }

可你會有同學問,為什麼采用這樣的擴容邏輯,使用這樣的計算邏輯能夠保證線程本地變量表的容量增長是高效、規律和穩定的,同時代碼實現也更加簡潔清晰。具體有以下這幾種優勢:

(1)、容量增長的規律性:通過找到比當前大小大且最接近的 2 的冪次方,可以保證表的容量增長是規律性的。有助於減少表的碎片化,使得內存的分配和管理更加高效。
(2)、位運算的效率:位運算的效率比較高,可以快速地找到比當前大小更大的 2 的冪次方。相比於一般的循環遞增方式,位運算的速度更快。
(3)、代碼簡潔性:使用位運算可以讓代碼更加簡潔清晰,避免了復雜的循環遞增邏輯。通過幾行簡單的位運算代碼,就能達到確定下一個更大的 2 的冪次方的目的。
(4)、算法穩定性:這種計算邏輯是一種經過驗證的算法,已經在實踐中得到了廣泛的應用和驗證。保證了線程本地變量表的容量增長是穩定和可預測的。在其他的語言中也有類似的邏輯,這樣的方式比較的通用和廣泛。例如:java 中的 HashMap 和 ArrayList、Python 中的 List 和 Dictionary、C   中的 std::vectorstd::unordered_map、JavaScript 中的 Array。

我們前面介紹了比較復雜的SetValueSlow()的內部實現,包含了對應的存儲結構和擴容機制等等,接下來我們在來看看相對要簡單一些的GetValueSlow()方法,該方法實現了在慢速路徑上獲取線程本地變量值的方法。

        private Func<T>? _valueFactory;
        private T? GetValueSlow()
        {
            // 將線程的ID計算出來。ID 的取反值在初始化時會被賦值給 
            // _idComplement,因此取反後的值就是線程的ID。
            int id = ~_idComplement;
            ObjectDisposedException.ThrowIf(id < 0, this);
            T value;
            // 確定線程本地變量的初始值
            if (_valueFactory == null)
            {
                value = default!;
            }
            else
            {
                value = _valueFactory();
            }
            Value = value;
            return value;
        }

這段代碼是在慢速路徑上獲取線程本地變量值的方法。它首先檢查線程是否已被銷毀,然後確定初始值,並將其設置為線程本地變量的值,最後返回獲取到的值。

本文截止到此已經借助一個樣例,逐步的打開ThreadLocal內部的實現機制,相信大傢已經對ThreadLocal已經有了一個全局的了解和掌握,接下來我們來看看本地線程儲存在實際應用的場景和使用建議。

1、應用場景:
    (1)、線程封閉數據:用於存儲線程封閉的數據,例如線程特定的配置信息、日志記錄器、數據庫連接等,這些數據對於每個線程都是唯一的。
    (2)、線程狀態管理:在多線程編程中,有時候需要存儲線程的狀態信息,而線程本地變量可以提供一種簡單有效的方式來管理線程狀態。
     (3)、線程上下文傳遞:有些情況下,需要在不同的線程間傳遞上下文信息,線程本地變量可以用於在調用鏈路中傳遞上下文,而不需要顯式地傳遞參數。
     (4)、性能優化:在需要頻繁訪問的數據上,可以使用線程本地變量來減少鎖競爭,提高性能。
2、使用建議:
      (1)、適度使用:線程本地變量在多線程編程中非常有用,但是應該適度使用,避免濫用。過多的線程本地變量可能會導致代碼可讀性和維護性下降。
      (2)、避免內存泄漏:需要註意及時釋放不再需要的線程本地變量,以防止內存泄漏問題。
      (3)、測試和調試:在使用線程本地變量時,要仔細測試和調試,確保線程之間的數據隔離和正確性。
      (4)、註意線程池和異步編程:在線程池和異步編程中,線程本地變量可能會導致意外的結果,需要格外小心處理。

本文如有錯漏之處,還望大傢多多指正。