第五章：C#并行編程

第五章：C#并行編程基礎(chǔ)

第五章：C#并行編程基礎(chǔ)

并行編程用來拆分CPU密集型任務(wù)，并將它們分發(fā)給多個(gè)線程，它利用多核處理器的能力，使程序中的任務(wù)能夠并行執(zhí)行，從而加快處理速度。本章實(shí)例僅考慮CPU密集型任務(wù)，I/O并行請參考第三章。

并行編程的核心思想是將工作分解成多個(gè)部分，并將這些部分分配給不同的處理單元（如 CPU 核心）同時(shí)執(zhí)行。常見的并行編程模式包括數(shù)據(jù)并行、任務(wù)并行和流水線并行。

在 .NET 中，常用的并行編程工具有：

Task Parallel Library (TPL)：提供了任務(wù)并行化的接口，常用的類如 Task 和 Parallel。
PLINQ (Parallel LINQ)：允許并行化 LINQ 查詢。
并行集合：如 ConcurrentDictionary，提供了線程安全的數(shù)據(jù)結(jié)構(gòu)，方便并行任務(wù)共享數(shù)據(jù)。

5.1 并行處理：使用 `Parallel.ForEach` 和 `Parallel.For`

問題

在進(jìn)行 CPU 密集型任務(wù)時(shí)，單線程執(zhí)行可能會導(dǎo)致程序運(yùn)行緩慢。例如，在對大量數(shù)據(jù)進(jìn)行復(fù)雜計(jì)算時(shí)，單線程處理無法充分利用多核處理器的性能。如何高效地并行處理這些計(jì)算任務(wù)呢？

解決方案

Parallel.ForEach 和 Parallel.For 是 .NET 中用于并行處理任務(wù)的工具，能夠幫助開發(fā)者輕松利用多核處理器，提升計(jì)算效率。Parallel.ForEach 適合遍歷集合，而 Parallel.For 適合處理索引范圍的循環(huán)。

示例 1：并行計(jì)算大量數(shù)字的平方根

假設(shè)我們有一組浮點(diǎn)數(shù)，需要對每個(gè)數(shù)字計(jì)算平方根。使用 Parallel.ForEach 可以并行執(zhí)行計(jì)算，加快處理速度。

void CalculateSquareRoots(IEnumerable<double> numbers)
{
    Parallel.ForEach(numbers, number =>
    {
        var result = Math.Sqrt(number);
        Console.WriteLine($"Sqrt({number}) = {result}");
    });
}

解釋：

Parallel.ForEach 并行遍歷 numbers 集合，對每個(gè)數(shù)字計(jì)算平方根。
多個(gè)線程同時(shí)處理，能顯著減少計(jì)算時(shí)間。

示例 2：提前終止并行計(jì)算

假設(shè)我們在處理一系列數(shù)學(xué)計(jì)算時(shí)，如果發(fā)現(xiàn)計(jì)算結(jié)果不符合預(yù)期，就希望立即停止進(jìn)一步計(jì)算。這時(shí)可以使用 ParallelLoopState 的 Stop() 方法。

void FindFirstPrime(IEnumerable<int> numbers)
{
    Parallel.ForEach(numbers, (number, state) =>
    {
        if (IsPrime(number))
        {
            Console.WriteLine($"First prime found: {number}");
            state.Stop();  // 提前終止循環(huán)
        }
    });
}

bool IsPrime(int number)
{
    if (number < 2) return false;
    for (int i = 2; i <= Math.Sqrt(number); i++)
    {
        if (number % i == 0) return false;
    }
    return true;
}

解釋：

一旦找到第一個(gè)素?cái)?shù)，state.Stop() 會停止分配新的任務(wù)。
已經(jīng)開始的任務(wù)會繼續(xù)執(zhí)行，因此不能保證完全停止所有計(jì)算。

示例 3：并行計(jì)算時(shí)使用 `CancellationToken`

假設(shè)用戶可以從外部取消計(jì)算任務(wù)，例如在 UI 應(yīng)用中用戶點(diǎn)擊“取消”按鈕。這時(shí)可以使用 CancellationToken 來實(shí)現(xiàn)取消功能。

void CalculateFactorials(IEnumerable<int> numbers, CancellationToken token)
{
    Parallel.ForEach(numbers, new ParallelOptions { CancellationToken = token }, number =>
    {
        var result = Factorial(number);
        Console.WriteLine($"Factorial({number}) = {result}");
    });
}

