理解線程同步

線程的數(shù)據(jù)訪問

在并行（多線程）環(huán)境中，不可避免地會存在多個線程同時訪問某個數(shù)據(jù)的情況。多個線程對共享數(shù)據(jù)的訪問有下面3種情形：

多個線程同時讀取數(shù)據(jù)；

單個線程更新數(shù)據(jù)，此時其他線程讀取數(shù)據(jù)；

多個線程同時更新數(shù)據(jù)。

顯而易見，多個線程同時讀取數(shù)據(jù)是不會產(chǎn)生任何問題的。僅有一個線程更新數(shù)據(jù)的時候，貌似也沒有問題，但真的沒有問題嗎？多個線程同時更新數(shù)據(jù)，很明顯，你可能把我的更改覆蓋掉了，數(shù)據(jù)從此不再可信。

什么是線程同步

為了解決多線程同時訪問共享數(shù)據(jù)可能導(dǎo)致數(shù)據(jù)被破壞的問題，我們需要采取一些措施來保證數(shù)據(jù)的一致性，讓每個線程都能準確地讀取或更新數(shù)據(jù)。

問題的根源在于多個線程同時訪問數(shù)據(jù)，那么只要我們保證同一時間只有一個線程訪問數(shù)據(jù)，就能解決問題。保證同一時間只有一個線程訪問數(shù)據(jù)的處理，就是線程同步了。我在訪問數(shù)據(jù)的時候，你們都先等著，我完事了你們再來。

C#中的線程同步

.NET提供了很多線程同步的方式，這些方式分為用戶模式和內(nèi)核模式以及混合模式（即用戶模式與內(nèi)核模式的結(jié)合），下面會總結(jié)C#/.NET中各模式下的線程同步。

用戶模式與內(nèi)核模式

Windows操作系統(tǒng)下，CPU跟據(jù)所執(zhí)行代碼的不同，會在兩種模式下進行切換。CPU執(zhí)行應(yīng)用程序代碼（如我們開發(fā)的.NET程序）時，一般運行在用戶模式下；執(zhí)行操作系統(tǒng)核心代碼（內(nèi)核函數(shù)或者某些設(shè)備驅(qū)動程序）時，CPU則切換到內(nèi)核模式。

用戶模式的代碼只能訪問自身進程的專有地址空間，代碼異常不會影響到其他程序或者操作系統(tǒng)；內(nèi)核模式的所有代碼共享單個地址空間，代碼異常將可能導(dǎo)致系統(tǒng)崩潰。CPU的模式切換，是為了保證應(yīng)用程序和操作系統(tǒng)的穩(wěn)定性。

應(yīng)用程序中，線程可以通過Windows?API調(diào)用操作系統(tǒng)內(nèi)核函數(shù)，這時候執(zhí)行線程的CPU將從用戶模式切換到內(nèi)核模式，執(zhí)行完操作系統(tǒng)函數(shù)后，再由內(nèi)核模式切換到用戶模式。CPU的模式切換是很耗時的，據(jù)《Windows核心編程》中的描述，CPU模式的切換，要占用1000個以上的CPU周期。因此，在我們的.NET程序中，應(yīng)該盡可能地避免CPU的模式切換。

用戶模式線程同步

用戶模式下，利用特殊的CPU指令來協(xié)調(diào)線程，使同一時間只有一個線程能訪問某內(nèi)存地址，這種協(xié)調(diào)在硬件中發(fā)生，速度很快。這種模式下，CPU指令對線程的阻塞很短暫，操作系統(tǒng)調(diào)度線程時不會認為該線程已被阻塞，這種情況下，線程池不會創(chuàng)建新的線程來替換該線程。

用戶模式下，等待資源的線程會一直被操作系統(tǒng)調(diào)度，導(dǎo)致線程的“自旋”并因此浪費很多的CPU資源。如果某線程一直占著資源不釋放，等待該資源的線程將一直處于自旋狀態(tài)，這樣就造成了“活鎖”，活鎖除了浪費內(nèi)存外，還會浪費大量CPU。

