Channel: 生產者–消費者 ( Channel)

1. 介紹

今天我們要升個級！來認識一個很特別的工具 — Channel。這玩意是為現代 .NET 的非同步應用設計的：當傳統的鎖機制救不了你，或會讓你性能大幅掉落時，Channel 就派上用場。

你會看到「生產者-消費者」(Producer-Consumer) 模式的實作，這個模式從 20 世紀 60 年代就一直很常用。你會建一個簡單的非同步「輸送帶」，有些 tasks 或執行緒負責生產（例如下載檔案、計算數字、等待事件），另一端負責消耗處理（例如存檔、寫資料庫、更新 UI）。

為什麼會有 Channel？

1. 「生產者-消費者」模式 傳統上用隊列來解決：生產者把任務放進隊列，消費者從隊列取出。但！BlockingCollection<T>、基於 ConcurrentQueue<T> 的實作，甚至用 lock 手動同步——這些都不是非同步的。也就是說執行緒會被阻塞等待資料，而不是把控制權交還給 async/await 的排程器。
2. 非同步 在 .NET 裡不是流行語，而是現代架構的基石。為了等待元素而去阻塞執行緒代價高且低效。我們需要能在不阻塞的情況下等待資料出現——這正是 Channel 要解決的事。
3. 彈性：用通道可以搭出複雜的處理 pipeline，拆分工作、加中間步驟、做負載平衡——而不用去寫那些低階的同步鎖邏輯。

什麼是 Channel？（比喻與架構）

想像你有接力棒（或輸送帶），可以把物件從一端傳到另一端，彼此之間不需要面對面。重點是：接力棒在路上不要丟就好。

Channel 是 .NET 裡用來在不同 tasks、執行緒或程式部分之間做 非同步資料傳遞 的內建工具。它實作了一個支援非同步「等待」的隊列，既能等待寫入也能等待讀取。

• 生產者 把元素放進 channel（例如處理請求）；
• 消費者 拿出元素——事情就完成了！

2. 類別 Channel<T> 與結構

一切從命名空間開始：

using System.Threading.Channels;

和大家熟悉的集合不太一樣，Channel 更像一個 工廠，用來建立專門用於傳遞資料的物件。

主要類型：

• ChannelWriter<T> — "writer"（生產者）。只能寫入元素。
• ChannelReader<T> — "reader"（消費者）。只能讀取元素。
• Channel（通道）分離了責任：writer 不需要知道 reader 的存在，反之亦然。

在 .NET 中有幾種通道實作，各有不同特性：unbounded（無界）、bounded（有界）、single-producer-single-consumer（SPSC）、multi-producer-multi-consumer（MPMC）等。我們會先從通用的開始。

簡單範例：非同步任務隊列

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

class Program
{
    static async Task Main()
    {
        // 建一個無上限的 channel
        var channel = Channel.CreateUnbounded<int>();

        // 生產者任務
        var producer = Task.Run(async () =>
        {
            for (int i = 0; i < 10; i++)
            {
                Console.WriteLine($"生產者: 放入 {i} 到 通道");
                await channel.Writer.WriteAsync(i); // 非同步寫入！
                await Task.Delay(100); // 模擬工作
            }
            channel.Writer.Complete(); // 通知不會再寫入了
        });

        // 消費者任務
        var consumer = Task.Run(async () =>
        {
            await foreach (var item in channel.Reader.ReadAllAsync())
            {
                Console.WriteLine($"消費者: 從通道取到 {item}");
                await Task.Delay(200); // 模擬處理
            }
            Console.WriteLine("消費者: 通道已關閉");
        });

        await Task.WhenAll(producer, consumer);
    }
}

這裡發生了什麼？

• Channel.CreateUnbounded<int>() — 建立一個無界的 channel。
• 生產者用 WriteAsync 把 0 到 9 寫入 channel。
• 寫完後呼叫 Complete() —— 表示「不會再有新元素了！」。
• 消費者用 ReadAllAsync()（也是非同步）遍歷所有元素，直到通道關閉為止。
• Task.Delay 的延遲用來模擬真實工作：可以看到寫入通常比讀取快。

3. 為什麼這一切是非同步的？

傳統的阻塞隊列（例如 BlockingCollection 或用 lock 保護的結構）只能把執行緒卡住。這會浪費寶貴資源，尤其當你有很多任務或追求高效能時。

用 channel：

• 如果生產者比較快，channel 會累積元素（受限於記憶體或你設定的最大長度）；
• 如果消費者比較快，消費者會等待資料出現（但不會把執行緒鎖死，而是把執行緒交還給排程器）。

這個特性非常適合當你不知道生產者跟消費者誰會先完成的情況。

實務應用

• 非同步日誌：把寫檔或寫 DB 的工作放到單獨任務去處理；
• 處理 web 請求：一個任務抓取一堆頁面，另一個任務解析內容；
• 掃描與索引資料夾：一些任務遍歷檔案系統，另一端統計檔案資訊；
• 複雜資料處理管線：像 ETL（Extract–Transform–Load），一個 step 把原料變半成品，另一個 step 變成最終產品。

