CodeGym /コース /C# SELF /Channel: プロデューサー–コンシューマー ( Channel)

Channel: プロデューサー–コンシューマー ( Channel)

C# SELF

レベル 62, レッスン 3

使用可能

1. はじめに

今日はちょっとレベルアップしよう。ここで紹介するのは特別な道具 — Channel。これは .NET のモダンな非同期アプリ向けに作られていて、従来のロックだらけの実装だと性能が出ない場面で本領を発揮するんだ。

「プロデューサー–コンシューマー」パターン（Producer-Consumer）は1960年代からずっと使われている鉄板だよ。ここではシンプルな非同期「パイプライン」を作る：あるタスクやスレッドが何かを作り（たとえばファイルをダウンロードしたり、数を計算したり、イベントを待ったり）、別のタスクがそれを処理する（保存したり、DB に書いたり、UI に表示したり）って感じね。

なぜ Channel が生まれたのか？

プロデューサー–コンシューマーパターンは昔からキューで解決してきた：プロデューサーがタスクをキューに入れ、コンシューマーが取り出す。でもね、BlockingCollection<T> や ConcurrentQueue<T>、あるいは手作業で lock を使う実装は非同期じゃない。つまりスレッドがデータ待ちでブロックされちゃって、async/await のスケジューリングを活かせないんだ。
非同期は単なる流行語じゃなくて今のアーキテクチャの基礎。要素を待つためにスレッドをブロックするのはコストが高すぎる。データの到着をブロックせずに待てる、そこが Channel の目的だよ。
柔軟性：チャネルを使えば複雑なパイプライン、スレッドロジックの分離、中間ステップの追加、ロードバランスなどが低レベルの同期無しでできちゃう。

Channel とは？（アナロジーとアーキテクチャ）

リレーバトンやコンベアベルトを思い浮かべてみて。オブジェクトを一方から他方へ渡せる道具で、隣の人と直接会う必要はない。肝心なのは途中でバトンを落とさないことだね。

Channel は .NET に組み込まれた、タスクやスレッド、プログラムの別々の部分間での非同期データ転送のための仕組みだよ。挿入と取り出しの両方で「待ち」をサポートする非同期キューを実装しているんだ。

プロデューサー はチャネルに要素を入れる（たとえば処理要求）；
コンシューマー は要素を取り出す — それでおしまい！

2. クラス Channel<T> とその構造

まずは名前空間から:

using System.Threading.Channels;

従来のコレクションと違って、Channel はデータ転送用のオブジェクトを作るためのファクトリだよ。

主な種類：

ChannelWriter<T> — 書き手（プロデューサー）。要素の挿入だけする。
ChannelReader<T> — 読み手（コンシューマー）。要素の取り出しだけする。
チャネル（Channel）は責務を分離する：ライターはリーダーのことを知らないし、その逆も同じ。

.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()
    {
        // サイズ制限なしのチャネルを作成する
        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); // WriteAsync を使った非同期書き込み！
                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>() — キューサイズ無制限のチャネルを作っているよ。
プロデューサーは 0 から 9 までの数を WriteAsync で書き込んでいる。
書き込みが終わったら Complete() を呼んで「もう要素は来ないよ！」と合図する。
コンシューマーは ReadAllAsync() で全要素を非同期に列挙して、チャネルが閉じられるまで読み続ける。
遅延（Task.Delay）は実際の仕事を模しただけ：書き込みの方が早くて読み込みが追いつかない状態を作ってるね。

3. なぜこれが非同期で動くの？

従来のブロッキングキュー（たとえば BlockingCollection や lock で守られたもの）はスレッドをブロックするしかない。たくさんのタスクがあるとスレッドを消費してしまって非効率だよね。

チャネルを使うと：

プロデューサーが速ければ、チャネルは要素を蓄える（メモリや設定した最大サイズまで）；
コンシューマーが速ければ、何か来るまで待つけどスレッドはブロックされず、スケジューラに返されるよ。

誰が速いか分からないシナリオにはぴったりだよ。

実用的な応用例

非同期ロギング：ファイルやDBへの書き込みを別タスクで行う；
Webリクエスト処理：あるタスクがページをダウンロードして、別のタスクが解析する；
ディレクトリのスキャンとインデックス作成：一方がファイルシステムを巡回し、他方が統計を集める；
複雑なデータパイプライン：ETL のように、あるステップで原料を半製品に変え、次で最終製品にする、みたいな処理。

4. 制限付きチャネル（Bounded Channel）

無制限チャネルは楽しいけど、メモリは有限（マシンがいくらでかくてもね）。

制限付きチャネル（bounded）は同時にチャネル内に存在できる要素の最大数を設定できる。満杯だとプロデューサーはコンシューマーが何か取り出すまで待つことになるよ。

例：

var channel = Channel.CreateBounded<int>(new BoundedChannelOptions(3)
{
    FullMode = BoundedChannelFullMode.Wait // (デフォルト) - 空きができるまで待つ
});

ここでは同時に最大 3 個の要素しかチャネルに入れられない。4 個目を書き込もうとするとプロデューサーは待たされるよ。

複数のプロデューサーとコンシューマー

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() は全プロデューサーが終わった後に1回だけ呼ぶべき。
// 例としては Task.WhenAll を使うことが多いね。
// コンシューマーの例:
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`	プロデューサーが1つだけなら true にして性能向上を狙える
`SingleReader`	コンシューマーが1つだけなら true にして性能向上を狙える
`FullMode`	チャネルが満杯のときどうするか。選択肢: Wait, DropWrite, DropOldest, DropNewest

オプションの例：

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 はスレッドをブロックしない！読むものがないとタスクは一時停止してスレッドを他に譲る。サーバーや UI アプリでは余計なブロックがフリーズの原因になるから、この性質は重要だよ。

ReadAllAsync — モダンな C# の便利さ

非同期でイテレーションできるよ：

await foreach (var item in channel.Reader.ReadAllAsync())
{
    // item を使って処理する
}

Channel<T> とスレッドセーフなコレクション：違いは何？

ConcurrentQueue<T> / BlockingCollection<T> はスレッドベースのシナリオでは頼りになるけれど、純粋な非同期（await シナリオ）には向かないんだ。

Channel<T> は非同期パイプライン向けに設計されている。スレッドセーフという観点ではどちらも良い仕事をするけれど、チャネルは C# の最新機能（IAsyncEnumerable など）との親和性が高くて柔軟なんだよ。

8. エラーとよくある落とし穴

必ず書き手で Complete() を呼んで、すべての要素を書き終えたことを通知してね！そうしないとコンシューマーはいつまでも新しい要素を待ち続けちゃうよ。

プロデューサーが複数いるときに Complete() を何度も呼ばないで — すべてのプロデューサーが終わってから1回だけ呼ぶんだ。

チャネルを閉じた後はもう書き込めないよ、でも残っている要素は読み取れる。

同時書き込みによるレースコンディション：チャネルが閉じられているときに誰かが書き込もうとすると例外になるよ。

前へ前のレッスン

次へ次のレッスン

TO VIEW ALL COMMENTS OR TO MAKE A COMMENT,
GO TO FULL VERSION