long Factorial(int n)
{
    long result = 1;
    for (int i = 2; i <= n; i++)
    {
        result *= i;
    }
    return result;
}

解釋：

當(dāng)外部調(diào)用 token.Cancel() 時(shí)，未開始的任務(wù)會被取消。
已經(jīng)開始的任務(wù)會檢查 CancellationToken，并在取消時(shí)拋出異常來取消任務(wù)。

示例 4：處理共享狀態(tài)

在并行計(jì)算時(shí)，如果多個(gè)線程訪問同一個(gè)共享變量，就會出現(xiàn)競態(tài)條件。為了避免這種情況，需要使用鎖進(jìn)行同步。以下示例展示了如何并行計(jì)算數(shù)據(jù)并統(tǒng)計(jì)符合特定條件的元素個(gè)數(shù)。

int CountEvenNumbers(IEnumerable<int> numbers)
{
    int evenCount = 0;
    object lockObj = new object();

    Parallel.ForEach(numbers, number =>
    {
        if (number % 2 == 0)
        {
            lock (lockObj)
            {
                evenCount++;
            }
        }
    });

    return evenCount;
}

解釋：

evenCount 是共享狀態(tài)，使用 lock 確保線程安全。
多個(gè)線程同時(shí)訪問時(shí)，lock 防止競態(tài)條件導(dǎo)致計(jì)數(shù)錯(cuò)誤。

`Parallel.For` 示例：并行處理數(shù)組

當(dāng)處理的是數(shù)組或列表這種可以使用索引訪問的數(shù)據(jù)結(jié)構(gòu)時(shí)，Parallel.For 是更好的選擇。例如，計(jì)算一組整數(shù)的平方值：

void SquareArrayElements(int[] numbers)
{
    Parallel.For(0, numbers.Length, i =>
    {
        numbers[i] = numbers[i] * numbers[i];
    });
}

解釋：

Parallel.For 使用索引范圍遍歷數(shù)組，對每個(gè)元素并行計(jì)算平方值。
適合處理大數(shù)組的 CPU 密集型任務(wù)。

小結(jié)

Parallel.ForEach：用于并行處理集合中的每個(gè)元素，適合處理非索引結(jié)構(gòu)的數(shù)據(jù)。
Parallel.For：用于并行處理索引循環(huán)，適合處理數(shù)組等索引結(jié)構(gòu)的數(shù)據(jù)。
Stop() 和 CancellationToken：用于在特定條件下提前終止循環(huán)或取消操作。
共享狀態(tài)的同步：當(dāng)并行訪問共享狀態(tài)時(shí)，需使用同步機(jī)制（如 lock）以避免競態(tài)條件。

5.2 并行聚合：利用 `Parallel.ForEach` 和 PLINQ 實(shí)現(xiàn)高效聚合

問題

在處理大規(guī)模數(shù)據(jù)時(shí)，聚合操作（如求和、求平均值、最大值等）通常會成為性能瓶頸。單線程聚合計(jì)算無法利用多核處理器的優(yōu)勢，尤其在數(shù)據(jù)量龐大時(shí)，處理時(shí)間會顯著增加。如何通過并行計(jì)算提升聚合操作的效率？

解決方案

.NET 中的 Parallel 類和 PLINQ 都支持并行聚合，能充分利用多核 CPU 加快計(jì)算速度。Parallel.ForEach 提供了局部變量 (localInit 和 localFinally) 來支持并行聚合，而 PLINQ 則提供了更加簡潔的 API，使代碼更具可讀性。

示例 1：使用 `Parallel.ForEach` 實(shí)現(xiàn)并行求和

在 Parallel.ForEach 中，使用 localInit 創(chuàng)建每個(gè)線程的局部變量，用于累加結(jié)果。最后通過 localFinally 聚合所有線程的局部結(jié)果。

int ParallelSum(IEnumerable<int> values)
{
    object mutex = new object();
    int result = 0;

    Parallel.ForEach(
        source: values,
        localInit: () => 0,
        body: (item, state, localSum) => localSum + item,
        localFinally: localSum =>
        {
            lock (mutex)
            {
                result += localSum;
            }
        }
    );

    return result;
}

解釋：

