第十三章：調度

第十三章：調度

我們都知道代碼的執行需要被安排到某個線程上，而調度器則負責決定這些代碼的執行位置和方式。C# 提供了多種調度工具和機制，例如任務調度器 (TaskScheduler)、同步上下文 (SynchronizationContext) 以及各種線程池策略。這些工具為開發者實現自定義調度邏輯提供了靈活性，同時也增強了并發程序的可控性和性能優化能力。

通常情況下，建議盡量使用默認調度器的行為，因為它們的默認設置已被設計得足夠高效。例如，異步代碼中的 await 運算符會自動在相同的同步上下文中恢復方法執行，除非顯式覆蓋這種默認行為（參見相關章節）。類似地，響應式編程中的觸發事件也有合理的默認上下文，可以通過 ObserveOn 來修改。

然而，有時需要在特定的上下文中執行代碼，比如在 UI 線程上下文或 ASP.NET 請求上下文中。在這些場景下，開發者可以使用調度器來靈活地控制代碼的執行方式。本章將深入探討調度相關的機制和最佳實踐。

13.1 任務調度器 (TaskScheduler) 基礎

在 .NET 中，TaskScheduler 是任務調度的核心組件，它負責決定任務 (Task) 何時、如何以及在哪個線程上執行。可以把 TaskScheduler 看作是一個橋梁，它連接了高層的任務模型（Task 和 Task<T>）與底層的線程管理（如線程池或用戶定義的線程）。

`TaskScheduler` 的作用與核心機制

任務排隊與執行：
當你通過 Task 提交一個任務時，任務會被傳遞給調度器。調度器可以決定立即執行任務、將其排隊，或者以其他策略（如優先級）調度任務。
選擇線程：
調度器可以選擇使用線程池線程、創建新的線程、或綁定到特定上下文（如 UI 線程）來運行任務。
異步與同步調度：
調度器可以決定任務是否異步執行（在線程池線程上運行）還是同步執行（在調用線程上運行）。
任務依賴：
調度器負責處理任務依賴鏈，確保父任務和子任務按照正確的順序或上下文調度。

默認任務調度器 `TaskScheduler.Default`

默認調度器返回的是一個線程池任務調度器ThreadPoolTaskScheduler的實例，因此會將任務分派到線程池線程上運行，自動處理父子任務的關系。例如，子任務默認繼承父任務的調度器。TaskScheduler.Default 盡量公平地調度任務，但不支持顯式的優先級管理（如高優先級任務優先運行）。

使用默認調度器將任務調度到線程池線程上

默認調度器通常通過 Task.Run 和 Task.Factory.StartNew 提交任務。

代碼示例：`Task.Run` 的使用

Task.Run 是最常用的創建任務的方式，適合將獨立的異步工作提交給默認調度器。

using System;
using System.Threading.Tasks;

class Program
{
    static async Task Main()
    {
        Console.WriteLine($"Main thread: {Environment.CurrentManagedThreadId}");

        Task task = Task.Run(() =>
        {
            Console.WriteLine($"Task running on thread: {Environment.CurrentManagedThreadId}");
        });

        await task;

        Console.WriteLine("Task completed.");
    }
}

輸出示例：

Main thread: 1
Task running on thread: 5
Task completed.

解釋：

Task.Run 使用線程池線程執行任務。
適用于輕量級異步操作，尤其是 CPU 密集型任務。

代碼示例：`Task.Factory.StartNew` 的使用

Task.Factory.StartNew 提供更多的任務創建選項，例如任務狀態、調度器、任務附加選項等，如非特殊情況，Task.Run 已經足夠，且更安全。

using System;
using System.Threading.Tasks;

class Program
{
    static void Main()
    {
        Console.WriteLine($"Main thread: {Environment.CurrentManagedThreadId}");

        Task task = Task.Factory.StartNew(() =>
        {
            Console.WriteLine($"Task running on thread: {Environment.CurrentManagedThreadId}");
        });

        task.Wait(); // 等待任務完成

        Console.WriteLine("Task completed.");
    }
}

輸出示例：

Main thread: 1
Task running on thread: 6
Task completed.

Task.Factory.StartNew不指定調度器的話，默認使用的是TaskScheduler.Default，當然也可以顯示指定：

Task.Factory.StartNew(() =>
{
    Console.WriteLine($"Task running on thread: {Environment.CurrentManagedThreadId}");
}, CancellationToken.None, TaskCreationOptions.None, scheduler: TaskScheduler.Default);

再次強調一下，優先使用簡單而高效的 Task.Run，僅在需要自定義選項或特殊場景時才考慮使用 Task.Factory.StartNew。

捕獲與恢復同步上下文

`TaskScheduler.FromCurrentSynchronizationContext`

TaskScheduler.FromCurrentSynchronizationContext 可捕獲當前的 SynchronizationContext，根據捕獲的同步上下文創建調度器，通過該調度器可以將代碼調度回捕獲的同步上下文。例如，在 UI 應用中，這可用于確保任務在 UI 線程中執行。

var scheduler = TaskScheduler.FromCurrentSynchronizationContext();
Task.Factory.StartNew(() =>
{
    Console.WriteLine("Running on captured context.");
    btn1.Text="我是一個按鈕";
}, CancellationToken.None, TaskCreationOptions.None, scheduler);

提示：如果當前線程的執行環境，無特定的同步上下文，比如線程池線程，執行TaskScheduler.FromCurrentSynchronizationContext()會拋出異常。

SynchronizationContext 的作用

這是一個通用型調度上下文，用于不同平臺抽象調度邏輯。
常見用法包括恢復到 UI 線程、HTTP 請求上下文（在 ASP.NET 中）等。
避免直接使用平臺特定的類型（如 Dispatcher 或 CoreDispatcher），以防代碼與特定平臺耦合。

詳細用法參考后續小節

`TaskScheduler` 的核心方法

要深入理解 TaskScheduler，需要了解它的三個核心方法。這些方法是自定義調度器的基礎。

1. `QueueTask(Task task)`

將任務加入調度隊列，準備執行。
默認實現會將任務排隊到線程池。
自定義調度器可以在此方法中實現自定義的任務排隊邏輯，比如優先級隊列。

2. `TryExecuteTaskInline(Task task, bool taskWasPreviouslyQueued)`

嘗試在線程的上下文中立即執行任務。
如果任務無法在當前線程執行，則返回 false，任務會被正常排隊。
用于優化場景，例如：避免線程池線程的切換。

3. `GetScheduledTasks()`

返回當前調度器排隊的任務集合（用于調試和診斷）。
由于性能原因，通常這個方法只在調試模式下實現。

13.2 默認調度器（`TaskScheduler.Default`）源碼解讀

TaskScheduler.Default 是 .NET 中的默認任務調度器，它負責將任務調度到線程池 (ThreadPool) 上執行。默認調度器的實現由 ThreadPoolTaskScheduler 類完成，該類繼承自 TaskScheduler 并封裝了線程池的任務隊列和執行邏輯，通過Task.Run啟動的任務自動由默認調度器進行調度。

默認任務調度器具有以下特點：