.NET提供兩種用戶模式的線程同步，volatile和interlocked，即易變和互鎖。

volatile關(guān)鍵字和Volatile

上面我們遺留了一個問題：只有一個線程更新數(shù)據(jù)，其他線程讀取數(shù)據(jù)，會不會出現(xiàn)問題？先看一個例子：

private static bool _stop;
public static void Run()
{
Task.Run(() =>
{
int number = 1;
while?(!_stop)?
{
number++;
}
Console.WriteLine($"increase stopped,value = {number}");
});

Thread.Sleep(1000);
_stop = true;
}

編譯器和CPU會對上面的代碼進行優(yōu)化（調(diào)試模式不會優(yōu)化），任務(wù)線程在執(zhí)行時，會把_stop讀取到CPU寄存器中，while循環(huán)的時候，每次都從當前CPU寄存器中讀取_stop；同樣，主線程執(zhí)行的時候CPU也會把_stop讀取到寄存器，更新_stop時，先更新是CPU寄存器中的_stop值，再把值存到變量_stop;在并行環(huán)境中，主線程和任務(wù)線程獨立執(zhí)行，主線程對_stop的更新并不會公開到任務(wù)線程，這樣，任務(wù)線程的while循環(huán)便不會停止，永遠無法得到輸出。

把變量讀到寄存器只是CPU優(yōu)化代碼的一種方式，CPU還可能調(diào)整代碼的執(zhí)行順序，當前，CPU任務(wù)這種調(diào)整不會改變代碼的意圖。上面的代碼說明，由于編譯器和CPU的優(yōu)化，只有一個線程更新數(shù)據(jù)，也可能存在問題。

這種情況，我們可以使用volatile關(guān)鍵字或者類System.Threading.Volatile來阻止編譯器和CPU的優(yōu)化，這種阻止利用的是內(nèi)存屏障MemoryBarrier，它告訴CPU在執(zhí)行完屏障之前的內(nèi)存存取后才能執(zhí)行屏障后面的內(nèi)存存取。上面代碼的問題在于，while循環(huán)讀取到的值總是CPU寄存器中的false。我們把while循環(huán)的條件改成!Volatile.Read(ref?_stop)或者把用volatile聲明變量_stop，while條件直接讀取內(nèi)存中的值，問題就能得到解決。

Interlocked原子訪問

.NET提供的另一種用戶模式線程同步方式是System.Threading.Interlocked。Interlocked的工作依賴于代碼運行的CPU平臺，如果是X86的CPU,Interlocked函數(shù)會在總線上維持一個硬件信號，來阻止其他CPU訪問同一內(nèi)存地址（《Windows核心編程第五版》）。計算機對變量的修改一般來說并不是原子性的，而是分為3個步驟：

將變量值加載到CPU寄存器

改變值

將更新后的值存儲到內(nèi)存中

假如執(zhí)行了前兩個步驟后，CPU被搶占，變量在之前線程中的修改將丟失。Interlocked函數(shù)保證對值的修改是原子性的，一個線程完成變量的修改和存儲后，另一個線程才能修改變量。

System.Threading.Interlocked提供了很多方法，例如遞增、遞減、求和等，下面用Interlocked的遞增方法展示其線程同步功能。

public static void Run()
{
DoIncrease(100000);
}

