前言 拋開死鎖不談,只聊性能問題,儘管鎖總能粗暴的滿足同步需求,但一旦存在競爭關係,意味著一定會有線程被阻塞,競爭越激烈,被阻塞的線程越多,上下文切換次數越多,調度成本越大,顯然在高併發的場景下會損害性能。在高併發高性能且要求線程安全的述求下,無鎖構造(非阻塞構造)閃亮登場。 參考文檔: C# - ...
前言
拋開死鎖不談,只聊性能問題,儘管鎖總能粗暴的滿足同步需求,但一旦存在競爭關係,意味著一定會有線程被阻塞,競爭越激烈,被阻塞的線程越多,上下文切換次數越多,調度成本越大,顯然在高併發的場景下會損害性能。在高併發高性能且要求線程安全的述求下,無鎖構造(非阻塞構造)閃亮登場。
參考文檔:
C# - 理論與實踐中的 C# 記憶體模型,第 2 部分 | Microsoft Docs
一、非阻塞同步
重排序與緩存
我們觀察下麵這個例子:
public class Foo
{
private int _answer;
private bool _complete;
void A() //A 1
{
_answer = 10;
_complete = true;
}
void B() //B 2
{
if (_complete) Console.WriteLine(_answer);
}
}
如果方法A
和B
在不同的線程上併發運行,B
可能會列印 “ 0 “ 嗎?答案是會的,原因如下:
- 編譯器、CLR 或 CPU 可能會對代碼/指令進行重排序(reorder)以提高效率。
- 編譯器、CLR 或 CPU 可能會進行緩存優化,導致其它線程不能馬上看到變數的新值。
請務必重視它們,它們將是幽靈般的存在
int x = 0, y = 0, a = 0, b = 0;
var task1 = Task.Run(() => // A 1
{
a = 1; // 1
x = b; // 2
});
var task2 = Task.Run(() => // B 2
{
b = 2; // 3
y = a; // 4
});
Task.WaitAll(task1, task2);
Console.WriteLine("x:" + x + " y:" + y);
直覺和經驗告訴我們,程式至頂向下執行:代碼1一定發生在代碼2之前,代碼3一定發生在代碼4之前,然鵝
在一個獨立的線程中,每一個語句的執行順序是可以被保證的,但在不使用lock,waithandle這樣的顯式同步操作時,我們就沒法保證事件在不同的線程中看到的執行順序是一致的了。儘管線程A中一定需要觀察到a=1執行成功之後才會去執行x=b,但它沒法確保自己觀察得到線程B中對b的寫入,所以A還可能會列印出y的一個舊版的值。這就叫指令重排序。
x:0 y:1 #1-2-3-4
x:2 y:0 #3-4-1-2
x:2 y:1 #1-3-2-4
可實際運行時還是有些讓我們驚訝的情況:
x:0 y:0 #??
這就是緩存問題,如果兩個線程在不同的CPU上執行,每一個核心有自己的緩存,這樣一個線程的寫入對於其它線程,在主存同步之前就是不可見的了。
C#編譯器和CLR運行時會非常小心的保證上述優化不會破壞普通的單線程代碼,和正確使用鎖的多線程代碼。但有時,你仍然需要通過顯示的創建記憶體屏障(memory barrier,也稱作記憶體柵欄 (memory fence))來對抗這些優化,限制指令重排序和讀寫緩存產生的影響。
記憶體屏障
處理器支持哪種記憶體重排序(LoadLoad重排序、LoadStore重排序、StoreStore重排序、StoreLoad重排序),就會提供相對應能夠禁止重排序的指令,而這些指令就被稱之為記憶體屏障(LoadLoad屏障、LoadStore屏障、StoreStore屏障、StoreLoad屏障)
屏障名稱 | 示例 | 具體作用 |
---|---|---|
StoreLoad | Store1;Store2;Store3;StoreLoad;Load1;Load2;Load3 | 禁止StoreLoad重排序,確保屏障之前任何一個寫(如Store2)的結果都會在屏障後任意一個讀操作(如Load1)載入之前被寫入 |
StoreStore | Store1;Store2;Store3;StoreStore;Store4;Store5;Store6 | 禁止StoreStore重排序,確保屏障之前任何一個寫(如Store1)的結果都會在屏障後任意一個寫操作(如Store4)之前被寫入 |
LoadLoad | Load1;Load2;Load3;LoadLoad;Load4;Load5;Load6 | 禁止LoadLoad重排序,確保屏障之前任何一個讀(如Load1)的數據都會在屏障後任意一個讀操作(如Load4)之前被載入 |
LoadStore | Load1;Load2;Load3;LoadStore;Store1;Store2;Store3 | 禁止LoadStore重排序,確保屏障之前任何一個讀(如Load1)的數據都會在屏障後任意一個寫操作(如Store1)的結果被寫入高速緩存(或主記憶體)前被載入 |
讀屏障告訴處理器在執行任何的載入前,執行所有已經在失效隊列(Invalidte Queues)中的失效(I)指令。即:所有load barrier之前的store指令對之後(本核心和其他核心)的指令都是可見的。
Store Memory Barrier:寫屏障,等同於前文的StoreStore Barriers 將store buffer都寫入緩存。
寫屏障告訴處理器在執行這之後的指令之前,執行所有已經在存儲緩存(store buffer)中的修改(M)指令。即:所有store barrier之前的修改(M)指令都是對之後的指令可見。
最簡單的記憶體屏障是完全記憶體屏障(full memory barrier,或全柵欄(full fence)),它可以阻止所有跨越柵欄的指令進行重排並提交修改和刷新緩存
。記憶體屏障之前的所有寫操作都要寫入記憶體,並將記憶體中的新值刷到緩存,使得其它CPU核心能夠讀取到最新值,完全保證了數據的強一致性,進而解決CPU緩存帶來的可見性問題。
我們簡單修改一下前面的案例
void A()
{
_answer = 10;
Thread.MemoryBarrier(); // 1
_complete = true;
Thread.MemoryBarrier(); // 3
}
void B()
{
Thread.MemoryBarrier(); // 2
if (_complete)
{
_testOutputHelper.WriteLine(_answer.ToString());
}
}
屏障1,3使得這個例子不可能列印出0,屏障2保證如果B在A之後執行,_complete一定讀到的是true
記憶體屏障離我們並不遙遠,以下方式都會隱式的使用全柵欄:
-
lock語法糖或
Monitor.Enter
/Monitor.Exit
-
Interlocked
類中的所有方法 -
使用線程池的非同步回調,包括非同步委托,APM回調,以及任務延續(task continuations)
-
信號構造的等待/複位
-
任何依賴信號同步的情況,比如啟動或等待Task,因此下麵的代碼也是線程安全的
int x = 0; Task t = Task.Factory.StartNew (() => x++); t.Wait(); Console.WriteLine (x); // 1
volatile
另一個(更高級的)解決這個問題的方法是對_complete
欄位使用volatile
關鍵字。
volatile bool _complete;
volatile
關鍵字通知編譯器在每個讀這個欄位的地方使用一個讀柵欄(acquire-fence),並且在每個寫這個欄位的地方使用一個寫柵欄(release-fence)。
這種“半柵欄(half-fences)”比全柵欄更快,因為它給了運行時和硬體更大的優化空間。
讀柵欄:也就是讀屏障(Store Memory Barrier),等同於前文的LoadLoad Barriers 將Invalidate的 都執行完成。告訴處理器在執行任何的載入前,執行所有已經在失效隊列(Invalidte Queues)中的失效(I)指令。即:所有load barrier之前的store指令對之後(本核心和其他核心)的指令都是可見的。
寫柵欄:也就是寫屏障(Store Memory Barrier),等同於前文的StoreStore Barriers 將store buffer都寫入主存。
告訴處理器在執行這之後的指令之前,執行所有已經在存儲緩存(store buffer)中的修改(M)指令。即:所有store barrier之前的修改(M)指令都是對之後的指令可見。
巧的是,Intel 的 X86 和 X64 處理器總是在讀時使用讀柵欄,寫時使用寫柵欄,無論是否使用
volatile
關鍵字。所以在使用這些處理器的情況下,這個關鍵字對硬體來說是無效的。然而,volatile
關鍵字對編譯器和 CLR 進行的優化是有作用的,以及在 64 位 AMD 和 Itanium 處理器上也是有作用的。這意味著不能因為你的客戶端運行在特定類型的 CPU 上而放鬆警惕。
註意:使用volatile
不能阻止寫-讀被交換
第一條指令 | 第二條指令 | 是否會被交換 |
---|---|---|
讀 | 讀 | 不會 |
讀 | 寫 | 不會 |
寫 | 寫 | 不會(CLR 確保寫-寫操作永遠不會被交換,就算是沒有volatile 關鍵字) |
寫 | 讀 | 會! |
在下麵案例中仍然有可能會列印00的情況(對a的讀取可能發生在寫入前--重排序)
int a = 0, b = 0;
int x = 0, y = 0;
var task1 = Task.Run(() =>
{
Thread.VolatileWrite(ref a, 1);
x = Thread.VolatileRead(ref b);
});
var task2 = Task.Run(() =>
{
Thread.VolatileWrite(ref b, 2);
y = Thread.VolatileRead(ref a);
});
Task.WaitAll(task1, task2);
Console.WriteLine("x:" + x + " y:" + y);
volatile
關鍵字不能應用於數組元素,不能用在捕獲的局部變數:這些情況下你必須使用VolatileRead
和VolatileWrite
方法
從上面的例子我們可以看出,寫-讀操作可能被重新排序,官方的解釋是:
在多處理器系統上,易失性讀取操作不保證獲取由任何處理器寫入該記憶體位置的最新值。 同樣,易失性寫入操作不保證寫入的值會立即對其他處理器可見。
(我的理解是:
volatile
關鍵字只能解決重排序問題,解決不了多處理器的緩存一致性問題)
註意double
和 long
無法標記為 volatile
,因為對這些類型的欄位的讀取和寫入不能保證是原子的。 若要保護對這些類型欄位的多線程訪問,請使用 Interlocked 類成員或使用 lock
語句保護訪問許可權。
Interlocked
位於System.Threading
,為多個線程共用的變數提供原子操作,這也是DOTNET為數不多的線程安全類型之一。
Interlocked
通過將原子性的需求傳達給操作系統和CLR來進行實現其功能,此類的成員不會引發異常。
可以防止 1.線程上下文切換,2.線程更新可由其他線程訪問的變數時,或者當兩個線程同時在不同的處理器上執行時 可能會出現的錯誤。
場景:
int i = 0;
i ++;
在大多數電腦上,自增並不是原子操作,需要以下步驟:
- 將變數
i
的值載入到寄存器中。 - 計算
i + 1
。 - 將上面的計算結果存儲在變數
i
中。
假設A線程執行完1-2時被搶占,B線程執行1-2-3,當A線程恢復時繼續執行3,此時B線程的值就被覆蓋掉了。
使用Increment
即可解決,123會被打包成一個操作,以原子的方式實現自增
CAS
CAS 操作包含三個操作數 —— 記憶體位置(V)、預期原值(A)和新值(B)。如果記憶體位置的值與預期原值相匹配,那麼處理器會自動將該位置值更新為新值。否則,處理器不做任何操作。無論哪種情況,它都會在 CAS 指令之前返回該位置的值。(在 CAS 的一些特殊情況下將僅返回 CAS 是否成功,而不提取當前值。)CAS 有效地說明瞭“我認為位置 V 應該包含值 A;如果包含該值,則將 B 放到這個位置;否則,不要更改該位置,只告訴我這個位置的值即可。”
Interlocked.CompareExchange,實現了CAS:比較兩個值是否相等,如果相等,則替換第一個值,否則什麼都不做,最終返回這個位置的原始值。
Interlocked.CompareExchange(ref _num, 1000, 500);
CAS在保證原子性讀寫的同時,沒有加鎖,保障了程式併發度,但也存在缺陷:
- ABA問題
- 只能保證一個地址的讀寫原子性
- 自旋CAS時間過長,容易給CPU帶來大開銷
二、延遲初始化
面試時候經常問:單例模式中的懶漢模式線程安全問題
場景:某個欄位構造開銷非常大,使得在初始化A
時需要承擔初始化Expensive
的開銷,即使Expensive欄位不會被用到。
public class A
{
public readonly Expensive Expensive = new Expensive();
// ..
}
public class Expensive
{
// 構造開銷非常昂貴
}
自然會想到懶漢模式:按需載入
public class B
{
private Expensive _expensive;
public Expensive GetExpensiveInstance()
{
if (_expensive == null) _expensive = new Expensive();
return _expensive;
}
}
新的問題產生:GetExpensiveInstance
是線程安全的嗎?我們可以通過加鎖解決
public class C
{
private readonly object _locker = new object();
private Expensive _expensive;
public Expensive GetExpensiveInstance()
{
lock (_locker)
{
if (_expensive == null) _expensive = new Expensive();
return _expensive;
}
}
}
現在面試官繼續問:還有性能更好的版本嗎?..
Lazy
net standard1.0 提供System.Lazy<T>
來幫助你以線程安全且高效的方式(DCL)解決延遲初始化問題,只需
public class D
{
private Lazy<Expensive> _expensive = new Lazy<Expensive>(() => new Expensive(), true);
public Expensive GetExpensiveInstance() => _expensive.Value;
}
第一個參數是一個委托,告知如何構建,第二個參數是boolean類型,傳false
實現的就是上面提到的plain B
非線程安全遲初始化
雙檢鎖 double checked locking會進行一次額外的易失讀(volatile read),在對象已經完成初始化時,能夠避免獲取鎖產生的開銷。
public class E
{
private readonly object _locker = new object();
private volatile Expensive _expensive;
public Expensive GetExpensiveInstance()
{
// 額外的易失讀(volatile read)
if (_expensive == null)
{
lock (_locker)
{
if (_expensive == null) _expensive = new Expensive();
}
}
return _expensive;
}
}
LazyInitializer
LazyInitializer
是一個靜態類,提供EnsureInitialized
方法,第一個參數是需要構造的變數地址,第二個參數是一個委托,告知如何構造
public class F
{
private Expensive _expensive;
public Expensive GetExpensiveInstance()
{
LazyInitializer.EnsureInitialized(ref _expensive,
() => new Expensive());
return _expensive;
}
}
它使用競爭初始化模式的實現,比雙檢鎖更快(在多核心情況下),因為它的實現完全不使用鎖。這是一個很少需要用到的極端優化,並且會帶來以下代價:
- 當參與初始化的線程數大於核心數時,它會更慢。
- 可能會因為進行了多餘的初始化而浪費 CPU 資源。
- 初始化邏輯必須是線程安全的(例如,
Expensive
的構造器對靜態欄位進行寫,就不是線程安全的)。 - 如果初始化的對象是需要進行銷毀的,多餘的對象需要額外的邏輯才能被銷毀。
競爭初始化(race-to-initialize)模式,通過易失性和CAS,實現無鎖構造
public class G
{
private volatile Expensive _expensive;
public Expensive Expensive
{
get
{
if (_expensive == null)
{
var instance = new Expensive();
Interlocked.CompareExchange (ref _expensive, instance, null);
}
return _expensive;
}
}
}
三、線程局部存儲
我們花費了大量篇幅來講併發訪問公共數據問題,前文提到的鎖構造,信號構造,無鎖構造本質上都是使用同步構造,使得多線程在訪問公共數據時能安全的進行,然而有時我們會希望數據線上程間是隔離的,局部變數就能實現這個目的,但他們的生命周期總是那麼短暫(隨代碼塊而釋放),我們期待更大作用域的隔離數據,線程局部變數(thread-local storage,TLS)就可以實現這個目的。
ThreadStatic
被ThreadStatic標記的static欄位不會線上程間共用,每個執行線程都有一個單獨的欄位實例
Note:
- 被標記的必須是static欄位,不能在實例欄位上使用(添加了也無效)
- 請不要給被標記的欄位指定初始值,因為這種初始化只會在類被構造時執行一次,影響一個線程,因此他依賴零值
如果你需要使用實例欄位,或者非零值,請使用ThreadLocal<T>
public class ThreadStatic測試
{
private readonly ITestOutputHelper _testOutputHelper;
[ThreadStatic] private static int _num;
public ThreadStatic測試(ITestOutputHelper testOutputHelper)
{
_testOutputHelper = testOutputHelper;
}
[Fact]
void Show()
{
void Work()
{
for (int i = 0; i < 100000; i++)
{
_num++;
_testOutputHelper.WriteLine(_num.ToString());
}
}
var t1 = new Thread(Work);
var t2 = new Thread(Work);
t1.Start();
t2.Start();
t1.Join();
t2.Join();
_testOutputHelper.WriteLine(_num.ToString());
}
}
輸出:
100000
100000
0
LocalDataStoreSlot
封裝記憶體槽以存儲本地數據。 此類不能被繼承。.NET Framework 1.1加入,但在standard2.0+才有。
public sealed class LocalDataStoreSlot
.NET Framework 提供了兩種機制,用於使用線程本地存儲 (TLS) :LocalDataStoreSlot
和ThreadStaticAttribute
LocalDataStoreSlot
比ThreadStaticAttribute
更慢,更尷尬。此外,數據存儲為類型 Object
,因此必須先將其強制轉換為正確的類型,然後再使用它。
有關使用 TLS 的詳細信息,請參閱 線程本地存儲。
同樣,.NET Framework 提供了兩種使用上下文本地存儲的機制:LocalDataStoreSlot
和ContextStaticAttribute
。 上下文相對靜態欄位是用屬性標記的 ContextStaticAttribute 靜態欄位。 請參考註解
// 同一個 LocalDataStoreSlot 對象可以跨線程使用。
LocalDataStoreSlot _slot = Thread.AllocateNamedDataSlot("mySlot");
void Work()
{
for (int i = 0; i < 100000; i++)
{
int num = (int)(Thread.GetData(_slot)??0);
Thread.SetData(_slot, num + 1);
}
_testOutputHelper.WriteLine(((int)(Thread.GetData(_slot)??0)).ToString());
}
var t1 = new Thread(Work);
var t2 = new Thread(Work);
t1.Start();
t2.Start();
t1.Join();
t2.Join();
_testOutputHelper.WriteLine(((int)(Thread.GetData(_slot)??0)).ToString());
輸出效果和ThreadStaticAttribute
一樣:
100000
100000
0
使用Thread.FreeNamedDataSlot("mySlot");
可以釋放所有線程上的指定槽,但是只有在所有對該槽的引用都出了其作用域,並且被垃圾回收後才會真正釋放。這確保了只要保持對槽的引用,就能繼續使用槽。
你也可以通過Thread.AllocateDataSlot()
來創建一個無名槽位,與命名槽的區別是無名槽需要自行控製作用域
當然我們也可以對上面複雜的᠍᠍᠍᠍᠍Thread.GetData
,Thread.SetData
進行封裝
LocalDataStoreSlot _secSlot = Thread.GetNamedDataSlot ("securityLevel");
int Num
{
get
{
object data = Thread.GetData(_secSlot);
return data == null ? 0 : (int) data; // null 相當於未初始化。
}
set { Thread.SetData (_secSlot, value); }
}
ThreadLocal
ThreadLocal<T>
是 Framework 4.0 加入的,涵蓋在netstandard1.0。它提供了可用於靜態欄位和實例欄位的線程局部存儲,並且允許設置預設值。
public class ThreadLocal測試
{
ThreadLocal<int> _num = new ThreadLocal<int> (() => 3);
private readonly ITestOutputHelper _testOutputHelper;
public ThreadLocal測試(ITestOutputHelper testOutputHelper)
{
_testOutputHelper = testOutputHelper;
}
[Fact]
void Show()
{
void Work()
{
for (int i = 0; i < 100000; i++)
{
_num.Value++;
}
_testOutputHelper.WriteLine(_num.ToString());
}
var t1 = new Thread(Work);
var t2 = new Thread(Work);
t1.Start();
t2.Start();
t1.Join();
t2.Join();
_testOutputHelper.WriteLine(_num.ToString());
}
}
輸出
100003
100003
3
下麵這個測試非常有意思
[Fact]
void Show()
{
var threadName = new ThreadLocal<string>(() => "Thread" + Thread.CurrentThread.ManagedThreadId);
Parallel.For(0, 13, x =>
{
bool repeat = threadName.IsValueCreated;
_testOutputHelper.WriteLine($"ThreadName = {threadName.Value} {(repeat ? "(repeat)" : "")}");
});
threadName.Dispose(); // 釋放資源
}
你會發現當Parallel.For第二個參數超過你的邏輯內核後,repeat出現了!
ThreadName = Thread5
ThreadName = Thread8
ThreadName = Thread31
ThreadName = Thread29
ThreadName = Thread31 (repeat)
ThreadName = Thread30
ThreadName = Thread18
ThreadName = Thread12
ThreadName = Thread32
ThreadName = Thread28
ThreadName = Thread33
ThreadName = Thread35
ThreadName = Thread34
Random
類不是線程安全的,所以我們要不然在使用Random
時加鎖(這樣限制了併發),如今我們有了ThreadLocal:
var localRandom = new ThreadLocal<Random>(() => new Random());
很輕易的就解決了線程安全問題,但是上面的版本使用的Random
的無參構造方法,會依賴系統時間作為生成隨機數的種子,在大概 10ms 時間內創建的兩個Random
對象可能會使用相同的種子,下邊是解決這個問題的一個辦法:
var localRandom = new ThreadLocal<Random>(() => new Random (Guid.NewGuid().GetHashCode()) );
特別註意,不要以為GUID全局唯一,GUID的HashCode也全局唯一,上面的隨機數仍然不是真隨機