localInit：為每個(gè)線程創(chuàng)建一個(gè)局部變量 localSum，初始值為 0。
body：并行遍歷 values 集合，計(jì)算局部求和。
localFinally：將每個(gè)線程的局部結(jié)果 (localSum) 聚合到最終結(jié)果 result 中，使用 lock 確保線程安全。

注意：

這種方法雖然有效，但鎖的使用可能會影響性能，尤其是在結(jié)果聚合階段有大量線程競爭時(shí)。

示例 2：使用 PLINQ 簡化并行求和

PLINQ 是 .NET 中用于并行 LINQ 查詢的擴(kuò)展，內(nèi)置了對常見聚合操作的支持，代碼更加簡潔。

int ParallelSum(IEnumerable<int> values)
{
    return values.AsParallel().Sum();
}

解釋：

AsParallel() 方法將集合轉(zhuǎn)換為并行查詢。
Sum() 方法在內(nèi)部自動進(jìn)行并行化，利用多核處理器提高計(jì)算效率。

優(yōu)勢：

代碼簡潔明了，不需要顯式管理線程和同步。
PLINQ 內(nèi)置的并行化支持能自動調(diào)整并行度，適應(yīng)系統(tǒng)的負(fù)載情況。

示例 3：使用 PLINQ 的 `Aggregate` 方法實(shí)現(xiàn)自定義聚合

如果需要執(zhí)行更加復(fù)雜的聚合操作，可以使用 PLINQ 的 Aggregate 方法，該方法提供了靈活的聚合功能。

int ParallelSum(IEnumerable<int> values)
{
    return values.AsParallel().Aggregate(
        seed: 0,
        func: (sum, item) => sum + item
    );
}

解釋：

seed 是初始值，即聚合操作的起點(diǎn)。
func 是用于累加的聚合函數(shù)，將每個(gè)元素累加到 sum 中。
Aggregate 方法支持更多自定義操作，適合復(fù)雜的聚合計(jì)算場景。

小結(jié)

Parallel.ForEach 提供了基于局部變量的聚合支持，可以在并行循環(huán)內(nèi)部高效計(jì)算局部結(jié)果，再進(jìn)行最終聚合。
PLINQ 提供了內(nèi)置的聚合操作（如 Sum 和 Aggregate），代碼更簡潔、可讀性更高，且性能表現(xiàn)通常優(yōu)于手動實(shí)現(xiàn)的并行聚合。
在大多數(shù)情況下，使用 PLINQ 會更簡單和高效，除非需要特殊控制并行任務(wù)的執(zhí)行方式或需要自定義復(fù)雜的聚合邏輯。

最佳實(shí)踐

優(yōu)先使用 PLINQ：如果聚合邏輯較為簡單，推薦使用 PLINQ，代碼更簡潔，性能優(yōu)化也較好。
避免鎖競爭：在并行聚合時(shí)，盡量減少鎖的使用，或者使用無鎖并發(fā)結(jié)構(gòu)（如 ThreadLocal<T> 或 Interlocked）。
注意線程安全：無論是使用 Parallel.ForEach 還是 PLINQ，都要確保聚合操作對共享狀態(tài)的訪問是線程安全的。

并行聚合能夠顯著提升性能，但需要正確使用線程同步機(jī)制，避免競態(tài)條件，才能充分發(fā)揮并行計(jì)算的優(yōu)勢。

5.3 并行調(diào)用：使用 `Parallel.Invoke` 實(shí)現(xiàn)并行執(zhí)行

問題

在程序中，有時(shí)需要并行執(zhí)行一系列獨(dú)立的方法，且這些方法之間沒有相互依賴。如果串行執(zhí)行這些方法，無法充分利用多核 CPU 的并行計(jì)算能力，導(dǎo)致性能浪費(fèi)。如何高效地并行調(diào)用多個(gè)方法？

解決方案

Parallel 類提供了 Invoke 方法，專門用于并行執(zhí)行多個(gè)彼此獨(dú)立的方法。Parallel.Invoke 能夠并行地調(diào)用多個(gè)委托，并等待所有方法執(zhí)行完畢。

示例 1：并行處理數(shù)組的兩個(gè)部分

假設(shè)我們有一個(gè)數(shù)組，需要將其拆分為兩個(gè)部分，并對每一部分進(jìn)行獨(dú)立處理。可以使用 Parallel.Invoke 來并行執(zhí)行這兩個(gè)處理任務(wù)。