private static void DoIncrease(int incrementPerThread)
{
int number1 = 0;
int number2 = 0;

Console.WriteLine($"use two threads to increase zero. each thread increase {incrementPerThread}.");

IList<Task> increaseTasks = new List<Task>();

increaseTasks.Add(Task.Run(() =>
{
Console.WriteLine($"thread #{Thread.CurrentThread.ManagedThreadId} increasing number1.");
for (int i = 0; i < incrementPerThread; i++)
{
Interlocked.Increment(ref number1);
}
}));
increaseTasks.Add(Task.Run(() =>
{
Console.WriteLine($"thread #{Thread.CurrentThread.ManagedThreadId} increasing number1.");
for (int i = 0; i < incrementPerThread; i++)
{
Interlocked.Increment(ref number1);
}
}));
increaseTasks.Add(Task.Run(() =>
{
Console.WriteLine($"thread #{Thread.CurrentThread.ManagedThreadId} increasing number2.");
for (int i = 0; i < incrementPerThread; i++)
{
number2++;
}
}));
increaseTasks.Add(Task.Run(() =>
{
Console.WriteLine($"thread #{Thread.CurrentThread.ManagedThreadId} increasing number2.");
for (int i = 0; i < incrementPerThread; i++)
{
number2++;
}
}));

Task.WaitAll(increaseTasks.ToArray());

Console.WriteLine($"use?interlocked:?number1?result?=?{number1}");
Console.WriteLine($"normal?increase:?number2?result?=?{number2}");
}

運行上面的代碼多次（每個線程增加的數(shù)量盡量大，否則不容易體現(xiàn)結(jié)果），每次number1的結(jié)果都一樣，number2的結(jié)果都不同，足以體現(xiàn)Interlocked的線程同步功能。

SpinLock自旋鎖

System.Threading.SpinLock是基于InterLocked和SpinWait實現(xiàn)的輕量級自旋鎖，具體的實現(xiàn)方式這里不去關(guān)心。SpinLock的簡單用法如下：

private static SpinLock _spinlock = new SpinLock();
public static void DoWork()
{
bool lockTaken = false;
try
{
_spinlock.Enter(ref lockTaken);

}
finally
{
if (lockTaken)
{
_spinlock.Exit(false);
}
}
}

SpinLock很輕量級，性能較高，但由于是自旋鎖，鎖定的操作應(yīng)該是很快完成，否則會因線程自旋而浪費CPU。

內(nèi)核模式線程同步

除了用戶模式的兩種線程同步方式，我們還會利用Windows系統(tǒng)的內(nèi)核對象實現(xiàn)線程的同步。使用系統(tǒng)內(nèi)核對象將會導(dǎo)致執(zhí)行線程的CPU運行模式的切換，這會有很大的消耗，所以能夠使用用戶模式的線程同步就盡量避免使用內(nèi)核模式。

內(nèi)核模式下，線程在等待資源時會被系統(tǒng)阻塞，避免了CPU的浪費，這是內(nèi)核模式優(yōu)勢。假如線程等待的資源一直被占用則線程將一直處于阻塞狀態(tài)，造成“死鎖”。相對于活鎖，死鎖只會浪費內(nèi)存資源。

我們使用系統(tǒng)內(nèi)核中的事件、信號量和互斥量進行內(nèi)核模式的線程同步。

利用內(nèi)核事件實現(xiàn)線程同步

事件實際上是由系統(tǒng)內(nèi)核維護的一個布爾值。

.NET提供System.Threading.EventWaitHandle進行線程的信號交互。EventWaitHandle繼承WaitHandle（封裝等待對共享資源獨占訪問的操作系統(tǒng)特定的對象），有三個關(guān)鍵方法：

• Set():將事件狀態(tài)設(shè)置為終止狀態(tài)，允許一個或多個等待線程繼續(xù)。

• Reset():將事件狀態(tài)設(shè)置為非終止狀態(tài)，導(dǎo)致線程阻塞

• WaitOne():阻塞線程直到收到事件狀態(tài)信號

線程交互事件有自動重置和手動重置兩種類型，分別由AutoResetEvent和ManualResetEvent繼承EventWaitHandle得到。自動重置事件在Set喚醒第一個阻塞線程之后，會自動Reset事件，其他阻塞線程仍保持阻塞狀態(tài)；而手動重置事件Set時，會喚醒所有被該事件阻塞的線程，手動Reset后，事件才會繼續(xù)起作用。手動重置事件的這種性質(zhì)，導(dǎo)致它不能用于線程同步，因為不能保證同一時間只有一個線程訪問資源;相反，自動重置時間則很適合用來處理線程同步。

