并發(fā)編程 - 異步 vs. 多線程代碼

并行編程是一個廣泛的術(shù)語，我們應(yīng)該通過觀察異步方法和實際的多線程之間的差異展開探討。 盡管 .NET Core 使用了任務(wù)來表達同樣的概念，一個關(guān)鍵的差異是內(nèi)部處理的不同。 調(diào)用線程在做其他事情時，異步方法在后臺運行。這意味著這些方法是 I/O 密集型的，即他們大部分時間用于輸入和輸出操作，例如文件或網(wǎng)絡(luò)訪問。 只要有可能，使用異步 I/O 方法代替同步操作很有意義。相同的時間，調(diào)用線程可以在處理桌面應(yīng)用程序中的用戶交互或處理服務(wù)器應(yīng)用程序中的同時處理其他請求，而不僅僅是等待操作完成。

計算密集型的方法要求 CPU 周期工作，并且只能運行在他們專用的后臺線程中。CPU 的核心數(shù)限制了并行運行時的可用線程數(shù)量。操作系統(tǒng)負責(zé)在剩余的線程之間切換，使他們有機會執(zhí)行代碼。 這些方法仍然被并發(fā)地執(zhí)行，卻不必被并行地執(zhí)行。盡管這意味著方法不是同時執(zhí)行，卻可以在其他方法暫停的時候執(zhí)行。

并行 vs 并發(fā)

本文將在最后一段中重點介紹 在 .NET Core中多線程并發(fā)編程。

任務(wù)并行庫

.NET Framework 4 引入了任務(wù)并行庫 (TPL) 作為編寫并發(fā)代碼的首選 API。.NET Core采用相同的編程模式。 要在后臺運行一段代碼，需要將其包裝成一個 任務(wù)：

var backgroundTask = Task.Run(() => DoComplexCalculation(42));
// do other work
var result = backgroundTask.Result;

當(dāng)需要返回結(jié)果時，Task.Run 方法接收一個 函數(shù) (Func) ；當(dāng)不需要返回結(jié)果時，方法 Task.Run 接收一個 動作 (Action) 。當(dāng)然，所有的情況下都可以使用 lambda 表達式，就像我上面例子中調(diào)用帶一個參數(shù)的長時間方法。 線程池中的某個線程將會處理任務(wù)。.NET Core 的運行時包含一個默認調(diào)度程序，使用線程池來處理隊列并執(zhí)行任務(wù)。您可以通過派生 TaskScheduler 類實現(xiàn)自己的調(diào)度算法，代替默認的，但這超過本文的討論范圍。 正如我們之前所見，我使用 Result 屬性來合并被調(diào)用的后臺線程。對于不需要返回結(jié)果的線程，我可以調(diào)用 Wait() 來代替。這兩種方式都將被堵塞到后臺任務(wù)完成。 為了避免堵塞調(diào)用線程 ( 如在ASP.NET Core應(yīng)用程序中) ，可以使用 await 關(guān)鍵字：

var backgroundTask = Task.Run(() => DoComplexCalculation(42));
// do other work
var result = await backgroundTask;

這樣被調(diào)用的線程將被釋放以便處理其他傳入請求。一旦任務(wù)完成，一個可用的工作線程將會繼續(xù)處理請求。當(dāng)然，控制器動作方法必須是異步的:

public async Task<iactionresult> Index() { // method body }

處理異常

將兩個線程合并在一起的時候，任務(wù)拋出的任何異常將被傳遞到調(diào)用線程中：

如果使用 Result 或 Wait() ，它們將被打包到 AggregateException 中。實際的異常將被拋出并存儲在其 InnerException 屬性中。

如果您使用 await，原來的異常將不會被打包。

在這兩種情況下，調(diào)用堆棧的信息將保持不變。

取消任務(wù)

由于任務(wù)是可以長時間運行的，所以你可能想要有一個可以提前取消任務(wù)的選項。實現(xiàn)這個選項，需要在任務(wù)創(chuàng)建的時候傳入取消的令牌 (token)，之后再使用令牌觸發(fā)取消任務(wù)：

var tokenSource = new CancellationTokenSource();
var cancellableTask = Task.Run(() =>
{
for (int i = 0; i < 100; i++)
{
if (tokenSource.Token.IsCancellationRequested)
{
// clean up before exiting
tokenSource.Token.ThrowIfCancellationRequested();
}
// do long-running processing
}
return 42;
}, tokenSource.Token);
// cancel the task
tokenSource.Cancel();
try
{
await cancellableTask;
}
catch (OperationCanceledException e)
{
// handle the exception
}　

實際上，為了提前取消任務(wù)，你需要檢查任務(wù)中的取消令牌，并在需要取消的時候作出反應(yīng):在執(zhí)行必要的清理操作后，調(diào)用 ThrowIfCancellationRequested() 退出任務(wù)。這個方法將會拋出 OperationCanceledException，以便在調(diào)用線程中執(zhí)行相應(yīng)的處理。

協(xié)調(diào)多任務(wù)

如果你需要運行多個后臺任務(wù)，這里有些方法可以幫助到你。 要同時運行多個任務(wù)，只需連續(xù)啟動它們并收集它們的引用，例如在數(shù)組中：

var backgroundTasks = new []
{
Task.Run(() => DoComplexCalculation(1)),
Task.Run(() => DoComplexCalculation(2)),
Task.Run(() => DoComplexCalculation(3))
};

現(xiàn)在你可以使用 Task 類的靜態(tài)方法，等待他們被異步或者同步執(zhí)行完畢。

// wait synchronously
Task.WaitAny(backgroundTasks);
Task.WaitAll(backgroundTasks);
// wait asynchronously
await Task.WhenAny(backgroundTasks);
await Task.WhenAll(backgroundTasks);

實際上，這兩個方法最終都會返回所有任務(wù)的自身，可以像任何其他任務(wù)一樣再次操作。為了獲取對應(yīng)任務(wù)的結(jié)果，你可以檢查該任務(wù)的 Result 屬性。 處理多任務(wù)的異常有點棘手。方法 WaitAll 和 WhenAll 不管哪個任務(wù)被收集到異常時都會拋出異常。不過，對于 WaitAll ，將會收集所有的異常到對應(yīng)的 InnerExceptions 屬性；對于 WhenAll ，只會拋出第一個異常。為了確認哪個任務(wù)拋出了哪個異常，您需要單獨檢查每個任務(wù)的 Status 和 Exception 屬性。 在使用 WaitAny 和 WhenAny 時必須足夠小心。他們會等到第一個任務(wù)完成 (成功或失敗)，即使某個任務(wù)出現(xiàn)異常時也不會拋出任何異常。他們只會返回已完成任務(wù)的索引或者分別返回已完成的任務(wù)。你必須等到任務(wù)完成或訪問其 result 屬性時捕獲異常，例如：

var completedTask = await Task.WhenAny(backgroundTasks);
try
{
var result = await completedTask;
}
catch (Exception e)
{
// handle exception
}

如果你想連續(xù)運行多個任務(wù)，代替并發(fā)任務(wù)，可以使用延續(xù) (continuations)的方式：

var compositeTask = Task.Run(() => DoComplexCalculation(42))
.ContinueWith(previous => DoAnotherComplexCalculation(previous.Result),
TaskContinuationOptions.OnlyOnRanToCompletion)

ContinueWith() 方法允許你把多個任務(wù)一個接著一個執(zhí)行。這個延續(xù)的任務(wù)將獲取到前面任務(wù)的結(jié)果或狀態(tài)的引用。 你仍然可以增加條件判斷是否執(zhí)行延續(xù)任務(wù)，例如只有在前面任務(wù)成功執(zhí)行或者拋出異常時。對比連續(xù)等待多個任務(wù)，提高了靈活性。 當(dāng)然，您可以將延續(xù)任務(wù)與之前討論的所有功能相結(jié)合：異常處理、取消和并行運行任務(wù)。這就有了很大的表演空間，以不同的方式進行組合：

var multipleTasks = new[]
{
Task.Run(() => DoComplexCalculation(1)),
Task.Run(() => DoComplexCalculation(2)),
Task.Run(() => DoComplexCalculation(3))
};
var combinedTask = Task.WhenAll(multipleTasks);
var successfulContinuation = combinedTask.ContinueWith(task =>
CombineResults(task.Result), TaskContinuationOptions.OnlyOnRanToCompletion);
var failedContinuation = combinedTask.ContinueWith(task =>
HandleError(task.Exception), TaskContinuationOptions.NotOnRanToCompletion);
await Task.WhenAny(successfulContinuation, failedContinuation);

任務(wù)同步

如果任務(wù)是完全獨立的，那么我們剛才看到的協(xié)調(diào)方法就已足夠。然而，一旦需要同時共享數(shù)據(jù)，為了防止數(shù)據(jù)損壞，就必須要有額外的同步。 兩個以及更多的線程同時更新一個數(shù)據(jù)結(jié)構(gòu)時，數(shù)據(jù)很快就會變得不一致。就好像下面這個示例代碼一樣：

var counters = new Dictionary< int, int >();
if (counters.ContainsKey(key))
{
counters[key] ++;
}
else
{
counters[key] = 1;
}

當(dāng)多個線程同時執(zhí)行上述代碼時，不同線程中的特定順序執(zhí)行指令可能導(dǎo)致數(shù)據(jù)不正確，例如：

• 所有線程將會檢查集合中是否存在同一個 key
• 結(jié)果，他們都會進入 else 分支，并將這個 key 的值設(shè)為1
• 最后結(jié)果將會是1，而不是2。如果是接連著執(zhí)行代碼的話，將會是預(yù)期的結(jié)果。
  

上述代碼中，臨界區(qū) (critical section) 一次只允許一個線程可以進入。在C# 中，可以使用 lock 語句來實現(xiàn):

var counters = new Dictionary< int, int >();
lock (syncObject)
{
if (counters.ContainsKey(key))
{
counters[key]++;
}
else
{
counters[key] = 1;
}
}

在這個方法中，所有線程都必須共享相同的的 syncObject 。作為最佳做法，syncObject 應(yīng)該是一個專用的 Object 實例，專門用于保護對一個獨立的臨界區(qū)的訪問，避免從外部訪問。 在 lock 語句中，只允許一個線程訪問里面的代碼塊。它將阻止下一個嘗試訪問它的線程，直到前一個線程退出。這將確保線程完整執(zhí)行臨界區(qū)代碼，而不會被另一個線程中斷。當(dāng)然，這將減少并行性并減慢代碼的整體執(zhí)行速度，因此您最好最小化臨界區(qū)的數(shù)量并使其盡可能的短。

使用 Monitor 類來簡化 lock 聲明：

var lockWasTaken = false;
var temp = syncObject;
try
{
Monitor.Enter(temp, ref lockWasTaken);
// lock statement body
}
finally
{
if (lockWasTaken)
{
Monitor.Exit(temp);
}
}

盡管大部分時間您都希望使用 lock 語句，但 Monitor 類可以在需要時給予額外的控制。例如，您可以使用 TryEnter() 而不是 Enter()，并指定一個限定時間，避免無止境地等待鎖釋放。

其他同步基元

Monitor 只是 .NET Core 中眾多同步基元的一員。根據(jù)實際情況，其他基元可能更適合。

Mutex 是 Monitor 更重量級的版本，依賴于底層的操作系統(tǒng)，提供跨多個進程同步訪問資源[1]， 是針對 Mutex 進行同步的推薦替代方案。

SemaphoreSlim 和 Semaphore 可以限制同時訪問資源的最大線程數(shù)量，而不是像 Monitor 一樣只能限制一個線程。 SemaphoreSlim 比 Semaphore 更輕量，但僅限于單個進程。如果可能，您最好使用 SemaphoreSlim 而不是 Semaphore。

ReaderWriterLockSlim 可以區(qū)分兩種對訪問資源的方式。它允許無限數(shù)量的讀取器 (readers) 同時訪問資源，并且限制同時只允許一個寫入器 (writers) 訪問鎖定資源。讀取時線程安全，但修改數(shù)據(jù)時需要獨占資源，很好地保護了資源。

AutoResetEvent、ManualResetEvent 和 ManualResetEventSlim 將堵塞傳入的線程，直到它們接收到一個信號 (即調(diào)用 Set() )。然后等待中的線程將繼續(xù)執(zhí)行。AutoResetEvent 在下一次調(diào)用 Set() 之前，將一直阻塞，并只允許一個線程繼續(xù)執(zhí)行。ManualResetEvent 和 ManualResetEventSlim 不會堵塞線程，除非 Reset() 被調(diào)用。ManualResetEventSlim 比前兩者更輕量，更值得推薦。

Interlocked 提供一種選擇——原子操作，這是替代 locking 和其他同步基元更好的選擇（如果適用）：

// non-atomic operation with a lock
lock (syncObject)
{
counter++;
}
// equivalent atomic operation that doesn't require a lock
Interlocked.Increment(ref counter);

并發(fā)集合

當(dāng)一個臨界區(qū)需要確保對數(shù)據(jù)結(jié)構(gòu)的原子訪問時，用于并發(fā)訪問的專用數(shù)據(jù)結(jié)構(gòu)可能是更好和更有效的替代方案。例如，使用 ConcurrentDictionary 而不是 Dictionary，可以簡化 lock 語句示例：

var counters = new ConcurrentDictionary< int, int >();
counters.TryAdd(key, 0);
lock (syncObject)
{
counters[key]++;
}

自然地，也有可能像下面一樣:

counters.AddOrUpdate(key, 1, (oldKey, oldValue) => oldValue + 1);

因為 update 的委托是臨界區(qū)外面的方法，因此，第二個線程可能在第一個線程更新值之前，讀取到同樣的舊值，使用自己的值有效地覆蓋了第一個線程的更新值，這就丟失了一個增量。錯誤使用并發(fā)集合也是無法避免多線程帶來的問題。 并發(fā)集合的另一個替代方案是 不變的集合 (immutable collections)。 類似于并發(fā)集合，同樣是線程安全的，但是底層實現(xiàn)是不一樣的。任何關(guān)改變數(shù)據(jù)結(jié)構(gòu)的操作將不會改變原來的實例。相反，它們返回一個更改后的副本，并保持原始實例不變：

var original = new Dictionary< int, int >().ToImmutableDictionary();
var modified = original.Add(key, value);

因此在一個線程中對集合任何更改對于其他線程來說都是不可見的。因為它們?nèi)匀灰迷瓉淼奈葱薷牡募希@就是不變的集合本質(zhì)上是線程安全的原因。 當(dāng)然，這使得它們對于解決不同集合的問題很有效。最好的情況是多個線程在同一個輸入集合的情況下，獨立地修改數(shù)據(jù)，在最后一步可能為所有線程合并變更。而使用常規(guī)集合，需要提前為每個線程創(chuàng)建集合的副本。