void ProcessArray(double[] array)
{
    Parallel.Invoke(
        () => ProcessPartialArray(array, 0, array.Length / 2),
        () => ProcessPartialArray(array, array.Length / 2, array.Length)
    );
}

void ProcessPartialArray(double[] array, int begin, int end)
{
    // 進(jìn)行 CPU 密集型處理……
}

解釋：

Parallel.Invoke 接收一組委托（匿名方法或 lambda 表達(dá)式），并并行執(zhí)行這些委托。
ProcessPartialArray 方法對數(shù)組的一部分進(jìn)行處理，兩個(gè)任務(wù)獨(dú)立執(zhí)行，能充分利用多核 CPU。

示例 2：動態(tài)數(shù)量的并行調(diào)用

如果在運(yùn)行時(shí)才確定需要調(diào)用多少次方法，可以將一組委托傳遞給 Parallel.Invoke 方法。例如，重復(fù)執(zhí)行某個(gè)操作 20 次：

void DoAction20Times(Action action)
{
    Action[] actions = Enumerable.Repeat(action, 20).ToArray();
    Parallel.Invoke(actions);
}

解釋：

使用 Enumerable.Repeat 創(chuàng)建一個(gè)包含 20 個(gè)相同委托的數(shù)組。
Parallel.Invoke 并行執(zhí)行數(shù)組中的所有委托，適用于動態(tài)數(shù)量的并行任務(wù)。

示例 3：支持取消操作的并行調(diào)用

Parallel.Invoke 支持使用 CancellationToken 來取消并行調(diào)用。用戶可以在運(yùn)行時(shí)請求取消操作，從而停止尚未完成的任務(wù)。

void DoAction20Times(Action action, CancellationToken token)
{
    Action[] actions = Enumerable.Repeat(action, 20).ToArray();
    Parallel.Invoke(new ParallelOptions { CancellationToken = token }, actions);
}

解釋：

ParallelOptions 中包含 CancellationToken，用于控制并行操作的取消。
如果在調(diào)用期間觸發(fā)取消請求，Parallel.Invoke 會停止尚未開始的任務(wù)，已開始的任務(wù)會繼續(xù)執(zhí)行完畢。

小結(jié)

Parallel.Invoke：適合并行執(zhí)行一組彼此獨(dú)立的方法，簡化了并行調(diào)用的代碼編寫。
動態(tài)任務(wù)支持：可以傳遞一組委托（數(shù)組或集合）給 Parallel.Invoke，適應(yīng)動態(tài)數(shù)量的并行任務(wù)。
取消支持：通過 CancellationToken 可以靈活控制并行操作的取消，適合用戶交互場景。

適用場景與限制

適用場景：
- 需要并行執(zhí)行一組獨(dú)立任務(wù)。
- 任務(wù)數(shù)量較少且無需返回值時(shí)，Parallel.Invoke 是簡潔的選擇。
不適用場景：
- 如果需要對集合中的每個(gè)元素執(zhí)行操作，使用 Parallel.ForEach 更合適。
- 如果任務(wù)需要返回結(jié)果，并且需要對結(jié)果進(jìn)行處理或聚合，使用 PLINQ 更為便捷。

最佳實(shí)踐

控制任務(wù)數(shù)量：Parallel.Invoke 適合數(shù)量有限的任務(wù)，過多的任務(wù)會增加線程上下文切換的開銷，影響性能。
避免阻塞操作：并行調(diào)用的方法應(yīng)盡量避免使用阻塞操作（如 I/O 操作），以免降低并行效率。
利用取消機(jī)制：在需要用戶動態(tài)控制的場景中，使用 CancellationToken 提供靈活的取消支持。

Parallel.Invoke 是一種簡單且高效的并行調(diào)用工具，能幫助開發(fā)者快速實(shí)現(xiàn)并行執(zhí)行。但在更復(fù)雜的并行場景下（如大規(guī)模數(shù)據(jù)處理），應(yīng)根據(jù)具體需求選擇合適的并行編程工具，如 Parallel.ForEach 或 PLINQ。

5.4 動態(tài)并行：使用 `Task` 實(shí)現(xiàn)復(fù)雜的動態(tài)并行結(jié)構(gòu)

問題

