dotnet-channels

dotnet-channels

.NET で高パフォーマンスかつスレッドセーフなプロデューサー／コンシューマー間通信を実現する System.Threading.Channels の詳細ガイドです。チャネルの作成、バックプレッシャー戦略、IAsyncEnumerable との連携、グレースフルシャットダウンのパターンを解説します。

対象外： ホステッドサービスのライフサイクルと BackgroundService の登録は、[skill:dotnet-background-services] の対象です。async／await の基本とキャンセレーショントークンの伝播は、[skill:dotnet-csharp-async-patterns] の対象です。

相互参照：ホステッドサービスとのチャネル連携については [skill:dotnet-background-services]、チャネルコンシューマーで使用する非同期パターンについては [skill:dotnet-csharp-async-patterns] を参照してください。

---

Channel の基本

Channel<T> は、ChannelWriter<T> と ChannelReader<T> のエンドポイントを分けて備えたスレッドセーフなデータ構造です。ライターが項目を生成し、リーダーが項目を消費します。同期処理はチャネルがすべて担います。

// Create a channel and separate the endpoints
Channel<WorkItem> channel = Channel.CreateUnbounded<WorkItem>();
ChannelWriter<WorkItem> writer = channel.Writer;
ChannelReader<WorkItem> reader = channel.Reader;

有界チャネルと無制限チャネル

項目	有界	無制限
作成	Channel.CreateBounded<T>(capacity)	Channel.CreateUnbounded<T>()
バックプレッシャー	あり。満杯時の動作は FullMode で制御します	なし。制限なく増加します
メモリの安全性	capacity 個までに制限されます	負荷が高いとメモリを使い果たす可能性があります
用途	本番環境のワークロード、信頼できない生成頻度	少量であることが保証されている場合、プロトタイピング

// Bounded -- preferred for production
var bounded = Channel.CreateBounded<WorkItem>(new BoundedChannelOptions(capacity: 1000)
{
    FullMode = BoundedChannelFullMode.Wait
});

// Unbounded -- use only when you control the producer rate
var unbounded = Channel.CreateUnbounded<WorkItem>();

---

BoundedChannelFullMode

有界チャネルが満杯のときに、プロデューサーが書き込もうとした場合の動作を制御します。

モード	動作	ユースケース
Wait	空きができるまで WriteAsync が待機します	デフォルト。バックプレッシャーを伴う信頼性の高い配信
DropOldest	空きを作るため、チャネル内の最も古い項目を破棄します	テレメトリ、メトリクス。最新データが最も重要な場合
DropNewest	書き込もうとしている項目（最新の項目）を破棄します	レート制限。過剰な受信処理を破棄する場合
DropWrite	書き込もうとしている項目を破棄し、TryWrite は false を返します	オーバーフローを検出しながら、ブロックせずに処理を投げっぱなしにする場合

// DropOldest -- telemetry pipeline where stale readings are expendable
var telemetryChannel = Channel.CreateBounded<SensorReading>(new BoundedChannelOptions(500)
{
    FullMode = BoundedChannelFullMode.DropOldest
});

// DropWrite -- non-blocking enqueue with overflow awareness
var logChannel = Channel.CreateBounded<LogEntry>(new BoundedChannelOptions(10_000)
{
    FullMode = BoundedChannelFullMode.DropWrite
});

if (!logChannel.Writer.TryWrite(entry))
{
    // Channel full -- item was dropped; track overflow metric
    overflowCounter.Add(1);
}

itemDropped コールバック（.NET 7 以降）

.NET 7 以降では、破棄モードの有界チャネルに itemDropped コールバックを設定できます。このコールバックは項目が破棄されるたびに呼び出されます。破棄された項目のメトリクス収集、ログ記録、リソースのクリーンアップに使用してください。

var channel = Channel.CreateBounded(new BoundedChannelOptions(100)
{
    FullMode = BoundedChannelFullMode.DropOldest
},
itemDropped: (item, writer) =>
{
    logger.LogWarning("Dropped item due to channel overflow: {Id}", item.Id);
    droppedItemsCounter.Add(1);
    // Clean up disposable items if needed
    (item as IDisposable)?.Dispose();
});