基于線程池：
默認調度器使用線程池來執行任務，線程池會自動管理線程的創建、銷毀以及復用，確保任務調度的高效性。
任務的公平調度：
默認調度器盡量公平地調度任務，但不支持具體的任務優先級管理。
父子任務關系：
默認調度器會自動處理父子任務關系，子任務默認繼承父任務的調度器。
支持長時間運行任務 (LongRunning)：
如果任務使用了 TaskCreationOptions.LongRunning，默認調度器會為其創建獨立的線程，而不是使用線程池線程。

`TaskScheduler.Default` 的定義

TaskScheduler.Default 是一個 ThreadPoolTaskScheduler 的單例實例，其定義如下：
TaskScheduler.cs

// AppDomain-wide默認任務調度器
private static readonly TaskScheduler s_defaultTaskScheduler = new ThreadPoolTaskScheduler();

public static TaskScheduler Default => s_defaultTaskScheduler;

s_defaultTaskScheduler 是一個靜態只讀字段，表示全局唯一的默認任務調度器實例。
ThreadPoolTaskScheduler 是默認任務調度器的核心實現。

核心實現：`ThreadPoolTaskScheduler`

ThreadPoolTaskScheduler 是默認調度器的具體實現，以下是它的核心方法分析：

ThreadPoolTaskScheduler完整源碼請參考：ThreadPoolTaskScheduler.cs

1. `QueueTask(Task task)`

QueueTask 是調度器將任務提交到線程池的入口方法。代碼如下：

protected internal override void QueueTask(Task task)
{
    TaskCreationOptions options = task.Options;
    
    if (Thread.IsThreadStartSupported && (options & TaskCreationOptions.LongRunning) != 0)
    {
        // 如果任務使用了 LongRunning 選項，則為其創建一個獨立線程
        new Thread(s_longRunningThreadWork)
        {
            IsBackground = true,
            Name = ".NET Long Running Task"
        }.UnsafeStart(task);
    }
    else
    {
        // 將普通任務提交到線程池
        ThreadPool.UnsafeQueueUserWorkItemInternal(task, (options & TaskCreationOptions.PreferFairness) == 0);
    }
}

邏輯說明：

LongRunning 任務：
如果任務設置了 TaskCreationOptions.LongRunning，調度器會為其創建一個獨立的后臺線程。這是因為長時間運行的任務可能會阻塞線程池線程，而線程池線程的數量是有限的。
普通任務：
對于沒有特殊選項的任務，調度器直接調用線程池的 UnsafeQueueUserWorkItemInternal 方法將任務提交到線程池隊列中。
公平調度：
如果任務設置了 TaskCreationOptions.PreferFairness，則會盡量保證任務按提交順序被調度（公平性）；否則，調度器可能選擇性能更優的方式來調度任務。

2. `TryExecuteTaskInline(Task task, bool taskWasPreviouslyQueued)`

TryExecuteTaskInline 方法用于嘗試在當前線程直接執行任務，而不是將任務排隊等待調度：

protected override bool TryExecuteTaskInline(Task task, bool taskWasPreviouslyQueued)
{
    // 如果任務之前已經被排隊，但現在無法從隊列中移除，則返回 false
    if (taskWasPreviouslyQueued && !ThreadPool.TryPopCustomWorkItem(task))
        return false;

    try
    {
        // 直接執行任務
        task.ExecuteEntryUnsafe(threadPoolThread: null);
    }
    finally
    {
        if (taskWasPreviouslyQueued)
            NotifyWorkItemProgress(); // 通知任務進度
    }

    return true;
}

邏輯說明：

任務從隊列移除失敗：
如果任務之前已經被排隊，但當前無法從隊列中移除（可能是因為任務已經在執行），則返回 false。
直接執行任務：
調用 task.ExecuteEntryUnsafe() 方法執行任務。
通知任務進度：
如果任務之前被排隊，則在執行完成后調用 NotifyWorkItemProgress 通知任務隊列更新。

3. `TryDequeue(Task task)`

TryDequeue 方法嘗試從任務隊列中移除一個任務：

protected internal override bool TryDequeue(Task task)
{
    return ThreadPool.TryPopCustomWorkItem(task);
}

直接調用線程池的 TryPopCustomWorkItem 方法，從任務隊列中移除指定的任務。

4. `GetScheduledTasks()`

GetScheduledTasks 返回當前調度器中已排隊的任務列表，通常用于調試：

protected override IEnumerable<Task> GetScheduledTasks()
{
    return FilterTasksFromWorkItems(ThreadPool.GetQueuedWorkItems());
}

private static IEnumerable<Task> FilterTasksFromWorkItems(IEnumerable<object> tpwItems)
{
    foreach (object tpwi in tpwItems)
    {
        if (tpwi is Task t)
        {
            yield return t;
        }
    }
}

通過線程池的 GetQueuedWorkItems 方法獲取所有已排隊的工作項，并篩選出其中的任務對象。

5. 長時間運行任務支持

對于長時間運行的任務，調度器會為其創建一個獨立線程，而不是使用線程池線程：

private static readonly ParameterizedThreadStart s_longRunningThreadWork = static s =>
{
    Debug.Assert(s is Task);
    ((Task)s).ExecuteEntryUnsafe(threadPoolThread: null);
};

使用 Thread.Start 方法創建一個獨立的后臺線程。
線程的任務執行邏輯由 Task.ExecuteEntryUnsafe 實現。

總結

TaskScheduler.Default 是 .NET 中的默認任務調度器，基于 ThreadPoolTaskScheduler 實現，其核心邏輯包括：

將普通任務提交到線程池隊列。
為長時間運行任務創建獨立線程。
支持公平調度，但不支持顯式優先級。
提供父子任務關系的自動管理。

默認調度器是大多數任務調度的首選，結合線程池提供了高效、靈活的任務執行機制。如果需要更細粒度的控制，可以通過自定義 TaskScheduler 來實現更復雜的調度邏輯。

13.3 自定義任務調度器

在 .NET 中，TaskScheduler 提供了一個靈活的機制，可以通過自定義調度器實現特殊的任務調度需求。例如，在某些場景下，你可能希望：

按任務的優先級執行任務；
強制某些任務在特定線程運行（如 UI 線程）；
控制任務的并發度；
對任務執行過程進行監控和限制。

通過繼承 TaskScheduler 并重寫其核心方法，可以實現自定義調度器來滿足這些需求。

自定義任務調度器的核心原理

自定義任務調度器需要繼承 TaskScheduler 類，并至少實現以下三個方法：

QueueTask(Task task)
將任務添加到調度器的內部隊列中，稍后進行處理。
TryExecuteTaskInline(Task task, bool taskWasPreviouslyQueued)
嘗試立即在當前線程內執行任務。如果任務不滿足條件（例如不允許在當前線程運行），則返回 false。
GetScheduledTasks()
返回當前調度器中已排隊的任務集合（用于調試和診斷）。如果不需要調試，可以拋出 NotSupportedException。

使用場景

任務優先級管理
某些任務需要根據優先級先執行，例如高優先級任務搶占低優先級任務。
任務綁定到特定線程
在某些實時性要求較高的場景下，需要將任務固定到指定線程執行，例如UI線程、游戲循環或硬件交互程序。
并發控制
限制并發任務的數量，例如數據庫操作、文件訪問等場景。
調試和任務監控
需要對任務調度的行為進行詳細的日志記錄或監控。