在實(shí)際應(yīng)用中，有時(shí)任務(wù)的數(shù)量和結(jié)構(gòu)是未知的，直到運(yùn)行時(shí)才確定。這種場景通常非常復(fù)雜，無法使用 Parallel.ForEach 或 PLINQ 等并行工具進(jìn)行簡單處理。例如，我們可能需要遍歷一個(gè)結(jié)構(gòu)復(fù)雜的二叉樹，針對每個(gè)節(jié)點(diǎn)執(zhí)行某些計(jì)算，而樹的結(jié)構(gòu)在運(yùn)行前無法確定。如何在這種動態(tài)并行場景下高效地處理任務(wù)？

解決方案

Task 是 .NET 中并行編程的核心類型。Parallel 類和 PLINQ 都是對 Task 的高級封裝。如果需要處理動態(tài)并行結(jié)構(gòu)，直接使用 Task 是最靈活的選擇。

示例 1：動態(tài)遍歷二叉樹

假設(shè)我們需要對二叉樹的每個(gè)節(jié)點(diǎn)執(zhí)行計(jì)算，并且計(jì)算需要并行處理，但子節(jié)點(diǎn)必須在父節(jié)點(diǎn)處理完后再開始。

void Traverse(Node current)
{
    // 對當(dāng)前節(jié)點(diǎn)進(jìn)行耗時(shí)操作
    DoExpensiveActionOnNode(current);

    // 遞歸處理左子節(jié)點(diǎn)
    if (current.Left != null)
    {
        Task.Factory.StartNew(
            () => Traverse(current.Left),
            CancellationToken.None,
            TaskCreationOptions.AttachedToParent,
            TaskScheduler.Default);
    }

    // 遞歸處理右子節(jié)點(diǎn)
    if (current.Right != null)
    {
        Task.Factory.StartNew(
            () => Traverse(current.Right),
            CancellationToken.None,
            TaskCreationOptions.AttachedToParent,
            TaskScheduler.Default);
    }
}

void ProcessTree(Node root)
{
    // 啟動頂層任務(wù)并等待其完成
    Task task = Task.Factory.StartNew(
        () => Traverse(root),
        CancellationToken.None,
        TaskCreationOptions.None,
        TaskScheduler.Default);

    task.Wait();
}

解釋：

Task.Factory.StartNew 用于啟動新的任務(wù)。
TaskCreationOptions.AttachedToParent 標(biāo)志使子任務(wù)附屬于父任務(wù)。這樣，父任務(wù)會等待所有子任務(wù)完成后再結(jié)束，并且子任務(wù)拋出的異常會傳遞到父任務(wù)。
ProcessTree 方法啟動對根節(jié)點(diǎn)的處理，并等待整個(gè)樹的遍歷任務(wù)完成。

為什么使用 `AttachedToParent`？

父子任務(wù)關(guān)系：AttachedToParent 建立了顯式的父子任務(wù)關(guān)系，確保父任務(wù)等待所有子任務(wù)完成。
異常傳播：如果子任務(wù)拋出異常，它會被傳播到父任務(wù)，這樣可以統(tǒng)一處理異常，而不需要單獨(dú)處理每個(gè)子任務(wù)的異常。

示例 2：任務(wù)延續(xù)（Task Continuation）

在某些情況下，任務(wù)之間沒有父子關(guān)系，而是需要在某個(gè)任務(wù)完成后繼續(xù)執(zhí)行另一個(gè)任務(wù)。可以使用 ContinueWith 方法實(shí)現(xiàn)任務(wù)的延續(xù)。

Task task = Task.Factory.StartNew(
    () => Thread.Sleep(TimeSpan.FromSeconds(2)),
    CancellationToken.None,
    TaskCreationOptions.None,
    TaskScheduler.Default);

Task continuation = task.ContinueWith(
    t => Console.WriteLine("任務(wù)已完成"),
    CancellationToken.None,
    TaskContinuationOptions.None,
    TaskScheduler.Default);

解釋：

ContinueWith 會在原始任務(wù)完成后執(zhí)行。這里，t 參數(shù)代表原始任務(wù)，允許在延續(xù)任務(wù)中訪問原始任務(wù)的狀態(tài)和結(jié)果。
這種方式適用于沒有父子依賴關(guān)系的任務(wù)鏈，常用于串行化一系列異步操作。

動態(tài)并行的最佳實(shí)踐