下面的例子演示了利用自動重置時間進行的線程同步。

public static void Run()
{
DoIncrease(100000);
}

private static void DoIncrease(int incrementPerThread)
{
int number = 0;
Console.WriteLine($"use two threads to increase zero. each thread increase {incrementPerThread}.");

AutoResetEvent are = new AutoResetEvent(true);

IList<Task> increaseTasks = new List<Task>();
increaseTasks.Add(Task.Run(() =>
{
Console.WriteLine($"thread #{Thread.CurrentThread.ManagedThreadId} is increasing the number.");
for (int i = 0; i < incrementPerThread; i++)
{
are.WaitOne();
number++;
are.Set();
}
}));
increaseTasks.Add(Task.Run(() =>
{
Console.WriteLine($"thread #{Thread.CurrentThread.ManagedThreadId} is increasing the number.");
for (int i = 0; i < incrementPerThread; i++)
{
are.WaitOne();
number++;
are.Set();
}
}));

Task.WaitAll(increaseTasks.ToArray());
are.Dispose();
Console.WriteLine($"use?AutoResetEvent:?result?=?{number}");
}

利用信號量進行線程同步

信號量是系統(tǒng)內(nèi)核維護的一個整型變量。

信號量值為0時，所有等待信號量的線程會被阻塞；信號量值大于零0，等待的線程會被解除阻塞，每喚醒一個阻塞的線程，系統(tǒng)內(nèi)核就會把信號量的值減1。此外，我們能夠?qū)π盘柫窟M行最大值限制，從而控制訪問同一資源的最大線程數(shù)量。

.Net中，利用System.Threading.Semaphore進行信號量操作。下面時利用信號量實現(xiàn)線程同步的一個例子。

public static void Run()
{
DoIncrease(100000);
}

private static void DoIncrease(int incrementPerThread)
{
int number = 0;
Console.WriteLine($"use two threads to increase zero. each thread increase {incrementPerThread}.");

Semaphore semaphore = new Semaphore(1,1);

IList<Task> increaseTasks = new List<Task>();
increaseTasks.Add(Task.Run(() =>
{
Console.WriteLine($"thread #{Thread.CurrentThread.ManagedThreadId} is increasing the number.");
for (int i = 0; i < incrementPerThread; i++)
{
semaphore.WaitOne();
number++;
semaphore.Release(1);

}
}));
increaseTasks.Add(Task.Run(() =>
{
Console.WriteLine($"thread #{Thread.CurrentThread.ManagedThreadId} is increasing the number.");
for (int i = 0; i < incrementPerThread; i++)
{
semaphore.WaitOne();
number++;
semaphore.Release(1);

}
}));

Task.WaitAll(increaseTasks.ToArray());
semaphore.Dispose();
Console.WriteLine($"use?Semaphore:?result?=?{number}");
}

利用互斥體進程線程同步

互斥體Mutex的使用與自動重置事件和信號量類似，這里不再進行詳細的總結(jié)。

互斥體常被用來保證應(yīng)用程序只有一個實例運行，具體用法如下：

bool createNew;
using (new Mutex(true, Assembly.GetExecutingAssembly().FullName, out createNew))
{
if?(!createNew)
{

Environment.Exit(0);
}
else
{

}
}

線程同步的混合模式

通過上面的總結(jié)我們知道，用戶模式和內(nèi)核模式由各自的優(yōu)缺點，需要有一種模式既能兼顧用戶和內(nèi)核模式的優(yōu)點又能避免他們的缺點，這就是混合模式。

混合模式會優(yōu)先使用用戶模式的線程同步處理，當多個線程競爭同步鎖的時候，才會使用內(nèi)核對象進行處理。如果多個線程一直不產(chǎn)生資源競爭，就不會發(fā)生CPU用戶模式到內(nèi)核模式的轉(zhuǎn)換，開始資源競爭時，又會通過線程阻塞來防止CPU資源的浪費。