代碼示例 1：實現優先級任務調度器

以下代碼展示了一個基于優先級隊列的任務調度器。

using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;

public class PriorityTaskScheduler : TaskScheduler
{
    private readonly SortedList<int, Queue<Task>> _tasks = new SortedList<int, Queue<Task>>();
    private readonly Thread[] _threads;

    public PriorityTaskScheduler(int maxConcurrency)
    {
        // 限制并發線程數
        _threads = new Thread[maxConcurrency];
        for (int i = 0; i < maxConcurrency; i++)
        {
            _threads[i] = new Thread(ExecuteTasks);
            _threads[i].IsBackground = true;
            _threads[i].Start();
        }
    }

    protected override IEnumerable<Task> GetScheduledTasks()
    {
        lock (_tasks)
        {
            return _tasks.Values.SelectMany(queue => queue).ToList();
        }
    }

    protected override void QueueTask(Task task)
    {
        if (task.AsyncState is not int priority)
        {
            priority = 0; // 默認優先級
        }

        lock (_tasks)
        {
            if (!_tasks.TryGetValue(priority, out var queue))
            {
                queue = new Queue<Task>();
                _tasks[priority] = queue;
            }
            queue.Enqueue(task);
            Monitor.PulseAll(_tasks);
        }
    }

    protected override bool TryExecuteTaskInline(Task task, bool taskWasPreviouslyQueued)
    {
        // 不支持內聯執行
        return false;
    }

    private void ExecuteTasks()
    {
        while (true)
        {
            Task task = null;
            lock (_tasks)
            {
                while (!_tasks.Any())
                {
                    Monitor.Wait(_tasks);
                }

                var highestPriority = _tasks.Keys.Max();
                var queue = _tasks[highestPriority];
                task = queue.Dequeue();

                if (!queue.Any())
                {
                    _tasks.Remove(highestPriority);
                }
            }

            TryExecuteTask(task);
        }
    }
}

使用優先級調度器

class Program
{
    static void Main()
    {
        var scheduler = new PriorityTaskScheduler2);

        var factory = new TaskFactory(scheduler);

        factory.StartNew(() => Console.WriteLine("Priority 1"), 1);
        factory.StartNew(() => Console.WriteLine("Priority 3"), 3);
        factory.StartNew(() => Console.WriteLine("Priority 2"), 2);

        Thread.Sleep(1000); // 等待任務完成
    }
}

輸出示例：

Priority 3
Priority 2
Priority 1

代碼說明

優先級隊列:
使用 SortedList<int, Queue<Task>> 維護任務隊列，int 表示優先級。
多線程支持：
Thread[] 數組限制了最大并發任務數。
任務執行順序：
任務會按照優先級從高到低依次執行。

代碼示例 2：將任務綁定到特定線程

以下代碼展示了一個將任務固定到特定線程運行的調度器。

class SingleThreadTaskScheduler : TaskScheduler
{
    private readonly Thread _thread;
    private readonly BlockingCollection<Task> _tasks = new();

    public SingleThreadTaskScheduler()
    {
        _thread = new Thread(Execute);
        _thread.IsBackground = true;
        _thread.Start();
    }

    protected override void QueueTask(Task task)
    {
        _tasks.Add(task);
    }

    protected override bool TryExecuteTaskInline(Task task, bool taskWasPreviouslyQueued)
    {
        return false; // 禁止內聯執行
    }

    protected override IEnumerable<Task> GetScheduledTasks()
    {
        return _tasks.ToArray();
    }

    private void Execute()
    {
        foreach (var task in _tasks.GetConsumingEnumerable())
        {
            TryExecuteTask(task);
        }
    }
}

class Program
{
    static void Main()
    {
        var scheduler = new SingleThreadTaskScheduler();

        var factory = new TaskFactory(scheduler);

        factory.StartNew(() => Console.WriteLine($"Task 1 on thread: {Thread.CurrentThread.ManagedThreadId}"));
        factory.StartNew(() => Console.WriteLine($"Task 2 on thread: {Thread.CurrentThread.ManagedThreadId}"));

        Thread.Sleep(1000); // 等待任務完成
    }
}

輸出示例：

Task 1 on thread: 4
Task 2 on thread: 4

最佳實踐

避免復雜性： 自定義調度器需要正確處理多線程同步，盡量避免過于復雜的邏輯。
使用線程池： 除非明確需要綁定到特定線程或任務優先級，否則推薦基于線程池實現調度器。
性能測試： 自定義調度器可能帶來額外的性能開銷，應通過測試評估對任務執行性能的影響。
調試模式支持： 如果需要調試任務隊列狀態，可以實現 GetScheduledTasks 方法。

通過自定義 TaskScheduler，可以根據需求構建靈活的任務調度方案，同時避免濫用，并發編程基礎功不扎實的酌情嘗試，請始終優先考慮內置調度器是否能夠滿足需求。

13.4 并行調度 (Parallel Scheduling)

在并行編程中，任務調度器 (TaskScheduler) 通過限制并行程度和控制執行上下文，可以顯著優化資源利用率和性能。本節探討如何在并行操作中應用任務調度器，以實現高效的并行代碼調度。

并行調度的基本概念

并行調度是指在并行操作（如 Parallel.ForEach 或 Parallel.Invoke）中控制代碼執行的方式，包括以下關鍵點：

限制并行度
控制并行任務的數量，以防止資源過度消耗或競爭。
任務上下文管理
通過調度器指定任務在哪種上下文中執行（例如線程池、專用線程、或特定上下文）。
代碼片段的分層調度
在嵌套的并行操作中，為每一層提供獨立的調度策略。

通過 `ParallelOptions` 配置調度

ParallelOptions 是用于配置 Parallel 操作的核心類，支持傳入自定義任務調度器 (TaskScheduler) 和并行度限制。

代碼示例：限制并行度

以下代碼展示如何為 Parallel.ForEach 配置任務調度器和并行度限制：

using System;
using System.Collections.Generic;
using System.Threading.Tasks;

class Program
{
    static void RotateMatrices(IEnumerable<IEnumerable<Matrix>> collections, float degrees)
    {
        // 創建一個受限并發調度器，限制最大并行任務數
        var schedulerPair = new ConcurrentExclusiveSchedulerPair(
            TaskScheduler.Default, maxConcurrencyLevel: 8);

        TaskScheduler scheduler = schedulerPair.ConcurrentScheduler;

        // 配置 ParallelOptions 使用該調度器
        ParallelOptions options = new ParallelOptions
        {
            TaskScheduler = scheduler
        };

        // 對矩陣集合執行嵌套的并行操作
        Parallel.ForEach(collections, options, matrices =>
        {
            Parallel.ForEach(matrices, options, matrix =>
            {
                matrix.Rotate(degrees);
            });
        });
    }

    static void Main()
    {
        // 示例矩陣集合
        var collections = new List<List<Matrix>>
        {
            new List<Matrix> { new Matrix(), new Matrix() },
            new List<Matrix> { new Matrix() }
        };

        RotateMatrices(collections, 90.0f);

        Console.WriteLine("Matrix rotation completed.");
    }
}