合理使用 AttachedToParent：
- 在需要顯式父子任務(wù)關(guān)系的情況下使用 AttachedToParent，可以簡化任務(wù)的等待和異常處理。
- 對于沒有父子依賴的任務(wù)，不需要使用 AttachedToParent，避免不必要的任務(wù)層級。
避免使用阻塞操作：
- 在并行任務(wù)中，避免使用 Task.Wait、Task.Result 等阻塞操作，盡量使用 await、Task.WhenAll 等異步方法。
- 阻塞操作會占用線程資源，影響并行性能，尤其是在高并發(fā)場景中。
選擇合適的任務(wù)調(diào)度器：
- 默認(rèn)使用 TaskScheduler.Default，它會根據(jù)線程池的調(diào)度策略動態(tài)分配線程。
- 如果有特殊的調(diào)度需求（如 UI 線程調(diào)度），可以指定不同的調(diào)度器。

5.5 并行 LINQ（PLINQ）：利用并行化提升 LINQ 性能

問題

假設(shè)需要對一個(gè)數(shù)據(jù)序列執(zhí)行計(jì)算，生成新的序列，或者對序列進(jìn)行聚合操作。傳統(tǒng)的 LINQ 操作是單線程順序執(zhí)行的，在處理大規(guī)模數(shù)據(jù)或 CPU 密集型任務(wù)時(shí)性能不足。如何在保留 LINQ 易用性的同時(shí)，提升并行處理性能？

解決方案

PLINQ（Parallel LINQ）是 LINQ 的并行版本，允許開發(fā)者使用與 LINQ 類似的語法來編寫并行查詢。PLINQ 會自動并行化 LINQ 查詢，以充分利用多核 CPU 的計(jì)算能力。它非常適合需要對輸入序列進(jìn)行變換或聚合的場景。

示例 1：使用 PLINQ 并行處理序列

假設(shè)我們有一個(gè)整數(shù)序列，想要將其中的每個(gè)元素乘以 2。對于簡單的 CPU 密集型計(jì)算任務(wù)，PLINQ 可以顯著提升性能。

IEnumerable<int> MultiplyBy2(IEnumerable<int> values)
{
    return values.AsParallel().Select(value => value * 2);
}

解釋：

AsParallel() 方法將序列轉(zhuǎn)換為 PLINQ 查詢，啟用并行化。
Select 操作會在多個(gè)線程上并行執(zhí)行，每個(gè)線程處理一部分?jǐn)?shù)據(jù)。

注意：PLINQ 默認(rèn)不保證輸出的順序。如果順序很重要，可以使用 AsOrdered() 方法。

示例 2：保留順序的并行查詢

在某些場景下，需要保持輸入序列和輸出序列的順序一致。例如，處理日志記錄數(shù)據(jù)時(shí)，順序可能很重要。

IEnumerable<int> MultiplyBy2Ordered(IEnumerable<int> values)
{
    return values.AsParallel().AsOrdered().Select(value => value * 2);
}

解釋：

AsOrdered() 告訴 PLINQ 保持順序，因此輸出序列的順序與輸入序列一致。
保留順序會有一定的性能損失，但在需要順序的一些場景中這是必要的。

示例 3：并行求和

PLINQ 還可以用于聚合操作，如求和、求平均等。以下示例展示了如何并行求和：

int ParallelSum(IEnumerable<int> values)
{
    return values.AsParallel().Sum();
}

解釋：

Sum() 是 LINQ 的標(biāo)準(zhǔn)聚合操作。通過 AsParallel()，它會在多個(gè)線程上并行執(zhí)行，利用多核處理器提升性能。
PLINQ 自動處理并行化的細(xì)節(jié)，不需要手動管理線程或同步狀態(tài)。

PLINQ 支持的操作符

PLINQ 提供了大量的并行版本運(yùn)算符，包括：

過濾（Where）：并行過濾序列中的元素。
投影（Select）：并行變換序列中的每個(gè)元素。
聚合（Sum、Average、Aggregate）：并行計(jì)算聚合結(jié)果。

例如，使用 Where 和 Select 的并行查詢：

IEnumerable<int> FilterAndMultiply(IEnumerable<int> values)
{
    return values.AsParallel()
                 .Where(value => value % 2 == 0)
                 .Select(value => value * 2);
}

解釋：

先使用 Where 過濾出偶數(shù)，再用 Select 對每個(gè)元素進(jìn)行變換。
AsParallel() 使整個(gè)查詢并行執(zhí)行。

使用 PLINQ 的最佳實(shí)踐