4. 有界通道 (Bounded Channel)

無界通道很好玩，但我們的記憶體不是無限的（即使你覺得你的機器很猛）。

有界通道 允許你設定同時在通道內的最大元素數量。當通道滿時，生產者會等待直到消費者取走元素。

範例：

var channel = Channel.CreateBounded<int>(new BoundedChannelOptions(3)
{
    FullMode = BoundedChannelFullMode.Wait //（預設）- 等待空位
});

這裡最多只能有三個元素同時在通道內。如果生產者試圖放第四個，會被阻塞（等待）。

多個生產者與消費者

var channel = Channel.CreateUnbounded<int>();

// 2 個生產者
for (int producerId = 0; producerId < 2; producerId++)
{
    Task.Run(async () =>
    {
        for (int i = 0; i < 5; i++)
        {
            int value = producerId * 100 + i;
            Console.WriteLine($"生產者 {producerId}: 放入 {value}");
            await channel.Writer.WriteAsync(value);
            await Task.Delay(50);
        }
        // 每個生產者都呼叫 Complete() —— 很危險！
    });
}
// 小技巧：Complete() 應該只在所有生產者都完成後呼叫一次。
// 這裡為了示範只留一個消費者任務：
Task.Run(async () =>
{
    await foreach (var item in channel.Reader.ReadAllAsync())
    {
        Console.WriteLine($"消費者收到 {item}");
        await Task.Delay(100);
    }
});

注意！ 通道應該在所有生產者完成後才關閉（透過 Complete()）。否則其他還在寫入的生產者會遇到已關閉的通道而拋例外。在真實情況下通常會用計數器或 Task.WhenAll 來處理這件事。

5. 實作範例：透過通道處理影像

我們把情境複雜一點：假設有個資料夾放了很多圖片。一個任務負責尋找圖片並把路徑放進通道，另一個任務讀路徑然後對檔案做處理（例如計算大小或轉檔）。

說明：為了簡單起見，範例只處理檔名（不做影像處理），但概念是一樣的。

using System;
using System.IO;
using System.Threading.Channels;
using System.Threading.Tasks;

class Program
{
    static async Task Main()
    {
        var channel = Channel.CreateBounded<string>(5);

        // 生產者：尋找資料夾中 .jpg 檔
        var producer = Task.Run(async () =>
        {
            foreach (var file in Directory.EnumerateFiles(@"images", "*.jpg"))
            {
                await channel.Writer.WriteAsync(file);
                Console.WriteLine($"加入佇列: {file}");
                await Task.Delay(50); // 模擬搜尋延遲
            }
            channel.Writer.Complete(); // 佇列結束
        });

        // 消費者：讀取並「處理」檔案
        var consumer = Task.Run(async () =>
        {
            await foreach (var file in channel.Reader.ReadAllAsync())
            {
                Console.WriteLine($"處理檔案: {file}");
                await Task.Delay(200); // 模擬處理
            }
            Console.WriteLine("所有圖片已處理完畢!");
        });

        await Task.WhenAll(producer, consumer);
    }
}

6. 設定 Channel：選項與細節

建立通道時可以用選項來調整行為——下面是有界通道常用的參數：

Capacity

SingleWriter

SingleReader

FullMode

帶選項的範例：

var options = new BoundedChannelOptions(10)
{
    SingleWriter = false,
    SingleReader = true,
    FullMode = BoundedChannelFullMode.Wait
};
var channel = Channel.CreateBounded<string>(options);

7. 非同步方法：ReadAsync, WriteAsync, ReadAllAsync

為什麼 async 很重要？

WriteAsync 和 ReadAsync 不會阻塞執行緒！如果沒有資料可讀，task 會被暫停，釋放執行緒給其他工作。這對伺服器或 UI 應用很重要，避免卡死或「當掉」。

ReadAllAsync — 現代 C# 的便利寫法

可以用非同步的方式迭代：

await foreach (var item in channel.Reader.ReadAllAsync())
{
    // 對 item 做處理
}

Channel<T> 和 thread-safe 集合：差異在哪？

ConcurrentQueue<T> / BlockingCollection<T> 對多執行緒場景很好，但不適合純粹的非同步（await）情境。

Channel<T> 是為非同步 pipeline 應用設計的。就執行緒安全性而言，這些集合都能勝任，但 channel 提供更多彈性，且和現代 C# 的功能（像 IAsyncEnumerable）整合得更好。

8. 錯誤與常見坑

別忘了在寫入結束後呼叫 Complete()！不然消費者會一直等下去。

如果有多個生產者，別讓每個都去呼叫 Complete() —— 只有在所有生產者都完成後再呼叫一次。

通道關閉後就不能再寫入，但可以繼續讀取剩下的元素。

同時寫入時的競態情況：如果通道已被關閉，還有人試圖寫入，就會發生例外。

2

任務

C# SELF, 等級 62, 課堂 3

上鎖

建立一個簡單的數字傳遞 Channel

建立一個簡單的數字傳遞 Channel

開啟