class Matrix
{
    public void Rotate(float degrees)
    {
        Console.WriteLine($"Rotating matrix by {degrees} degrees on thread {Environment.CurrentManagedThreadId}.");
    }
}

輸出示例：

Rotating matrix by 90 degrees on thread 5.
Rotating matrix by 90 degrees on thread 7.
...
Matrix rotation completed.

說明

限制并行任務數：maxConcurrencyLevel: 8 限制最多同時執行 8 個任務。
嵌套調度：外層和內層的 Parallel.ForEach 都使用相同的調度器。
調度器作用范圍：通過 ParallelOptions 傳遞給 Parallel 方法。

ConcurrentExclusiveSchedulerPair是.Net運行時提供的并發/排他調度器對，提供兩個調度器：一個支持并發任務執行（ConcurrentScheduler），另一個支持排他性任務執行（ExclusiveScheduler），用于高效地管理需要并發與互斥執行的任務場景。

在 `Parallel.Invoke` 中使用調度器

與 Parallel.ForEach 類似，Parallel.Invoke 也支持接收 ParallelOptions 來配置任務調度器。

代碼示例：使用自定義調度器

ParallelOptions options = new ParallelOptions
{
    TaskScheduler = scheduler
};

Parallel.Invoke(options,
    () => Console.WriteLine("Task 1 executed."),
    () => Console.WriteLine("Task 2 executed."));

動態并行中的調度器應用

動態并行指任務在運行過程中生成更多任務。例如，可以將調度器直接傳遞給 Task.Factory.StartNew 或 Task.ContinueWith，確保動態生成的任務使用同一調度器。

代碼示例：動態任務調度

var schedulerPair = new ConcurrentExclusiveSchedulerPair(TaskScheduler.Default, maxConcurrencyLevel: 4);
TaskScheduler scheduler = schedulerPair.ConcurrentScheduler;

Task.Factory.StartNew(() =>
{
    Console.WriteLine($"Task 1 running on thread {Environment.CurrentManagedThreadId}");
    Task.Factory.StartNew(() =>
    {
        Console.WriteLine($"Nested task running on thread {Environment.CurrentManagedThreadId}");
    }, TaskCreationOptions.None, scheduler);
}, TaskCreationOptions.None, scheduler).Wait();

限制：PLINQ 與任務調度器

需要注意的是，任務調度器無法直接應用于 PLINQ（并行 LINQ）。PLINQ 使用自己的并行調度機制，無法通過 ParallelOptions 配置。

13.5 使用調度器實現數據流同步

問題背景

假設需要在數據流代碼中控制獨立代碼片段的執行方式，例如在特定線程（如 UI TaskScheduler 允許我們對數據流塊的執行進行更精細的控制。

解決方案

在 .NET 中，可以通過 ExecutionDataflowBlockOptions 為數據流塊指定一個 TaskScheduler 實例。這樣，數據流塊會在指定的調度器上調度和執行任務。以下示例展示了如何在數據流網格的不同部分中使用調度器來控制任務的執行上下文。

代碼示例1：在 WinForms 中使用數據流塊更新 UI

以下代碼創建了一個簡單的數據流網格：

multiplyBlock：將輸入值乘以 2，在線程池中執行。
displayBlock：將結果添加到 UI 的 ListBox 中，在 UI 線程中執行。

using System;
using System.Threading.Tasks;
using System.Threading.Tasks.Dataflow;
using System.Windows.Forms;

public class MainForm : Form
{
    private Button addButton;
    private ListBox resultListBox;
    private TransformBlock<int, int> multiplyBlock;
    private ActionBlock<int> displayBlock;
    private int counter = 1;

    public MainForm()
    {
        // 初始化控件
        this.Text = "TaskScheduler 數據流示例";
        this.Width = 400;
        this.Height = 300;

        addButton = new Button
        {
            Text = "添加數據",
            Dock = DockStyle.Top,
            Height = 50
        };
        addButton.Click += OnAddButtonClick;

        resultListBox = new ListBox
        {
            Dock = DockStyle.Fill
        };

        this.Controls.Add(resultListBox);
        this.Controls.Add(addButton);

        // 配置數據流塊
        ConfigureDataflowBlocks();
    }

    private void OnAddButtonClick(object sender, EventArgs e)
    {
        // 每次點擊按鈕，向 multiplyBlock 發送一個數字
        multiplyBlock.Post(counter++);
    }

    private void ConfigureDataflowBlocks()
    {
        // 創建 multiplyBlock：在后臺線程中將輸入值乘以 2
        multiplyBlock = new TransformBlock<int, int>(item =>
        {
            Task.Delay(500).Wait(); // 模擬耗時操作
            return item * 2;
        });

        // 創建 displayBlock：在 UI 線程中更新 ListBox
        var uiOptions = new ExecutionDataflowBlockOptions
        {
            TaskScheduler = TaskScheduler.FromCurrentSynchronizationContext() // 確保在 UI 線程中執行
        };

        displayBlock = new ActionBlock<int>(result =>
        {
            resultListBox.Items.Add($"Processed: {result}"); // 更新 ListBox
        }, uiOptions);

        // 鏈接數據流塊
        multiplyBlock.LinkTo(displayBlock, new DataflowLinkOptions { PropagateCompletion = true });
    }

    protected override void OnFormClosing(FormClosingEventArgs e)
    {
        // 關閉窗口時標記數據流結束
        multiplyBlock.Complete();
        displayBlock.Completion.Wait(); // 確保所有任務完成
        base.OnFormClosing(e);
    }

    [STAThread]
    public static void Main()
    {
        Application.EnableVisualStyles();
        Application.Run(new MainForm());
    }
}

代碼解析

UI 控件初始化：
- 創建一個按鈕 addButton，用于觸發任務。
- 創建 ListBox，用于展示處理結果。
數據流塊配置：
- multiplyBlock：
  執行一個模擬耗時的計算任務（如將輸入值乘以 2），在后臺線程（默認線程池）中運行。
- displayBlock：
  通過 ExecutionDataflowBlockOptions 指定 TaskScheduler.FromCurrentSynchronizationContext()，確保在 UI 線程中執行任務，安全地更新 ListBox。
數據流塊鏈接：
- 使用 LinkTo 將 multiplyBlock 的輸出連接到 displayBlock，實現數據在塊之間的流動。
- 設置 PropagateCompletion = true，確保任務完成時會正確關閉后續的數據流塊。
按鈕事件處理：
- 每次點擊按鈕時，向 multiplyBlock 發送新的數據（遞增的整數）。
窗口關閉處理：
- 在關閉窗口時，調用 Complete() 標記數據流結束，并等待所有任務完成，避免后臺任務未完成時強制退出的潛在問題。

代碼示例2：同步與排他調度的結合

當需要協調數據流網格中不同塊的執行行為時，TaskScheduler 的作用尤為重要。例如，可以利用 ConcurrentExclusiveSchedulerPair 來確保某些塊不會同時執行，而其他塊可以隨時執行。

代碼示例：使用 ConcurrentExclusiveSchedulerPair 控制數據流塊的執行

using System;
using System.Threading.Tasks;
using System.Threading.Tasks.Dataflow;

