【上級】TPL Dataflowで極めるVB.NET非同期パイプライン設計：数百万件のデータを瞬殺するメッセージングアーキテクチャ
こんにちは。エンタープライズ領域のシステムアーキテクトである私から、現場のエンジニア諸君へ問う。
君たちの書くVB.NETのコードは、未だに「巨大なForループの中で同期的にDBやファイルを叩く」という前世紀の遺物のような構造になってはいないか？
「1件ずつ処理するから安全だ」などという言い訳は、現代のマルチコアプロセッサとクラウド全盛の時代においては、ただの怠慢であり、業務効率化の最大のボトルネックでしかない。
特に、数万〜数百万件のCSVインポート、巨大なXML/JSONのパース、あるいは複数の外部APIを叩いてデータをマージするようなバッチ処理において、スレッドをブロックし続ける設計は、リソースの無駄遣いであり、スケーラビリティの完全な死を意味する。
今回は、TPL Dataflow (Task Parallel Library Dataflow) を用いて、VB.NET上で極限まで効率的かつ堅牢な「非同期メッセージング・パイプライン」を構築する極意を伝授しよう。
—
1. なぜ従来のループ処理は非効率なのか？
業務アプリケーションでよく見かける以下のパターンを見てほしい。
.net
‘ 【アンチパターン】同期ループによる逐次処理
For Each filePath As String In filePaths
Dim rawData As String = File.ReadAllText(filePath) ‘ I/Oブロック
Dim processedData As Dto = Transform(rawData) ‘ CPUバウンド
SaveToDatabase(processedData) ‘ DBブロック
Next
このコードの何が問題か？
1. I/O待ちの完全な遊休状態：ディスク読み込みやDB書き込みの待機時間中、CPUは何もせずスレッドを拘束している。
2. 関心の分離の欠如：データ読み込み、変換、保存のロジックが密結合しており、エラーハンドリングやスロットリング（流量制御）の制御が破綻する。
3. 拡張性の欠如：例えば「変換処理だけを並列化したい」と思ったとき、コードの大部分を書き直す羽目になる。
これを解決するのが、「データを独立した処理ブロック（Block）の流路（パイプライン）に流し込み、非同期で並列処理させる」というTPL Dataflowの思想である。
—
2. TPL Dataflow アーキテクチャの核心
TPL Dataflowは、メッセージ指向プログラミング（Actorモデルに近い概念）を.NET上で実現するための強力なライブラリだ（`System.Threading.Tasks.Dataflow` NuGetパッケージが必要）。
パイプラインを構築する上で、主に以下の3つのブロックを使い分ける。
- `BufferBlock(T)`: データを一時保持するキュー。生産者と消費者の速度差をバッファリングで吸収する。
- `TransformBlock(TInput, TOutput)`: 受け取ったデータを非同期（または同期）で変換し、次のブロックへ流す。
- `ActionBlock(TInput)`: パイプラインの終端。最終的な処理（DB保存やファイル出力など）を実行する。
これらを `LinkTo` メソッドで連結することで、まるで工場のコンベアベルトのような疎結合な処理フローが完成する。
—
3. 【プロダクションコード】堅牢な非同期パイプラインの実装
百聞は一見に如かず。ファイル読み込み ➔ データ変換（CPUバウンド） ➔ データベース保存（I/Oバウンド）を、スロットリング制御（同時実行数の制限）と例外伝播を考慮して実装した、プロダクション品質のVB.NETコードを提示する。
事前にプロジェクトへ NuGet から `System.Threading.Tasks.Dataflow` を導入しておいてほしい。
.net
Imports System.IO
Imports System.Threading.Tasks
Imports System.Threading.Tasks.Dataflow
Namespace Enterprise.Pipelines
‘ 処理対象のデータモデル
Public Class ProcessingItem
Public Property FileId As String
Public Property RawContent As String
Public Property ConvertedValue As Decimal
End Class
Public NotInheritable Class DataPipelineOrchestrator
‘ パイプラインの実行
Public Async Function ExecutePipelineAsync(filePaths As IEnumerable(Of String)) As Task
‘ 1. データの極性に応じた実行オプションの設定
‘ MaxDegreeOfParallelism でCPUコア数やDBのコネクションプールを枯渇させないよう制御する
Dim transformOptions = New ExecutionDataflowBlockOptions With {
.MaxDegreeOfParallelism = Environment.ProcessorCount,
.BoundedCapacity = 100 ‘ メモリ爆発を防ぐためのバックプレッシャー（逆圧）制御
}
Dim dbOptions = New ExecutionDataflowBlockOptions With {
.MaxDegreeOfParallelism = 4, ‘ DBの同時接続数を4に制限
.BoundedCapacity = 50
}
‘ 2. ブロックの定義
‘ 【ブロックA】ファイル読み込み（非同期でバッファに流し込む）
Dim readerBlock As New ActionBlock(Of String)(
Async Function(path)
Dim content As String = Await File.ReadAllTextAsync(path)
‘ 次のブロックへデータを渡すためのロジックをここに繋ぎ込む
End Function)
‘ ※今回はより実践的な TransformBlock を中心としたパイプラインを構築する
‘ [Block 1]: ファイルパスを受け取り、中身を読み込んでDTOにする
Dim loadBlock As New TransformBlock(Of String, ProcessingItem)(
Async Function(filePath)
Try
Dim text = Await File.ReadAllTextAsync(filePath)
Return New ProcessingItem With {
.FileId = Path.GetFileName(filePath),
.RawContent = text
}
Catch ex As Exception
Console.WriteLine($”[Read Error] {filePath}: {ex.Message}”)
Return Nothing ‘ エラー時はnullを返す等のハンドリング
End Try
End Function, transformOptions)
‘ [Block 2]: データの変換・計算処理（CPUバウンドな重い処理を想定）
Dim processBlock As New TransformBlock(Of ProcessingItem, ProcessingItem)(
Function(item)
If item Is Nothing Then Return Nothing
‘ 重い計算処理やパース処理
item.ConvertedValue = LongRunningCpuBoundCalculation(item.RawContent)
Return item
End Function, transformOptions)
‘ [Block 3]: データベースへの保存（I/Oバウンドな終端ブロック）
Dim saveBlock As New ActionBlock(Of ProcessingItem)(
Async Function(item)
If item Is Nothing Then Return
‘ 非同期DB保存処理のシミュレーション
Await SaveToDatabaseAsync(item)
Console.WriteLine($”[Completed] File: {item.FileId} -> Value: {item.ConvertedValue}”)
End Function, dbOptions)
‘ 3. ブロックの結合 (パイプラインの構築)
‘ PropagateCompletion = True により、上流のエラーや完了状態が下流へ伝播する
Dim linkOptions = New DataflowLinkOptions With {.PropagateCompletion = True}
loadBlock.LinkTo(processBlock, linkOptions, Function(item) item IsNot Nothing)
processBlock.LinkTo(saveBlock, linkOptions, Function(item) item IsNot Nothing)
‘ 4. データの投入（生産者フェーズ）
For Each path In filePaths
Await loadBlock.SendAsync(path)
Next
‘これ以上の入力がないことを通知し、末端までの完了を待機
loadBlock.Complete()
‘ saveBlock がすべての処理を終えるまで非同期で待機
Await saveBlock.Completion
Console.WriteLine(“すべてのパイプライン処理が正常に完了しました。”)
End Function
‘ ダミーのCPUバウンド処理
Private Function LongRunningCpuBoundCalculation(raw As String) As Decimal
‘ 実際の業務ロジック（正規表現、複雑な数値計算など）
Return raw.Length 1.05D
End Function
‘ ダミーの非同期DB保存処理
Private Async Task SaveToDatabaseAsync(item As ProcessingItem)
‘ 例: Using conn As New SqlConnection(…)
Await Task.Delay(100) ‘ DB書き込み待ちの模擬
End Function
End Class
End Namespace
—
4. プロフェッショナルが押さえるべき設計上の注意点
このコードを現場に導入するにあたり、シニアエンジニアとして知っておくべき「罠」と「対策」を共有する。
① バックプレッシャー（逆圧）の制御
`BoundedCapacity` の設定を怠ってはならない。例えば、ファイル読み込みが爆速で行われ、DB書き込みが追いつかない場合、無限にメモリ上にデータが蓄積され、最終的に `OutOfMemoryException` でプロセスがクラッシュする。
`BoundedCapacity` を明示することで、上流の読み込み速度が自動的に下流（DB書き込み）の速度にスロットリングされるようになる。これがTPL Dataflowの真骨頂である。
② 例外ハンドリングと Fault
パイプライン途中のブロックで例外が発生した場合、何もしないと処理が途中でフリーズするか、予期せぬサイレントエラーになる。
`PropagateCompletion = True` を設定し、最終的な `saveBlock.Completion` を `Try-Catch` で囲むことで、パイプライン全体で発生した例外をキャッチできるように設計すること。
.net
Try
Await saveBlock.Completion
Catch ex As AggregateException
‘ TPL Dataflowの例外はAggregateExceptionでラップされることが多い
For Each innerEx In ex.InnerExceptions
Console.WriteLine($”パイプライン異常終了: {innerEx.Message}”)
Next
End Try
③ ファイル・DB連携におけるリソース競合
マルチスレッドでファイルにアクセスする場合や、DBのコネクションプールを共有する場合、スレッドセーフティに配慮する必要がある。
特にEntity Framework Coreなどを利用する場合、DbContextはスレッドセーフではないため、`ActionBlock` の中で `DbContext` を都度生成・破棄（`New` して `Using` する）する設計にしなければ、並列実行時に致命的なコンテキスト競合を引き起こす。
—
5. まとめ
今回解説した TPL Dataflow によるパイプライン設計は、単なる「テクニック」ではなく、モダンなエンタープライズ開発における必須の教養である。
- 処理の疎結合化により、保守性とテスト容易性が劇的に向上する。
- `MaxDegreeOfParallelism` と `BoundedCapacity` のチューニングにより、ハードウェアリソースを限界まで引き出しつつ、システムを安定稼働させることができる。
「動けばいい」というアマチュアのコードから脱却し、スケーラビリティと堅牢性を兼ね備えた美しいアーキテクチャを、君たちのプロジェクトにも取り入れてほしい。健闘を祈る。