.NET中提供了多種混合模式的線程同步方式。例如手工重置事件和信號量的簡化版本ManualResetEventSlim及SemaphoreSlim，他們是線程在用戶模式中自旋，直到發(fā)生資源競爭。具體使用與各自的內(nèi)核模式一樣，這里不再贅述。

lock關(guān)鍵字和Monitor

相信lock加鎖是很多人做常用的線程同步方式。lock的使用很簡單，如下：

private static readonly object _syncObject = new object();
public static void DoWork()
{
lock (_syncObject)
{

}
}

實際上，lock語法是對System.Threading.Monitor使用的一種簡化，Monitor的用法如下：

private static readonly object _syncObject = new object();
public static void DoWork()
{
Monitor.Enter(_syncObject);

Monitor.Exit(_syncObject);
}

使用Monitor的可能會出先一些意象不到的問題。例如，如果不相關(guān)的業(yè)務(wù)代碼在使用Monitor進行線程同步的時候，鎖定了同一字符串，將會造成不相關(guān)業(yè)務(wù)代碼的同步執(zhí)行；此外需要注意的是，Monitor不能使用值類型作為鎖對象，值類型會被裝箱，裝箱后的對象不同，將導(dǎo)致無法同步。

讀寫鎖ReaderWriterLockSlim

ReaderWriterLockSlim可以用來實現(xiàn)多線程讀取或獨占寫入的資源訪問。讀寫鎖的線程控制邏輯如下：

• 一個線程寫數(shù)據(jù)時，其他請求資源的線程全部被阻塞；

• 一個線程讀數(shù)據(jù)時，寫線程被阻塞，其他讀線程能繼續(xù)運行；

• 寫結(jié)束時，解除其他某個寫線程的阻塞，或者解除所有讀線程的阻塞；

• 讀結(jié)束時，解除一個寫線程的阻塞。

下面是讀寫鎖的簡單用法，詳細用法可參考msdn文檔。

private static readonly ReaderWriterLockSlim _rwlock = new ReaderWriterLockSlim();
public static void DoWork()
{
_rwlock.EnterWriteLock();

_rwlock.ExitWriteLock();
}

ReaderWriterLockSlim還有一個比較老的版本ReaderWriterLock，據(jù)說存在較多問題應(yīng)盡量避免使用。

線程安全集合

.NET除了提供包含上面總結(jié)到的各種線程同步的諸多方式外，還封裝了一些線程安全集合。這些集合在內(nèi)部實現(xiàn)了線程同步，我們直接使用即可，很友好。線程安全集合在命名空間System.Collections.Concurrent下，包括ConcurrentQueue (T),ConcurrentStack<T>,ConcurrentDictionary<TKey,TValue>,ConcurrentBag<T>,BlockingCollection<T>，具體可閱讀《何時使用線程安全集合》。

各種線程同步性能對比

下面我們對整數(shù)零進行多線程遞增操作，每個線程固定遞增量，來測試以下各種同步方式的性能對比。測試代碼如下。

private static int _numberToIncrease;

public static void Run()
{
int increment = 100000;
int threadCount = 4;
DoIncrease(increment, threadCount, DoIncreaseByInterLocked);
DoIncrease(increment, threadCount, DoIncreaseWithSpinLock);
DoIncrease(increment, threadCount, DoIncreaseWithEvent);
DoIncrease(increment, threadCount, DoIncreaseWithSemaphore);
DoIncrease(increment, threadCount, DoIncreaseWithMonitor);
DoIncrease(increment, threadCount, DoIncreaseWithReaderWriterLockSlim);

}

public static void DoIncrease(int increment, int threadCount, Action<int> action)
{
_numberToIncrease = 0;
IList<Task> increaseTasks = new List<Task>(threadCount);
Stopwatch watch = Stopwatch.StartNew();
for (int i = 0; i < threadCount; i++)
{
increaseTasks.Add(Task.Run(() => action(increment)));
}
Task.WaitAll(increaseTasks.ToArray());
Console.WriteLine($"{action.Method.Name}=>?Result:?{_numberToIncrease}?,?Time:?{watch.ElapsedMilliseconds} ms.");
}