使用 AsParallel() 慎重：
- 并非所有的 LINQ 查詢都能從并行化中獲益。對于小數(shù)據(jù)集，串行執(zhí)行可能更快。
- 使用 AsParallel() 前，應(yīng)確保查詢包含 CPU 密集型操作，且數(shù)據(jù)集較大。
避免使用 AsOrdered() 除非必要：
- 保留順序會降低并行性能。如果順序不重要，盡量使用默認(rèn)的無序查詢。
處理異常：
- PLINQ 中的異常會被包裝在 AggregateException 中。可以使用 try-catch 捕獲異常，并通過 AggregateException.Flatten() 查看所有內(nèi)部異常。

try
{
    var result = values.AsParallel().Select(value => 1 / value).ToList();
}
catch (AggregateException ex)
{
    foreach (var innerException in ex.Flatten().InnerExceptions)
    {
        Console.WriteLine(innerException.Message);
    }
}

監(jiān)控性能：
- 使用 ParallelQuery 的方法時(shí)，可以通過 WithDegreeOfParallelism 設(shè)置并行度來優(yōu)化性能。
- 默認(rèn)情況下，PLINQ 使用所有可用的核心線程。可以根據(jù)場景調(diào)整并行度以避免線程爭用。

var result = values.AsParallel()
                   .WithDegreeOfParallelism(4)
                   .Select(value => value * 2);

posted @ 2024-12-09 15:53 平元兄閱讀(405) 評論(0) 收藏舉報(bào)

刷新頁面返回頂部

Michael Shen

平元兄

第五章：C#并行編程

第五章：C#并行編程基礎(chǔ)

5.1 并行處理：使用 Parallel.ForEach 和 Parallel.For

問題

解決方案

示例 1：并行計(jì)算大量數(shù)字的平方根

示例 2：提前終止并行計(jì)算

示例 3：并行計(jì)算時(shí)使用 CancellationToken

示例 4：處理共享狀態(tài)

Parallel.For 示例：并行處理數(shù)組

小結(jié)

5.2 并行聚合：利用 Parallel.ForEach 和 PLINQ 實(shí)現(xiàn)高效聚合

問題

解決方案

示例 1：使用 Parallel.ForEach 實(shí)現(xiàn)并行求和

示例 2：使用 PLINQ 簡化并行求和

示例 3：使用 PLINQ 的 Aggregate 方法實(shí)現(xiàn)自定義聚合

小結(jié)

最佳實(shí)踐

5.3 并行調(diào)用：使用 Parallel.Invoke 實(shí)現(xiàn)并行執(zhí)行

問題

解決方案

示例 1：并行處理數(shù)組的兩個(gè)部分

示例 2：動態(tài)數(shù)量的并行調(diào)用

示例 3：支持取消操作的并行調(diào)用

小結(jié)

適用場景與限制

最佳實(shí)踐

5.4 動態(tài)并行：使用 Task 實(shí)現(xiàn)復(fù)雜的動態(tài)并行結(jié)構(gòu)

問題

解決方案

示例 1：動態(tài)遍歷二叉樹

為什么使用 AttachedToParent？

示例 2：任務(wù)延續(xù)（Task Continuation）

動態(tài)并行的最佳實(shí)踐

5.5 并行 LINQ（PLINQ）：利用并行化提升 LINQ 性能

問題

解決方案

示例 1：使用 PLINQ 并行處理序列

示例 2：保留順序的并行查詢

示例 3：并行求和

PLINQ 支持的操作符

使用 PLINQ 的最佳實(shí)踐

公告

5.1 并行處理：使用 `Parallel.ForEach` 和 `Parallel.For`

示例 3：并行計(jì)算時(shí)使用 `CancellationToken`

`Parallel.For` 示例：并行處理數(shù)組

5.2 并行聚合：利用 `Parallel.ForEach` 和 PLINQ 實(shí)現(xiàn)高效聚合

示例 1：使用 `Parallel.ForEach` 實(shí)現(xiàn)并行求和

示例 3：使用 PLINQ 的 `Aggregate` 方法實(shí)現(xiàn)自定義聚合

5.3 并行調(diào)用：使用 `Parallel.Invoke` 實(shí)現(xiàn)并行執(zhí)行

5.4 動態(tài)并行：使用 `Task` 實(shí)現(xiàn)復(fù)雜的動態(tài)并行結(jié)構(gòu)

為什么使用 `AttachedToParent`？