class Program
{
    static void Main()
    {
        // 創建調度器對：互斥調度器和并發調度器
        var schedulerPair = new ConcurrentExclusiveSchedulerPair();
        var exclusiveScheduler = schedulerPair.ExclusiveScheduler;
        var concurrentScheduler = schedulerPair.ConcurrentScheduler;

        // A 塊：使用互斥調度器
        var blockA = new ActionBlock<int>(
            item => Console.WriteLine($"Block A processing {item}"),
            new ExecutionDataflowBlockOptions { TaskScheduler = exclusiveScheduler });

        // B 塊：使用并發調度器
        var blockB = new ActionBlock<int>(
            item => Console.WriteLine($"Block B processing {item}"),
            new ExecutionDataflowBlockOptions { TaskScheduler = concurrentScheduler });

        // C 塊：使用互斥調度器
        var blockC = new ActionBlock<int>(
            item => Console.WriteLine($"Block C processing {item}"),
            new ExecutionDataflowBlockOptions { TaskScheduler = exclusiveScheduler });

        // 向數據流塊發送數據
        for (int i = 1; i <= 5; i++)
        {
            blockA.Post(i);
            blockB.Post(i);
            blockC.Post(i);
        }

        // 標記完成
        blockA.Complete();
        blockB.Complete();
        blockC.Complete();

        Task.WaitAll(blockA.Completion, blockB.Completion, blockC.Completion);
    }
}

代碼說明：

exclusiveScheduler：確保 blockA 和 blockC 互斥執行，即同一時間只能有一個塊在處理數據。
concurrentScheduler：允許 blockB 并發執行，不受 blockA 和 blockC 的限制。
協調行為：
當 blockA 或 blockC 在運行時，另一塊會等待，而 blockB 可以隨時執行。

輸出示例：

Block A processing 1
Block B processing 1
Block B processing 2
Block C processing 1
Block A processing 2
Block C processing 2
Block A processing 3
Block B processing 3
Block C processing 3
...

注意事項

異步代碼的調度限制：
通過 TaskScheduler 實現的同步僅對代碼的執行部分生效。如果數據流塊中執行的是異步代碼（例如 await 操作），任務在等待時并不被視為正在執行，因此調度器的限制無法完全約束其行為。
數據流塊的內部任務：
即使某些數據流塊本身不執行代碼（如 BufferBlock<T>），它們也需要處理內部任務，這些任務會使用指定的 TaskScheduler 執行。
上下文恢復：
在使用數據流塊時，盡量通過 TaskScheduler.FromCurrentSynchronizationContext 或 ExecutionDataflowBlockOptions 來顯式指定需要恢復的上下文，避免潛在的同步上下文丟失問題。

13.6 同步上下文 (SynchronizationContext)

SynchronizationContext 是 .NET 中一個重要的抽象，定義了如何調度代碼回到某個特定的上下文（如 UI 線程或請求線程），這一點跟TaskScheduler有幾分相似。它在并行、異步等編程場景中扮演著非常重要的角色，尤其是 async/await 操作背后的機制。

`SynchronizationContext` 的核心概念

抽象的調度上下文
它是一個抽象類，允許不同平臺和框架定義自己的實現，如 WPF 的 DispatcherSynchronizationContext 和 ASP.NET 的 AspNetSynchronizationContext。
線程與上下文分離
同步上下文并不直接等同于線程，它表示一個調度環境，調度代碼如何執行，在那個線程上執行。例如，UI 的同步上下文會將代碼調度回 UI 線程。
與異步的關系
默認情況下，async/await 會捕獲當前線程的 SynchronizationContext 并在異步操作完成后恢復到此上下文，這一點在后續章節async/await原理解析中會詳細介紹。

同步上下文的常見實現

UI 應用程序
- WPF、Windows Forms、UWP 等框架提供了專門的同步上下文，用于在異步操作后調度回 UI 線程。
- 示例：DispatcherSynchronizationContext（WPF）、WindowsFormsSynchronizationContext（WinForms）。
ASP.NET 請求上下文
- 在傳統的 ASP.NET 中，每個 HTTP 請求有其獨立的同步上下文AspNetSynchronizationContext，用于確保操作回到原始請求線程。
默認上下文
- 控制臺應用和后臺任務沒有特定的同步上下文，通過SynchronizationContext.Current會返回一個null。

示例 1：UI 應用中的同步上下文

以下代碼展示如何利用 SynchronizationContext 在異步任務完成后返回到 UI 線程更新界面。

using System;
using System.Threading;
using System.Threading.Tasks;
using System.Windows.Forms;

class Program
{
    static async Task Main()
    {
        Application.EnableVisualStyles();

        var form = new Form();
        var label = new Label { Text = "Initial text", Dock = DockStyle.Fill };
        form.Controls.Add(label);

        form.Shown += async (sender, e) =>
        {
            // 拿到 UI 的同步上下文
            var syncContext = SynchronizationContext.Current;

            await Task.Run(() =>
            {
                // 在后臺線程中運行
                Thread.Sleep(2000);
                Console.WriteLine($"Background thread: {Environment.CurrentManagedThreadId}");

                // 通過syncContext將操作調度回 UI 線程
                syncContext.Post(_ =>
                {
                    // 如下代碼會在 UI 線程中執行
                    Console.WriteLine($"UI thread: {Environment.CurrentManagedThreadId}");
                    label.Text = "Updated text";
                }, null);
            });
        };

        Application.Run(form);
    }
}

輸出示例：

Background thread: 5
UI thread: 1

說明

當執行new Form()時候，會自動為當前線程安裝一個WindowsFormsSynchronizationContext上下文，Form的基類Control的構造函數中：

/// <summary>
///  Initializes a new instance of the <see cref="Control"/> class.
/// </summary>
public Control() : this(true)
{
}

internal Control(bool autoInstallSyncContext) : base()
{
    // 初始化工作
    ...
    // 為當前線程設置同步上下文為`WindowsFormsSynchronizationContext`.
    if (autoInstallSyncContext)
    {
        WindowsFormsSynchronizationContext.InstallIfNeeded();
    }
}

SynchronizationContext.Current 捕獲當前 UI 上下文。
syncContext.Post 將代碼調度回 UI 線程。
在 UI 應用中，async/await 會自動處理這個過程，無需手動調用 SynchronizationContext。

示例 2：后臺服務中的同步上下文

控制臺應用或后臺服務默認沒有特定的同步上下文。在這種情況下，await 會回到線程池線程，而不是原始線程。

using System;
using System.Threading;
using System.Threading.Tasks;

class Program
{
    static async Task Main()
    {
        Console.WriteLine($"Main thread: {Environment.CurrentManagedThreadId}");

        await Task.Run(() =>
        {
            Console.WriteLine($"Task running on thread: {Environment.CurrentManagedThreadId}");
        });

        Console.WriteLine($"Back to thread: {Environment.CurrentManagedThreadId}");
    }
}

輸出示例：

Main thread: 1
Task running on thread: 4
Back to thread: 4

說明

控制臺應用默認沒有同步上下文，異步操作不會回到原始線程。
可以通過設置自定義的 SynchronizationContext 來實現類似 UI 應用的行為。

自定義同步上下文

有時需要為特定的場景實現自定義調度邏輯，可以通過繼承 SynchronizationContext 來完成。