コールバックは破棄された項目と ChannelWriter<T> を受け取ります（項目をフォールバックチャネルに再ルーティングする必要がある場合に便利です）。

---

プロデューサーパターン

単一プロデューサー

// Write with back-pressure (bounded channels)
await writer.WriteAsync(item, cancellationToken);

// Non-blocking write attempt (returns false if channel is full or completed)
if (!writer.TryWrite(item))
{
    // Handle overflow -- log, retry, or discard
}

複数プロデューサー

複数のプロデューサーは、外部ロックなしで同時に WriteAsync や TryWrite を呼び出せます。チャネルは内部でスレッドセーフに処理されます。

// Multiple API endpoints enqueueing work into a shared channel
app.MapPost("/api/orders/{id}/process", async (
    string id,
    ChannelWriter<OrderCommand> writer,
    CancellationToken ct) =>
{
    await writer.WriteAsync(new OrderCommand(id, "process"), ct);
    return Results.Accepted();
});

app.MapPost("/api/orders/{id}/cancel", async (
    string id,
    ChannelWriter<OrderCommand> writer,
    CancellationToken ct) =>
{
    await writer.WriteAsync(new OrderCommand(id, "cancel"), ct);
    return Results.Accepted();
});

完了の通知

これ以上項目を生成しない場合は、Complete() または TryComplete() を呼び出します。これにより、コンシューマーはストリームの終端を検出できます。

// Signal completion -- no more items will be written
writer.Complete();

// TryComplete is idempotent -- safe to call multiple times
writer.TryComplete();

// Signal completion with an error
writer.TryComplete(new InvalidOperationException("Source failed"));

---

コンシューマーパターン

単一コンシューマー — ReadAsync ループ

基本的なパターンです。項目を待ち、処理し、これを繰り返します。

while (await reader.WaitToReadAsync(cancellationToken))
{
    while (reader.TryRead(out var item))
    {
        await ProcessAsync(item, cancellationToken);
    }
}

この二重ループパターンは、ReadAsync だけを使う方法より推奨されます。再び待機する前に利用可能な項目をすべて読み取るため、非同期ステートマシンのオーバーヘッドを削減できます。

単一コンシューマー — ReadAsync（よりシンプル）

項目ごとのオーバーヘッドを許容できる、よりシンプルなケースでは次のようにします。

try
{
    while (true)
    {
        var item = await reader.ReadAsync(cancellationToken);
        await ProcessAsync(item, cancellationToken);
    }
}
catch (ChannelClosedException)
{
    // Writer called Complete() -- no more items
}

複数コンシューマー（ファンアウト）

複数のコンシューマータスクを実行して処理をスケールできます。各項目が必ず 1 つのコンシューマーだけに読み取られるよう、チャネルが保証します。

public sealed class ScaledChannelProcessor(
    ChannelReader<WorkItem> reader,
    IServiceScopeFactory scopeFactory,
    ILogger<ScaledChannelProcessor> logger) : BackgroundService
{
    private const int WorkerCount = 3;

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        var workers = Enumerable.Range(0, WorkerCount)
            .Select(i => ConsumeAsync(i, stoppingToken));

        await Task.WhenAll(workers);
    }

    private async Task ConsumeAsync(int workerId, CancellationToken ct)
    {
        logger.LogDebug("Consumer {WorkerId} started", workerId);

        while (await reader.WaitToReadAsync(ct))
        {
            while (reader.TryRead(out var item))
            {
                try
                {
                    using var scope = scopeFactory.CreateScope();
                    var handler = scope.ServiceProvider
                        .GetRequiredService<IWorkItemHandler>();
                    await handler.HandleAsync(item, ct);
                }
                catch (Exception ex)
                {
                    logger.LogError(ex,
                        "Consumer {WorkerId}: error processing {ItemId}",
                        workerId, item.Id);
                }
            }
        }

        logger.LogDebug("Consumer {WorkerId} stopped", workerId);
    }
}

---

IAsyncEnumerable との連携