#region 使用Interlocked，用戶模式

public static void DoIncreaseByInterLocked(int increment)
{
for (int i = 0; i < increment; i++)
{
Interlocked.Increment(ref _numberToIncrease);
}
}

#endregion

#region 使用SpinLock,用戶模式

private static SpinLock _spinlock = new SpinLock();
public static void DoIncreaseWithSpinLock(int increment)
{
for (int i = 0; i < increment; i++)
{
bool lockTaken = false;
try
{
_spinlock.Enter(ref lockTaken);
_numberToIncrease++;
}
finally
{
if (lockTaken)
{
_spinlock.Exit(false);
}
}
}
}

#endregion

#region 使用信號量Semaphore，內(nèi)核模式

private static readonly Semaphore _semaphore = new Semaphore(1, 10);

public static void DoIncreaseWithSemaphore(int increment)
{
for (int i = 0; i < increment; i++)
{
_semaphore.WaitOne();
_numberToIncrease++;
_semaphore.Release(1);
}
}

#endregion

#region 使用事件AutoResetEvent，內(nèi)核模式

private static readonly AutoResetEvent _are = new AutoResetEvent(true);
public static void DoIncreaseWithEvent(int increment)
{
for (int i = 0; i < increment; i++)
{
_are.WaitOne();
_numberToIncrease++;
_are.Set();
}
}

#endregion

#region 使用Monitor，混合模式

private static readonly object _monitorLocker = new object();
public static void DoIncreaseWithMonitor(int increment)
{
for (int i = 0; i < increment; i++)
{
bool lockTaken = false;
try
{
Monitor.Enter(_monitorLocker, ref lockTaken);
_numberToIncrease++;
}
finally
{
if (lockTaken)
{
Monitor.Exit(_monitorLocker);
}
}
}
}

#endregion

#region 使用ReaderWriterLockSlim，混合模式

private static readonly ReaderWriterLockSlim _rwlock = new ReaderWriterLockSlim();
public static void DoIncreaseWithReaderWriterLockSlim(int increment)
{
for (int i = 0; i < increment; i++)
{
_rwlock.EnterWriteLock();
_numberToIncrease++;
_rwlock.ExitWriteLock();
}
}

#endregion

下面是一組測試結(jié)果，可以很明顯地看出，內(nèi)核模式是相當耗時的，應(yīng)盡量避免使用。而用戶模式和混合模式，也需要根據(jù)具體的場景進行選擇。這個測試過于簡單，不具有普遍性。

DoIncreaseByInterLocked=>?Result:?400000?,?Time:?15?ms.
DoIncreaseWithSpinLock=>?Result:?400000?,?Time:?75?ms.
DoIncreaseWithEvent=>?Result:?400000?,?Time:?1892?ms.
DoIncreaseWithSemaphore=>?Result:?400000?,?Time:?1779?ms.
DoIncreaseWithMonitor=>?Result:?400000?,?Time:?14?ms.
DoIncreaseWithReaderWriterLockSlim=>?Result:?400000?,?Time:?22?ms.

小結(jié)

本文對C#/.NET中的線程同步進行了盡量詳盡的總結(jié)，并行環(huán)境中在追求程序的高性能、響應(yīng)性的同時，務(wù)必要保證數(shù)據(jù)的安全性。

C#并行編程系列的文章暫時就告一段落了。剛開始寫博客，文章肯定存在不少問題，歡迎各位博友指出。

原文地址：https://www.cnblogs.com/chenbaoshun/p/10695343.html

.NET社區(qū)新聞，深度好文，歡迎訪問公眾號文章匯總?http://www.csharpkit.com?

總結(jié)

以上是生活随笔為你收集整理的C#并行编程（6）：线程同步面面观的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。