最佳實踐

UI 應用
- 避免手動使用 SynchronizationContext，優先使用 async/await 自動處理上下文切換。
- 確保耗時任務運行在線程池上，避免阻塞 UI 線程。
后臺服務
- 確保理解沒有同步上下文的行為，必要時可引入自定義上下文以實現特定的調度邏輯。
避免綁定到平臺特性
- 使用通用的 SynchronizationContext 抽象，而非特定平臺的調度實現（如 WPF 的 Dispatcher）。
性能優化
- 在性能關鍵場景下，通過設置 ConfigureAwait(false) 來避免捕獲上下文，提高異步代碼的運行效率。

SynchronizationContext 是管理代碼調度和線程上下文的強大工具，但需要合理使用。通過理解其機制和常見場景，可以有效編寫高效、穩定的異步程序。

13.7 自定義同步上下文

上一節對同步上下文有個基本介紹，這一節整幾個有特殊功能的自定義同步上下文加深理解。

自定義 SynchronizationContext 可以用來解決特殊場景下的線程調度需求，例如實現任務的優先級管理、針對特定線程的任務調度或處理消息隊列等。以下是幾個更實用的 SynchronizationContext 示例，它們展示了如何在不同場景中擴展或替換默認行為。

通過 SynchronizationContext.SetSynchronizationContext可以設置當前線程的同步上下文，并通過SynchronizationContext.Current返回線程的同步上下文實例

示例代碼1：將任務調度到特定線程

在某些場景中，你可能需要將所有任務調度到一個指定的線程（例如專用的后臺線程）進行處理。

using System;
using System.Collections.Concurrent;
using System.Threading;
using System.Threading.Tasks;

class SingleThreadSynchronizationContext : SynchronizationContext
{
    private readonly Thread _workerThread;
    private readonly BlockingCollection<(SendOrPostCallback, object?)> _taskQueue = new();

    public SingleThreadSynchronizationContext()
    {
        _workerThread = new Thread(() =>
        {
            while (true)
            {
                var task = _taskQueue.Take();
                task.Item1(task.Item2);
            }
        })
        {
            IsBackground = true
        };
        _workerThread.Start();
    }

    public override void Post(SendOrPostCallback d, object? state)
    {
        _taskQueue.Add((d, state));
    }

    public override void Send(SendOrPostCallback d, object? state)
    {
        if (Thread.CurrentThread == _workerThread)
        {
            d(state);
        }
        else
        {
            using var doneEvent = new ManualResetEvent(false);
            _taskQueue.Add((s =>
            {
                d(s);
                doneEvent.Set();
            }, state));
            doneEvent.WaitOne();
        }
    }
}

class Program
{
    static async Task Main()
    {
        var syncContext = new SingleThreadSynchronizationContext();
        SynchronizationContext.SetSynchronizationContext(syncContext);

        Console.WriteLine($"Main thread: {Thread.CurrentThread.ManagedThreadId}");

        await Task.Run(() => Console.WriteLine($"Task thread: {Thread.CurrentThread.ManagedThreadId}"));

        SynchronizationContext.Current!.Post(_ =>
        {
            Console.WriteLine($"Posted to single thread: {Thread.CurrentThread.ManagedThreadId}");
        }, null);

        await Task.Delay(1000);
    }
}

實現要點

使用 BlockingCollection 實現任務隊列。
在 Post 方法中將任務添加到隊列，后臺線程逐個執行任務。
在 Send 方法中確保同步調用任務。
通過 SynchronizationContext.SetSynchronizationContext(syncContext)將當前線程的同步上下文設置為syncContext，并通過SynchronizationContext.Current返回設置的同步上下文實例syncContext

使用場景

在游戲開發中，調度任務到一個固定線程以操作游戲狀態或渲染。
為資源密集型任務提供專用線程。

示例代碼2：支持優先級的同步上下文

在某些場景中，任務可能需要按照優先級執行，例如高優先級任務應該先于低優先級任務運行。

using System;
using System.Collections.Concurrent;
using System.Linq;
using System.Threading;

class PrioritySynchronizationContext : SynchronizationContext
{
    private readonly Thread _workerThread;
    private readonly BlockingCollection<(int Priority, SendOrPostCallback Callback, object? State)> _taskQueue 
        = new(new PriorityComparer());

    public PrioritySynchronizationContext()
    {
        _workerThread = new Thread(() =>
        {
            foreach (var task in _taskQueue.GetConsumingEnumerable())
            {
                task.Callback(task.State);
            }
        })
        {
            IsBackground = true
        };
        _workerThread.Start();
    }

    public override void Post(SendOrPostCallback d, object? state)
    {
        Post(d, state, 0);
    }

    public void Post(SendOrPostCallback d, object? state, int priority)
    {
        _taskQueue.Add((priority, d, state));
    }

    private class PriorityComparer : IComparer<(int Priority, SendOrPostCallback, object?)>
    {
        public int Compare((int Priority, SendOrPostCallback, object?) x, (int Priority, SendOrPostCallback, object?) y)
        {
            return y.Priority.CompareTo(x.Priority);
        }
    }
}

class Program
{
    static void Main()
    {
        var syncContext = new PrioritySynchronizationContext();
        SynchronizationContext.SetSynchronizationContext(syncContext);

        Console.WriteLine("Scheduling tasks...");
        syncContext.Post(_ => Console.WriteLine("Task 1: Low Priority"), null, priority: 1);
        syncContext.Post(_ => Console.WriteLine("Task 2: High Priority"), null, priority: 10);
        syncContext.Post(_ => Console.WriteLine("Task 3: Medium Priority"), null, priority: 5);

        Thread.Sleep(1000); // Allow background thread to process tasks
    }
}

實現要點

使用 BlockingCollection 和自定義 IComparer 實現優先級隊列。
擴展 Post 方法以支持優先級參數。
后臺線程從高優先級任務開始依次處理。

使用場景

任務調度需要嚴格按照優先級順序執行，例如實時系統或多任務操作系統模擬。
高優先級任務（如報警信號）優先于低優先級任務。

示例代碼3：模擬異步消息處理的同步上下文

在消息驅動的系統中，可以使用同步上下文將消息處理調度到一個模擬的事件循環中。

using System;
using System.Collections.Concurrent;
using System.Threading;
using System.Threading.Tasks;

class EventLoopSynchronizationContext : SynchronizationContext
{
    private readonly BlockingCollection<SendOrPostCallback> _queue = new();
    private readonly Thread _loopThread;

    public EventLoopSynchronizationContext()
    {
        _loopThread = new Thread(() =>
        {
            foreach (var callback in _queue.GetConsumingEnumerable())
            {
                callback(null);
            }
        })
        {
            IsBackground = true
        };
        _loopThread.Start();
    }

    public override void Post(SendOrPostCallback d, object? state)
    {
        _queue.Add(s => d(state));
    }

    public Task PostAsync(Func<Task> func)
    {
        var tcs = new TaskCompletionSource();
        Post(async _ =>
        {
            try
            {
                await func();
                tcs.SetResult();
            }
            catch (Exception ex)
            {
                tcs.SetException(ex);
            }
        }, null);
        return tcs.Task;
    }
}