ChannelReader<T>.ReadAllAsync() は IAsyncEnumerable<T> を返すため、await foreach による消費や、LINQ の非同期演算子との連携が可能です。

基本的な await foreach

await foreach (var item in reader.ReadAllAsync(cancellationToken))
{
    await ProcessAsync(item, cancellationToken);
}
// Loop exits when writer calls Complete() and all items are consumed

ReadAllAsync は最もシンプルな消費パターンです。内部で WaitToReadAsync／TryRead を処理し、チャネルが閉じられると完了します。

API エンドポイントからのストリーミング

チャネルは ASP.NET Core のストリーミングレスポンスと自然に組み合わせられます。IAsyncEnumerable<T> を直接返すと、ミニマル API が項目を JSON 配列の要素としてストリーミングします。

app.MapGet("/api/events/stream", (
    ChannelReader<ServerEvent> reader,
    CancellationToken ct) => reader.ReadAllAsync(ct));

LINQ 非同期オペレーター

System.Linq.Async NuGet パッケージを使うと、チャネルストリームを使い慣れた LINQ オペレーターと組み合わせられます。

// NuGet: System.Linq.Async
await foreach (var batch in reader.ReadAllAsync(ct)
    .Where(item => item.Priority >= Priority.High)
    .Buffer(50)  // Collect into batches of 50
    .WithCancellation(ct))
{
    await BulkProcessAsync(batch, ct);
}

チャネルから IAsyncEnumerable を生成する

async IAsyncEnumerable<PriceUpdate> StreamPricesAsync(
    string symbol,
    [EnumeratorCancellation] CancellationToken ct = default)
{
    var channel = Channel.CreateUnbounded<PriceUpdate>();

    // Start producer in background
    _ = Task.Run(async () =>
    {
        try
        {
            await foreach (var tick in marketFeed.SubscribeAsync(symbol, ct))
            {
                await channel.Writer.WriteAsync(tick, ct);
            }
            channel.Writer.TryComplete();
        }
        catch (Exception ex)
        {
            // Propagate error to reader -- ReadAllAsync will throw
            channel.Writer.TryComplete(ex);
        }
    }, ct);

    await foreach (var update in channel.Reader.ReadAllAsync(ct))
    {
        yield return update;
    }
}

---

パフォーマンス

SingleReader / SingleWriter フラグ

チャネルオプションで SingleReader = true または SingleWriter = true を設定すると、ロックフリーの最適化が有効になります。チャネルはこれらのヒントを信頼します。ヒントに違反すると（SingleReader = true の場合に複数のリーダーが同時に読み取るなど）、データが破損します。

// Optimal for single-producer, single-consumer pipeline
var channel = Channel.CreateBounded<T>(new BoundedChannelOptions(1000)
{
    SingleReader = true,   // One consumer task
    SingleWriter = true,   // One producer task
    FullMode = BoundedChannelFullMode.Wait
});

WaitToReadAsync + TryRead パターン

最も効率的なコンシューマーパターンです。WaitToReadAsync はデータが利用可能になるまで中断し、その後 TryRead がバッファー内のすべての項目を同期的に取り出すため、項目ごとの非同期ステートマシンのオーバーヘッドを回避できます。

while (await reader.WaitToReadAsync(ct))
{
    // Drain all currently buffered items synchronously
    while (reader.TryRead(out var item))
    {
        Process(item);
    }
}

TryWrite の高速パス

チャネルに空きがある場合、TryWrite は同期的に実行され、割り当てが発生しません。false の戻り値を処理できるホットパスでは、WriteAsync よりもこちらを優先してください。

// Hot path -- avoid async overhead when channel has space
if (!writer.TryWrite(item))
{
    // Slow path -- wait for space (or handle overflow)
    await writer.WriteAsync(item, ct);
}

有界チャネルのメモリ動作

有界チャネルは、capacity 個のスロットを持つ内部配列を事前に割り当てます。項目は（参照型の場合）参照として格納されるため、チャネルは項目が消費されるまで参照を保持します。メモリに制約のあるワークロードでは、次の点に注意してください。