并行LINQ (PLINQ)

并行LINQ (PLINQ) 是 Task Parallel Library 的替代方案。顧名思義，它很大程度上依賴于 LINQ（語言集成查詢）功能。對于在大集合中執(zhí)行相同的昂貴操作的場景是很有用的。與所有操作都是順序執(zhí)行的普通 LINQ to Objects 不同的是，PLINQ可以在多個CPU上并行執(zhí)行這些操作。 發(fā)揮優(yōu)勢所需要的代碼改動也是極小的:

// sequential execution
var sequential = Enumerable.Range(0, 40)
.Select(n => ExpensiveOperation(n))
.ToArray();
// parallel execution
var parallel = Enumerable.Range(0, 40)
.AsParallel()
.Select(n => ExpensiveOperation(n))
.ToArray();

如你所見，這兩個代碼片段的不同僅僅是調(diào)用 AsParallel()。這將IEnumerable 轉(zhuǎn)換為 ParallelQuery，導(dǎo)致查詢的部分并行運行。要切換為回順序執(zhí)行，您可以調(diào)用 AsSequential()，它將再次返回一個IEnumerable。 默認情況下，PLINQ 不保留集合中的順序，以便讓進程更有效率。但是當(dāng)順序很重要時，可以調(diào)用 AsOrdered():