class Program
{
    static async Task Main()
    {
        var syncContext = new EventLoopSynchronizationContext();
        SynchronizationContext.SetSynchronizationContext(syncContext);

        Console.WriteLine("Posting async tasks...");
        await syncContext.PostAsync(async () =>
        {
            Console.WriteLine($"Task 1 start on thread: {Thread.CurrentThread.ManagedThreadId}");
            await Task.Delay(1000);
            Console.WriteLine("Task 1 complete");
        });

        await syncContext.PostAsync(async () =>
        {
            Console.WriteLine($"Task 2 start on thread: {Thread.CurrentThread.ManagedThreadId}");
            await Task.Delay(500);
            Console.WriteLine("Task 2 complete");
        });
    }
}

實現要點

使用 BlockingCollection 作為消息隊列。
通過 PostAsync 方法實現對異步操作的支持。
主線程模擬消息循環，處理異步消息。

使用場景

模擬事件驅動系統或消息總線。
異步任務需要有序地逐個執行，例如在后臺線程中實現隊列式的任務處理。

總結

自定義同步上下文	關鍵功能	使用場景
`SingleThreadSynchronizationContext`	調度任務到指定線程	游戲開發、專用線程處理
`PrioritySynchronizationContext`	按任務優先級調度	實時系統、優先級任務
`EventLoopSynchronizationContext`	異步消息處理	消息驅動系統、事件總線

這些示例展示了如何通過自定義同步上下文應對特定場景中的調度需求。根據實際需求，可以進一步調整任務處理的策略和方式。

13.7 線程池 (ThreadPool) 和調度策略

線程池通過復用線程來減少線程創建和銷毀的開銷。在現代并發編程中，線程池在任務調度和執行中起著核心作用。本節主要理解線程池的默認行為、調度策略。在不了解線程池工作原理的情況下，不建議手動優化線程池，默認情況下已經是經過了反復調優。

1. 線程池的基本概念

線程池是由系統管理的一組預先創建或動態分配的線程，用于執行短期任務。線程池的主要特點包括：

復用線程：減少線程創建和銷毀的開銷。
動態擴展：根據工作負載動態調整線程的數量。
任務調度：線程池內置了任務排隊機制，以合理分配 CPU 資源。

2. 線程池的核心結構

線程池是由以下幾個主要組件組成的：

工作線程（Worker Threads）：
- 用于執行普通的 CPU 密集型任務。
- 線程池會自動管理工作線程的創建、銷毀和復用。
I/O 線程（I/O Completion Port Threads）：
- 專門處理異步 I/O 操作（如文件操作、網絡請求等）。
- 這些線程由操作系統內核管理，通過 I/O 完成端口（I/O Completion Port，簡稱 IOCP）實現高效的異步完成通知。
任務隊列（Task Queue）：
- 每個任務有一個單獨的任務隊列。
- 線程池維護一個全局的任務隊列，用于存儲待執行的任務。
- 如果當前沒有空閑線程，任務會被暫時放入隊列，等待調度。

3. 工作線程（Worker Thread）的調度機制

任務提交與處理

任務提交：
- 當調用 ThreadPool.QueueUserWorkItem 或 Task.Run 時，任務被添加到線程池的任務隊列中。
- 如果有空閑線程，線程池會立即從隊列中取出任務并執行。
- 如果沒有空閑線程，任務會在隊列中等待，直到線程池創建新線程或釋放現有線程。
線程池的線程復用：
- 線程池會嘗試復用已有的線程來執行任務，而不是每次都創建新線程，從而減少線程創建和銷毀的開銷。

線程擴展與收縮

線程擴展：
- 當任務量超過線程池當前的線程數量時，線程池會動態增加線程。
- 線程池的擴展速度是受限制的，默認每秒最多增加 2 個線程，以避免過度消耗系統資源。
線程回收：
- 如果線程長時間（默認 10 秒）沒有處理任務，線程池會回收這些空閑線程以減少資源占用。
- 回收的線程并不會被銷毀，而是進入一個備用狀態，等待再次被復用。

4. I/O 線程與異步操作的調度機制

I/O 完成端口（IOCP）

.NET 的線程池通過操作系統的 I/O Completion Port (IOCP) 高效地管理異步 I/O 操作：

異步任務提交：
- 當應用程序發起異步 I/O 操作（如 FileStream.ReadAsync 或網絡請求）時，操作系統會將任務提交到 IOCP，此時并沒有任何線程去等待這個操作，詳情參考There Is No Thread。
I/O 異步完成：
- 操作系統會在 I/O 操作完成時，將完成通知放入 IOCP 隊列，并喚醒一個 I/O 線程來處理完成事件此時才有線程去使用 I/O 操作的結果。
I/O 線程池管理：
- .NET 為 IOCP 提供了專門的線程池（即 I/O 線程池），以高效處理異步 I/O 操作的完成回調。

I/O 線程與工作線程的區別

特性	工作線程	I/O 線程
用途	執行普通任務	處理異步 I/O 操作的完成事件
來源	由 .NET 管理	由操作系統 IOCP 提供
線程數量	可配置（默認動態擴展）	系統自動管理
使用場景	CPU 密集型任務	網絡、文件等異步 I/O 操作

5. 任務隊列的實現

線程池的任務隊列是一個線程安全的結構，用于存儲待執行的任務。它的實現主要依賴于以下機制：

全局任務隊列：
- 線程池維護一個全局的任務隊列，用于存儲所有未被處理的任務。
- 當工作線程空閑時，會從全局隊列中取任務執行。
局部任務隊列（Work-Stealing Queue）：
- 每個線程還維護一個局部任務隊列，用于存儲本線程的任務。
- 如果一個線程的局部任務隊列為空，它可以嘗試從其他線程的隊列中“竊取”（Work-Stealing）任務執行。
任務調度：
- 線程池優先處理局部隊列中的任務，以減少線程間競爭。
- 如果局部隊列為空，則會從全局任務隊列中取任務。
- 如果全局任務隊列也為空，線程會進入等待狀態。

任務隊列的優勢

通過局部任務隊列和 Work-Stealing 機制，線程池能有效減少線程間的任務爭用，提高任務調度的效率。
任務隊列的實現是線程安全的，確保多線程環境下的任務提交與調度不會出現數據競爭問題。

6. 線程池默認行為

1. 默認配置

線程池線程默認是后臺線程，隨進程退出而銷毀。
使用工作隊列管理任務，任務會按先進先出（FIFO）順序調度。
最大線程數和最小線程數的默認值與操作系統和 .NET 運行時相關：
- 默認最小線程數：CPU 核心數。
- 默認最大線程數：約 32,767（具體值視運行時而定）。

2. 動態擴展策略

如果線程池中的線程不夠用，它會動態創建線程。
新線程的創建有一定延遲，避免因短期任務暴增而頻繁創建銷毀線程。

3. 工作線程與 IO 線程

工作線程：處理 CPU 密集型任務。
IO 線程：處理異步 IO 完成回調。

7. 調整線程池行為

1. 設置最小線程數

通過設置最小線程數，可以減少線程池在高負載下的擴展延遲。

ThreadPool.SetMinThreads(workerThreads: 8, completionPortThreads: 8);

2. 獲取線程池狀態