• 想定される項目サイズと件数に基づいて容量を選択します
• TryRead／ReadAsync が項目を返すと、その項目はすぐに GC の対象になります
• ドロップモード（DropOldest、DropNewest）ではメモリを一定に保てますが、データが失われます

---

キャンセルと正常なシャットダウン

基本的なキャンセル

すべての非同期チャネル操作に CancellationToken を渡してください。キャンセルされると、操作は OperationCanceledException をスローします。

try
{
    await foreach (var item in reader.ReadAllAsync(stoppingToken))
    {
        await ProcessAsync(item, stoppingToken);
    }
}
catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
{
    // Expected during shutdown
}

排出パターン

ライターを完了させて新しい項目が届かないことを通知してから、停止する前に残りの項目を排出します。これにより、シャットダウン中のデータ損失を防げます。

public sealed class DrainableProcessor(
    Channel<WorkItem> channel,
    IServiceScopeFactory scopeFactory,
    ILogger<DrainableProcessor> logger) : BackgroundService
{
    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        var reader = channel.Reader;

        try
        {
            while (await reader.WaitToReadAsync(stoppingToken))
            {
                while (reader.TryRead(out var item))
                {
                    using var scope = scopeFactory.CreateScope();
                    var handler = scope.ServiceProvider
                        .GetRequiredService<IWorkItemHandler>();
                    await handler.HandleAsync(item, stoppingToken);
                }
            }
        }
        catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
        {
            // Shutdown requested -- fall through to drain
        }

        // Signal producers to stop -- any concurrent WriteAsync will throw ChannelClosedException
        channel.Writer.TryComplete();

        // Drain remaining items with a deadline
        logger.LogInformation("Draining remaining work items");
        using var drainCts = new CancellationTokenSource(TimeSpan.FromSeconds(25));

        while (reader.TryRead(out var remaining))
        {
            try
            {
                using var scope = scopeFactory.CreateScope();
                var handler = scope.ServiceProvider
                    .GetRequiredService<IWorkItemHandler>();
                await handler.HandleAsync(remaining, drainCts.Token);
            }
            catch (Exception ex)
            {
                logger.LogWarning(ex, "Error during drain");
            }
        }

        logger.LogInformation("Drain complete");
    }
}

ホストのシャットダウンタイムアウト

ホストのデフォルトのシャットダウンタイムアウトは30秒です。排出にさらに時間が必要な場合は、次のように設定してください。

builder.Services.Configure<HostOptions>(options =>
{
    options.ShutdownTimeout = TimeSpan.FromSeconds(60);
});

---

エージェントが陥りやすい注意点

1. レート制御なしで、本番環境で有界でないチャネルを使用しないでください -- プロデューサーから継続的に負荷がかかると、メモリを使い果たす可能性があります。必ず明示的な容量を指定した有界チャネルを優先してください。
2. SingleReader／SingleWriter の約束に違反しないでください -- これらのフラグはロックフリーの最適化を有効にします。SingleReader = true で複数のリーダーが同時に読み取ると、例外ではなくデータの破損が発生します。
3. ライターで Complete() を呼び出し忘れないでください -- 完了させないと、ReadAllAsync() や WaitToReadAsync を使用するコンシューマーは、最後の項目の後も無期限に待機します。
4. ChannelClosedException をグローバルにキャッチしないでください -- これは、ライターが Complete() を呼び出したことを示し、エラーを伴う場合もあります。ReadAsync の呼び出しの周囲でのみキャッチしてください。WaitToReadAsync／TryRead のループでは、false の戻り値によって完了を処理できます。
5. ホットパスで ReadAsync を使用しないでください -- バッファー内の項目を同期的に排出し、非同期ステートマシンの割り当てを減らすには、WaitToReadAsync + TryRead パターンを優先してください。
6. itemDropped コールバック内でブロックしないでください -- このコールバックはライターのスレッドで同期的に実行されます。処理を高速に保ち（カウンターのインクリメントやログ記録など）、重い処理は別の場所で実行してください。

---

参考資料

• System.Threading.Channels の概要
• Channel API リファレンス
• BoundedChannelOptions
• ChannelReader.ReadAllAsync