var parallel = Enumerable.Range(0, 40)
.AsParallel()
.AsOrdered()
.Select(n => ExpensiveOperation(n))
.ToArray();

同理，你可以通過調(diào)用 AsUnordered() 切換回來。

在完整的 .NET Framework 中并發(fā)編程

由于 .NET Core 是完整的 .NET Framework 的簡化實現(xiàn)，所以 .NET Framework 中所有并行編程方法也可以在.NET Core 中使用。唯一的例外是不變的集合，它們不是完整的 .NET Framework 的組成部分。它們作為單獨的 NuGet 軟件包(System.Collections.Immutable)分發(fā)，您需要在項目中安裝使用。

結(jié)論：

每當(dāng)應(yīng)用程序包含可以并行運行的 CPU 密集型代碼時，利用并發(fā)編程來提高性能并提高硬件利用率是很有意義的。 .NET Core 中的 API 抽象了許多細節(jié)，使編寫并發(fā)代碼更容易。然而需要注意某些潛在的問題， 其中大部分涉及從多個線程訪問共享數(shù)據(jù)。 如果可以的話，你應(yīng)該完全避免這種情況。如果不行，請確保選擇最合適的同步方法或數(shù)據(jù)結(jié)構(gòu)。

以上就是本文的全部內(nèi)容，希望對大家的學(xué)習(xí)有所幫助，也希望大家多多支持我們。