使用 ThreadPool.GetMinThreads 和 ThreadPool.GetMaxThreads 獲取線程池配置：

ThreadPool.GetMinThreads(out int workerMin, out int ioMin);
ThreadPool.GetMaxThreads(out int workerMax, out int ioMax);

Console.WriteLine($"Min Threads: Worker={workerMin}, IO={ioMin}");
Console.WriteLine($"Max Threads: Worker={workerMax}, IO={ioMax}");

3. 調整最大線程數

可以通過 ThreadPool.SetMaxThreads 限制線程池的并發任務數量：

ThreadPool.SetMaxThreads(workerThreads: 16, completionPortThreads: 16);

8. 線程池的優缺點

優點

線程復用： 減少線程創建和銷毀的開銷。
自動管理： 動態調整線程數以適應任務負載。
線程安全： 內部實現了高效的任務隊列和調度機制。
I/O 異步支持： 通過 IOCP 高效處理異步操作。

缺點

不適合長時間任務： 長時間運行的任務會阻塞線程池線程，影響其他任務的執行。
調度延遲： 如果任務過多且線程池擴展速度跟不上，可能出現任務調度延遲。

9. `ThreadPool.QueueUserWorkItem` 與 `Task.Run` 的差異

1. 使用方式和目標

特性	`ThreadPool.QueueUserWorkItem`	`Task.Run`
抽象級別	低級別，直接向線程池提交任務	高級別，基于線程池實現的任務抽象
返回值	無返回值，無法捕獲任務的結果	返回 `Task` 對象，可跟蹤任務狀態
異常處理	異常會導致程序崩潰	支持異常捕獲和處理
適用場景	輕量級任務，無需結果或復雜管理	需要更強大的任務管理功能，如鏈式任務

2. 示例代碼

使用 ThreadPool.QueueUserWorkItem 提交任務：

using System;
using System.Threading;

class Program
{
    static void Main()
    {
        ThreadPool.QueueUserWorkItem(_ =>
        {
            Console.WriteLine($"ThreadPool Task: {Thread.CurrentThread.ManagedThreadId}");
        });

        Console.WriteLine("Main Thread Completed");
        Thread.Sleep(100); // 等待任務完成
    }
}

使用 Task.Run 提交任務：

using System;
using System.Threading.Tasks;

class Program
{
    static async Task Main()
    {
        var task = Task.Run(() =>
        {
            Console.WriteLine($"Task Run: {Task.CurrentId} on Thread {Thread.CurrentThread.ManagedThreadId}");
        });

        await task;
        Console.WriteLine("Main Task Completed");
    }
}

關鍵區別：

ThreadPool.QueueUserWorkItem 更適合簡單的、無需跟蹤的任務。
Task.Run 提供更高級的任務跟蹤和組合能力，推薦在現代并發代碼中使用。

若要將某個代碼調度到線程池線程中執行，推薦使用Task.Run，不要再使用ThreadPool.QueueUserWorkItem了，已經過時了

總結

線程池是 .NET 中一個高效的線程管理工具，其底層機制包括：

線程復用與動態擴展： 減少線程創建銷毀開銷，動態調整線程數。
任務隊列與調度： 通過全局隊列和局部隊列的 Work-Stealing 機制提高效率。
I/O 完成端口： 高效處理異步 I/O 操作。
線程回收： 空閑線程在一定時間后被回收，節省資源。

posted @ 2024-12-09 16:18 平元兄閱讀(386) 評論(0) 收藏舉報

刷新頁面返回頂部

Michael Shen

平元兄

第十三章：調度

第十三章：調度

13.1 任務調度器 (TaskScheduler) 基礎

TaskScheduler 的作用與核心機制

默認任務調度器 TaskScheduler.Default

使用默認調度器將任務調度到線程池線程上

代碼示例：Task.Run 的使用

代碼示例：Task.Factory.StartNew 的使用

捕獲與恢復同步上下文

TaskScheduler.FromCurrentSynchronizationContext

SynchronizationContext 的作用

TaskScheduler 的核心方法

1. QueueTask(Task task)

2. TryExecuteTaskInline(Task task, bool taskWasPreviouslyQueued)

3. GetScheduledTasks()

13.2 默認調度器（TaskScheduler.Default）源碼解讀

TaskScheduler.Default 的定義

核心實現：ThreadPoolTaskScheduler

1. QueueTask(Task task)

2. TryExecuteTaskInline(Task task, bool taskWasPreviouslyQueued)

3. TryDequeue(Task task)

4. GetScheduledTasks()

5. 長時間運行任務支持

總結

13.3 自定義任務調度器

自定義任務調度器的核心原理

使用場景

代碼示例 1：實現優先級任務調度器

使用優先級調度器

代碼說明

代碼示例 2：將任務綁定到特定線程

最佳實踐

13.4 并行調度 (Parallel Scheduling)

并行調度的基本概念

通過 ParallelOptions 配置調度

代碼示例：限制并行度

說明

在 Parallel.Invoke 中使用調度器

代碼示例：使用自定義調度器

動態并行中的調度器應用

代碼示例：動態任務調度

限制：PLINQ 與任務調度器

13.5 使用調度器實現數據流同步

問題背景

解決方案

代碼示例1：在 WinForms 中使用數據流塊更新 UI

代碼解析

代碼示例2：同步與排他調度的結合

注意事項

13.6 同步上下文 (SynchronizationContext)

SynchronizationContext 的核心概念

同步上下文的常見實現

示例 1：UI 應用中的同步上下文

說明

示例 2：后臺服務中的同步上下文

說明

自定義同步上下文

最佳實踐

13.7 自定義同步上下文

示例代碼1：將任務調度到特定線程

實現要點

使用場景

示例代碼2：支持優先級的同步上下文

實現要點

使用場景

示例代碼3：模擬異步消息處理的同步上下文

實現要點

使用場景

總結

13.7 線程池 (ThreadPool) 和調度策略

1. 線程池的基本概念

2. 線程池的核心結構

3. 工作線程（Worker Thread）的調度機制

任務提交與處理

線程擴展與收縮

4. I/O 線程與異步操作的調度機制

I/O 完成端口（IOCP）

I/O 線程與工作線程的區別

`TaskScheduler` 的作用與核心機制

默認任務調度器 `TaskScheduler.Default`

代碼示例：`Task.Run` 的使用

代碼示例：`Task.Factory.StartNew` 的使用

`TaskScheduler.FromCurrentSynchronizationContext`

`TaskScheduler` 的核心方法

1. `QueueTask(Task task)`

2. `TryExecuteTaskInline(Task task, bool taskWasPreviouslyQueued)`

3. `GetScheduledTasks()`

13.2 默認調度器（`TaskScheduler.Default`）源碼解讀

`TaskScheduler.Default` 的定義

核心實現：`ThreadPoolTaskScheduler`

1. `QueueTask(Task task)`

2. `TryExecuteTaskInline(Task task, bool taskWasPreviouslyQueued)`

3. `TryDequeue(Task task)`

4. `GetScheduledTasks()`

通過 `ParallelOptions` 配置調度

在 `Parallel.Invoke` 中使用調度器

`SynchronizationContext` 的核心概念

9. `ThreadPool.QueueUserWorkItem` 與 `Task.Run` 的差異