VB.NETでのTPL Dataflowを用いたパイプライン処理の構築：非同期メッセージングによる疎結合なデータ処理フロー
レガシーなVB 6.0やVBAの時代から、私たちは「上から下へ流れる同期処理」の呪縛と共に生きてきた。何万件ものレコードをループで回し、1件ずつ外部APIを叩き、データベースに書き込む。この古典的な逐次処理モデルは、現代のマルチコアプロセッサと高速なI/O環境の前では、ただのボトルネック製造機に過ぎない。
「画面がフリーズする」「バッチ処理が夜間ウィンドウに収まらない」――そのような現場の悲鳴に対し、ThreadやTaskを直接生で制御するスパゲッティコードを書くのはもう終わりだ。
今回は、CLR（Common Language Runtime）の深部を知り尽くしたアーキテクトへ向けて、.NETのTPL Dataflow (Task Parallel Library Dataflow) を用いた、極限まで洗練されたモダンな非同期パイプライン処理の構築法を伝授する。
—
1. なぜTPL Dataflowなのか：レガシー並行処理の限界
VB.NETで並行処理を実装する場合、長らく `System.Threading.Tasks.Task` や `Parallel.For` が使われてきた。しかし、これらは「計算の並列化」には強いものの、「データの流れ（ストリーム）の制御」には向いていない。
例えば、以下のようなシステム間連携の要件を考えてみてほしい。
1. 巨大なCSVファイルの読み込み（メモリ枯渇を防ぐためストリーミング処理が必要）
2. データ型変換とバリデーション（CPUバウンド）
3. 外部Web APIへのリクエスト（I/Oバウンド・レートリミットあり）
4. データベースへのバルクインサート（バッファリングとトランザクション制御が必要）
これを従来の `Task` と `List(Of T)` の組み合わせで実装すると、スレッド間の同期排他（Mutex/Monitor）やキューの管理でコードが破綻し、デバッグ不可能なデッドロックの温床となる。
TPL Dataflowは、データを「メッセージ」としてカプセル化し、独立したブロック（Block）の間を非同期で流す。各ブロックは独自のバッファを持ち、スレッドプールを効率的に共有しながら、宣言的にパイプラインを構築できる。
—
2. アーキテクチャの全体像：メッセージングによる疎結合
TPL Dataflowの基本単位は `ITargetBlock(Of TInput)` と `ISourceBlock(Of TOutput)`、そしてその両方を兼ねる `IPropagatorBlock(Of TInput, TOutput)` である。
今回の実戦投入モデルでは、以下の3つのブロックを連結（LinkTo）する。
[ ReaderBlock ] (生成)
│ (BufferBlock: バッファリング)
▼
[ TransformBlock ] (非同期APIコール / 変換：並列度制御)
│ (BufferBlock: 順序・流量制御)
▼
[ ActionBlock ] (DB一括書き込み：メモリ効率化)
この構造により、APIの応答速度が低下しても、前段のBufferBlockが背圧（Backpressure）をかけ、メモリの爆発的消費（OutOfMemoryException）を物理的に防ぐことができる。
—
3. 実装コード：限界まで最適化されたパイプライン
以下のコードは、NuGetから `System.Threading.Tasks.Dataflow` を導入した環境で動作する、プロダクション品質のVB.NETコードである。
Imports System.Threading.Tasks
Imports System.Threading.Tasks.Dataflow
Imports System.IO
Module DataflowPipelineDemo
‘ 処理対象のデータモデル
Public Class RawRecord
Public Property Id As Integer
Public Property RawData As String
End Class
Public Class ProcessedRecord
Public Property Id As Integer
Public Property ResultValue As String
End Class
Public Sub Main()
‘ 同期のエントリポイントから非同期処理を安全に呼び出す
RunPipelineAsync().GetAwaiter().GetResult()
End Sub
Private Async Function RunPipelineAsync() As Task
Console.WriteLine(“=== TPL Dataflow パイプライン処理を開始します ===”)
Dim sw = Stopwatch.StartNew()
‘ —————————————————————–
‘ 1. 実行オプションの定義 (BoundedCapacityによるメモリ保護)
‘ —————————————————————–
Dim linkOptions As New DataLinkOptions With {.PropagateCompletion = True}
‘ 変換ブロックの並列度とバッファ上限を設定
Dim transformOptions As New ExecutionDataflowBlockOptions With {
.MaxDegreeOfParallelism = Environment.ProcessorCount, ‘ コア数をフル活用
.BoundedCapacity = 1000 ‘ メモリ溢れを防ぐための背圧制御
}
‘ 書き込みブロックのオプション（直列化してDB競合を防ぐ場合など）
Dim actionOptions As New ExecutionDataflowBlockOptions With {
.MaxDegreeOfParallelism = 1,
.BoundedCapacity = 500
}
‘ —————————————————————–
‘ 2. パイプライン・ブロックの構築
‘ —————————————————————–
‘ 【ブロックA】データ変換・API連携ブロック (I/Oバウンド想定)
Dim enricherBlock As New TransformBlock(Of RawRecord, ProcessedRecord)(
Async Function(record)
‘ 外部API呼び出しや重い処理をシミュレート (非同期)
Await Task.Delay(50)
Return New ProcessedRecord With {
.Id = record.Id,
.ResultValue = $”Processed: {record.RawData.ToUpper()}”
}
End Function, transformOptions)
‘ 【ブロックB】結果出力・ストレージ書き込みブロック
Dim batchBuffer = New List(Of ProcessedRecord)()
Dim writerBlock As New ActionBlock(Of ProcessedRecord)(
Sub(processed)
‘ ここでバッファリングやDBバルクインサートを行う
SyncLock batchBuffer
batchBuffer.Add(processed)
If batchBuffer.Count >= 100 Then
FlushToDatabase(batchBuffer)
batchBuffer.Clear()
End If
End SyncLock
End Sub, actionOptions)
‘ —————————————————————–
‘ 3. ブロックの接続 (パイプラインの形成)
‘ —————————————————————–
‘ enricherBlock の出力が writerBlock の入力へ流れるようにリンク
Using enricherBlock.LinkTo(writerBlock, linkOptions)
‘ —————————————————————–
‘ 4. データ供給 (Producer)
‘ —————————————————————–
For i As Integer = 1 To 5000
Dim item As New RawRecord With {
.Id = i,
.RawData = $”data_payload_{i}”
}
‘ Postは非同期ブロックのバッファに空きがあれば即座にtrueを返す
‘ BoundedCapacityにより、消費スピードを超えた投入はここで自然にウェイトがかかる
Await enricherBlock.SendAsync(item)
Next
‘ データ供給の終了を通知
enricherBlock.Complete()
‘ 最後の書き込みブロックが全データの処理を完了するまで待機
Await writerBlock.Completion
‘ 端数データのフラッシュ
SyncLock batchBuffer
If batchBuffer.Count > 0 Then
FlushToDatabase(batchBuffer)
batchBuffer.Clear()
End If
End SyncLock
End Using
sw.Stop()
Console.WriteLine($”=== すべての処理が完了しました。実行時間: {sw.ElapsedMilliseconds} ms ===”)
End Function
Private Sub FlushToDatabase(ByVal records As List(Of ProcessedRecord))
‘ 実際のシステムではここでADO.NETのSqlBulkCopyやDapper等を用いた一括処理を行う
Console.WriteLine($”[DB Flush] {records.Count} 件のレコードを一括書き込みしました。 (先頭ID: {records.First().Id})”)
End Sub
End Module
—
4. チーフアーキテクトが教える「現場でハマる罠」と極限の知見
このコードは美しく動作するが、エンタープライズの荒波にもまれる現場では、さらなる知見が必要だ。
① `.PropagateCompletion = True` の重要性
ブロックを `LinkTo` で繋ぐ際、このオプションを忘れると、前段（Producer）が完了しても後段（Consumer）にシグナルが伝播せず、パイプラインが永遠に終了待ち（ハングアップ）の状態に陥る。データフロー設計の基本原則は「例外時・正常終了時ともにシグナルを全段に伝播させること」である。
② メモリ最適化とオブジェクトのライフサイクル
レガシーシステムからの移行期において、巨大な `DataSet` や `DataTable` をそのままDataflowに流し込む愚は避けてほしい。ガベージコレクション（GC）の第2世代、第3世代を圧迫し、LOH（Large Object Heap）の断片化を引き起こす。
TPL Dataflowを使うときは、必ず「1レコード（DTO）単位のストリーミング処理」を徹底し、不要になったオブジェクトは速やかにスコープ外へ追いやること。必要であれば、カスタムの `IDisposable` 実装クラスを流し、ブロックの最終端で明示的に `Dispose()` を呼ぶ設計を取り入れるべきだ。
③ Windows API / レガシーDLLとの連携における注意点
もしこのパイプライン内部（TransformBlockなど）で、スレッドセーフではないレガシーなWindows API（COMコンポーネントや古い3rdパーティ製DLL）を呼び出す必要がある場合、`MaxDegreeOfParallelism = 1` にするか、または `System.Threading.ApartmentState.STA` スレッドを明示的に要求する設計が必要となる。TPL DataflowのデフォルトのスレッドプールはMTA（Multithreaded Apartment）で動作するため、STA必須のコンポーネントを並列呼び出しすると `InvalidCastException` や謎のクラッシュを引き起こす。その場合は、ブロック内で個別に `Task.Run` と `TaskCreationOptions` を調整するか、排他制御を挟むこと。
—
5. おわりに
Visual Basic (.NET) は、単なる「古い言語の互換レイヤー」ではない。CLRの進化をダイレクトに享受できる、極めて強力なモダンプログラミング言語である。
今回解説した TPL Dataflow によるパイプライン処理は、バッチ処理の高速化だけでなく、システム全体の堅牢性（レジリエンス）を劇的に向上させる。レガシーな発想から脱却し、メッセージングと非同期の思想をコードに宿すことで、あなたのシステムは次世代の負荷に耐えうる「壊れない要塞」へと生まれ変わるはずだ。
