課題の背景：午前2時14分のOutOfMemory障害
午前2時14分、ベッドサイドでスマートフォンが激しく振動しました。PagerDutyからのアラートを確認すると、主要な決済処理マイクロサービスが未処理のOutOfMemoryExceptionでクラッシュしていました。Grafanaのダッシュボードを開くと、メモリ使用量のグラフがほぼ垂直に跳ね上がり、わずか数分で320MBから6.2GBに達していました。CPU使用率は100%近くに張り付き、スレッドプールの枯渇によって依存するすべてのHTTPエンドポイントの応答が停止していました。
根本原因は極めてシンプルでした。深夜に開始された突発的なマーケティングセールにより、APIに毎秒25,000件ものトランザクションリクエストが殺到したのです。ASP.NET Coreのコントローラーは受信したペイロードをすべて受け入れ、PostgreSQLへレコードを書き出すバックグラウンドタスクを前提に、容量無制限のConcurrentQueue<PaymentEvent>へ即座に投入していました。
しかし、PostgreSQLの書き込み性能は毎秒4,200件が上限でした。わずか8分の間に990万件以上の未消費オブジェクトがヒープ領域を埋め尽くしました。これによりGen-2ガベージコレクションによる停止が頻発し、最終的にKubernetesによってPodが強制終了されました。
このような障害が発生すると、多くのエンジニアは即座にRabbitMQ、AWS SQS、Apache Kafkaといった外部メッセージブローカーの導入を検討しがちです。しかし、分散ブローカーの追加には運用オーバーヘッド、ネットワークのラウンドトリップ、ペイロードのシリアライズ処理、そして毎月のインフラコストといった現実的な負担が伴います。
ログイベントのバッファリング、監査ログの集約、データベースへのバッチ書き込みなど、ワークフローが単一のプロセス内で完結する場合、外部メッセージブローカーは不要です。本当に必要なのは、プロセス内におけるバックプレッシャーです。そこで真価を発揮するのがSystem.Threading.Channelsです。
コアコンセプト：なぜChannel<T>は従来のコレクションより優れているのか
.NETにSystem.Threading.Channelsが導入される前、開発者は通常ConcurrentQueue<T>かBlockingCollection<T>の二者択一を迫られていました。しかし、本番環境の高負荷下では、どちらも致命的な限界に直面します。
- ConcurrentQueue<T>：容量制限が組み込まれておらず、非同期待機プリミティブも備えていません。コンシューマーは新しいデータを待つためにループ内でスピンまたはスリープせざるを得ず、無駄にCPUサイクルを消費します。
- BlockingCollection<T>：固定の容量制限はサポートしているものの、
Take()による同期的なスレッドブロックが発生します。トラフィック急増時にワーカースレッドがアイドル状態でブロックされ続けると、.NETのThreadPoolがあっという間に枯渇し、HTTPリクエストのレイテンシが跳ね上がります。
Channel<T>はこれら両方の問題を解決します。スレッドセーフかつ完全に非同期な通信を提供し、役割が2つのコントラクトに明確に分離されています。
ChannelWriter<T>：プロデューサーがWriteAsync()またはノンブロッキングなTryWrite()を使ってアイテムを追加します。ChannelReader<T>：コンシューマーがWaitToReadAsync()やReadAllAsync()を使ってアイテムを取り出します。
最大の強みは、境界付き（有界）のフロー制御にあります。バッファの上限に達した際、WriteAsync()は実行中のOSスレッドをブロックしません。代わりに、コンシューマーがキューの空きを確保するまで、標準のタスク継続を介して非同期に実行権を譲渡します。
制御不能になったキューがノードをダウンさせる光景を一度でも経験すれば、バックプレッシャーが決して選択肢ではなく必須であることが分かります。これこそが、高スループットなサービスを安定稼働させ続けるための防波堤なのです。
実践：堅牢なインメモリパイプラインの構築
先ほどの決済受付エンジンを再設計してみましょう。有界チャネルを設定し、非同期プロデューサーとなるコントローラーを実装し、BackgroundServiceでキューを処理していきます。
ステップ1：チャネルオプションの設定
まず、有界チャネルをインスタンス化し、バッファが満杯になった際の挙動を指定します。
using System.Threading.Channels;
public static class ChannelFactory
{
public static Channel<PaymentEvent> CreatePaymentChannel(int capacity = 10_000)
{
var options = new BoundedChannelOptions(capacity)
{
// ライターを非同期に一時停止させてバックプレッシャーを適用
FullMode = BoundedChannelFullMode.Wait,
// 複数のWebリクエストスレッドが並行してデータをプッシュ
SingleWriter = false,
// 単一のバックグラウンドワーカーがキューを消費
SingleReader = true,
// 書き込みスレッド上での同期継続によるデッドロックを防止
AllowSynchronousContinuations = false
};
return Channel.CreateBounded<PaymentEvent>(options);
}
}
public record PaymentEvent(Guid TransactionId, decimal Amount, DateTime CreatedAt);SingleReader = trueを指定すると、チャネルは複数コンシューマー向けの同期ロックをバイパスします。この小さな設定により、コンシューマーループでのアトミック競合が排除され、CPUキャッシュラインのバウンシングが軽減されます。
ステップ2：プロデューサーの実装（APIコントローラー）
APIコントローラーにChannelWriter<PaymentEvent>をDI（依存性の注入）します。決済リクエストを受信したら、バッファに非同期でプッシュします。
using System.Threading.Channels;
using Microsoft.AspNetCore.Mvc;
[ApiController]
[Route("api/[controller]")]
public class PaymentsController : ControllerBase
{
private readonly ChannelWriter<PaymentEvent> _writer;
private readonly ILogger<PaymentsController> _logger;
public PaymentsController(ChannelWriter<PaymentEvent> writer, ILogger<PaymentsController> logger)
{
_writer = writer;
_logger = logger;
}
[HttpPost]
public async Task<IActionResult> ProcessPayment([FromBody] PaymentRequest request, CancellationToken ct)
{
var paymentEvent = new PaymentEvent(Guid.NewGuid(), request.Amount, DateTime.UtcNow);
// 10,000件のバッファが満杯の場合は非同期で待機
await _writer.WriteAsync(paymentEvent, ct);
return Accepted(new { paymentEvent.TransactionId, Status = "キュー登録済み" });
}
}
public record PaymentRequest(decimal Amount);保留中のトランザクションが10,000件に達したときの挙動に注目してください。WriteAsyncはリクエストスレッドを束縛せず、即座にKestrelへ解放します。受信したHTTPリクエストはメモリを肥大化させたりワーカースレッドを枯渇させたりすることなく、順番が来るのを待機します。
ステップ3：コンシューマーとなるバックグラウンドワーカーの実装
続いて、リーダーからアイテムを取り出して下流へ書き出すBackgroundServiceを実装します。
using System.Threading.Channels;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
public class PaymentProcessorService : BackgroundService
{
private readonly ChannelReader<PaymentEvent> _reader;
private readonly ILogger<PaymentProcessorService> _logger;
public PaymentProcessorService(ChannelReader<PaymentEvent> reader, ILogger<PaymentProcessorService> logger)
{
_reader = reader;
_logger = logger;
}
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
_logger.LogInformation("決済処理ワーカーを開始しました。");
try
{
// ReadAllAsyncはアイテムが到着するたびに取り出し、チャネル完了時に終了する
await foreach (PaymentEvent payment in _reader.ReadAllAsync(stoppingToken))
{
await ProcessPaymentAsync(payment, stoppingToken);
}
}
catch (OperationCanceledException)
{
_logger.LogInformation("決済処理ワーカーを正常にシャットダウンしています。");
}
}
private async Task ProcessPaymentAsync(PaymentEvent payment, CancellationToken ct)
{
// データベース永続化のレイテンシをシミュレート
await Task.Delay(2, ct);
_logger.LogDebug("トランザクション {Id} を処理しました。", payment.TransactionId);
}
}ステップ4：依存関係をDIコンテナに登録
Program.csでチャネルをシングルトンとして登録し、責務の境界を明確に保つためにリーダーとライターを個別に公開します。
var builder = WebApplication.CreateBuilder(args);
var paymentChannel = ChannelFactory.CreatePaymentChannel(capacity: 10_000);
builder.Services.AddSingleton(paymentChannel.Reader);
builder.Services.AddSingleton(paymentChannel.Writer);
builder.Services.AddHostedService<PaymentProcessorService>();
builder.Services.AddControllers();
var app = builder.Build();
app.MapControllers();
app.Run();本番運用のチェックリストと重要なまとめ
無制限のインメモリキューからChannel<T>へ移行したことで、深夜のアラートに悩まされることは完全になくなりました。その後のピークセールでは、流入トラフィックが毎秒28,000リクエストを超えた際も、メモリ使用量はわずか185MBで完全に横ばいを維持しました。
チャネルを本番環境へ導入する前に、以下の4つの原則を確認してください。
- 必ずバッファ容量（境界）を設定する： パブリックエンドポイントの背後で
Channel.CreateUnbounded<T>を使用してはいけません。無制限バッファは短期的なスループットと引き換えに、最終的なプロセスクラッシュを招きます。 - 適切なFullModeを選択する： データの欠落が許されない場合は
BoundedChannelFullMode.Waitを使用します。重要度の低いテレメトリデータであれば、DropOldestやDropNewestに切り替えて、古いデータより最新データを優先します。 - グレースフルシャットダウン時に残データを処理する： アプリケーションのライフサイクル停止フック内で
writer.Complete()を呼び出します。これにより、ランタイムが終了する前にReadAllAsync()がバッファ内の未処理アイテムを最後まで完了できます。 - シングルアクターフラグを活用する： チャネルの片側を処理するワーカーが単一の場合は、必ず
SingleReaderまたはSingleWriterを有効にします。ランタイム内部のロック処理が省略され、測定可能なレベルでスループットが向上します。
