非同期・並列・バッチ処理の実践設計 - 金融情報システム開発なら20年以上の実績があるテンファイブ株式会社

本章では、VB.NET業務システムにおける非同期処理、並列処理、バッチ処理を、本番運用に耐える形で設計する方法を学習します。

大量CSV取込、夜間集計、帳票一括生成、外部システム連携、メール一括送信、データ移行など、企業システムでは長時間実行される処理が数多く存在します。

このような処理では、単純にTask.RunやParallel.ForEachを使って高速化するだけでは不十分です。途中で障害が発生した場合にどこから再開するのか、同じデータを二重処理しないか、複数ジョブが同時実行されても安全か、処理状況を運用担当者が確認できるかといった設計が必要になります。

本章のゴールは、単なる非同期プログラミングではなく、大量処理を安全・高速・再実行可能な形で設計し、本番環境で安定運用できるバッチアーキテクチャを構築できるようになることです。

1. バッチ処理とは

バッチ処理とは、大量のデータや定型処理を、人間の操作を介さずまとめて実行する処理方式です。

代表的なバッチ処理

夜間売上集計
請求データ作成
入金消込
CSV一括取込
大量メール送信
帳票一括生成
在庫同期
マスタ連携
ログ集計
データアーカイブ

バッチ処理は、画面処理と異なり、数千件、数万件、数百万件のデータを扱うことがあります。そのため、性能だけでなく、障害復旧、再実行、ログ、監視、排他制御を考慮する必要があります。

オンライン処理とバッチ処理の違い

項目	オンライン処理	バッチ処理
実行契機	ユーザー操作	時刻・イベント・ジョブ
処理件数	比較的少ない	大量
応答時間	短さが重要	全体処理時間が重要
失敗時	画面へ通知	ログ・監視・再実行が必要
運用	利用者中心	運用担当者中心

2. バッチアーキテクチャ

本格的なバッチ処理では、処理本体だけでなく、ジョブ制御、状態管理、ログ、エラー処理、再実行などを含めたアーキテクチャが必要です。

基本構成

Scheduler
    ↓
BatchRunner
    ↓
Application Service
    ↓
Domain
    ↓
Repository / External API
    ↓
Database / External System

構成要素

要素	役割
Scheduler	指定時刻にジョブを起動する
BatchRunner	バッチ開始・終了・例外を管理する
Service	バッチの業務処理を実行する
Repository	DBアクセスを担当する
JobStatus	処理状態を管理する
Logger	処理結果・障害情報を記録する

バッチ処理を1つの巨大なMainメソッドに書くのではなく、責務ごとに分離することが重要です。

3. ジョブ管理

バッチ処理では、「いつ、どのジョブが、どの状態で実行されたか」を管理する必要があります。

ジョブ管理テーブル例

BatchJobs
- JobExecutionId
- JobName
- StartTime
- EndTime
- Status
- ProcessedCount
- SuccessCount
- ErrorCount
- Message
- ServerName

代表的なステータス

Waiting
Running
Succeeded
Failed
Canceled
Retrying

Enum例

Public Enum BatchStatus
    Waiting
    Running
    Succeeded
    Failed
    Canceled
    Retrying
End Enum

開始時にRunningへ変更し、正常終了時はSucceeded、異常終了時はFailedへ変更します。

バッチRunner例

Public Class BatchRunner

    Public Sub Run(job As IBatchJob)

        Dim executionId As Long = CreateExecution(job.Name)

        Try
            UpdateStatus(executionId, BatchStatus.Running)

            job.Execute()

            UpdateStatus(executionId, BatchStatus.Succeeded)

        Catch ex As Exception

            UpdateStatus(executionId, BatchStatus.Failed)

            Logger.Error(
                "バッチ処理に失敗しました。Job=" & job.Name,
                ex)

            Throw
        End Try

    End Sub

End Class

業務処理とジョブ管理を分離することで、すべてのバッチに共通した運用制御を適用できます。

4. キュー処理

キュー処理とは、処理対象を一度キューに登録し、ワーカーが順次または並列で処理する方式です。

外部API送信、メール送信、帳票生成、画像処理など、即時完了する必要がない処理に適しています。

処理フロー

業務処理
    ↓
Queueへ登録
    ↓
Commit
    ↓
Worker
    ↓
対象取得
    ↓
処理実行
    ↓
成功 / 失敗記録

キューテーブル例

ProcessingQueue
- QueueId
- JobType
- TargetId
- Status
- RetryCount
- NextRetryAt
- CreatedAt
- StartedAt
- CompletedAt
- ErrorMessage

キュー登録例

Public Sub Enqueue(targetId As Integer)

    Dim queueItem As New QueueItem With {
        .TargetId = targetId,
        .Status = QueueStatus.Waiting,
        .RetryCount = 0,
        .CreatedAt = DateTime.Now
    }

    _queueRepository.Insert(queueItem)

End Sub

キュー処理のメリット

オンライン処理を短くできる
失敗した処理だけ再実行できる
並列度を制御しやすい
外部システム障害の影響を分離できる
処理状況を可視化しやすい

5. 非同期処理の実践設計

バッチ処理では、外部API、ファイルI/O、DBアクセスなど、待ち時間が発生する処理を非同期化することで効率を改善できます。

非同期メソッド例

Public Async Function ExecuteAsync() As Task

    Dim targets = Await _repository.FindTargetsAsync()

    For Each target In targets
        Await ProcessAsync(target)
    Next

End Function

外部API呼び出し例

Public Async Function SendAsync(data As SendData) As Task

    Dim response As Net.Http.HttpResponseMessage =
        Await _httpClient.PostAsync(
            data.Url,
            data.Content)

    response.EnsureSuccessStatusCode()

End Function

I/O待ちでは、無理にTask.Runへ包むのではなく、利用するライブラリが提供する非同期APIを利用する方が効率的です。

Task.Runが向いている処理

CPU負荷の高い計算
画像変換
既存の同期処理を一時的にバックグラウンド化する

Async APIが向いている処理

HTTP通信
DB通信
ファイルI/O
ネットワーク通信

6. 並列処理

大量の独立データを処理する場合、並列処理によって全体時間を短縮できる場合があります。

Parallel.ForEachの例

Parallel.ForEach(targets,
    Sub(target)

        ProcessTarget(target)

    End Sub)

ただし、並列処理は高速化の万能手段ではありません。 DBや外部APIへ大量アクセスすると、相手側の処理能力を超えて逆に遅くなる可能性があります。

並列化に向いている処理

データ同士が独立している
CPU負荷の高い処理
画像変換
ファイル単位の処理
独立した検証処理

注意が必要な処理

同じDBテーブルへの大量更新
採番処理
在庫更新
同一ファイルへの書き込み
外部APIへの大量リクエスト

並列化を検討するときは、アプリケーション側だけでなくDBや外部サービス側の許容量も考える必要があります。

7. 並列度の制御

1000件の処理があるからといって、1000件すべてを同時実行してはいけません。同時実行数を制御する必要があります。

SemaphoreSlimを使うと、同時実行数を制限できます。

SemaphoreSlimの例

Public Async Function ProcessAllAsync(
    targets As List(Of TargetData)) As Task

    Dim semaphore As New Threading.SemaphoreSlim(5)

    Dim tasks =
        targets.Select(
            Async Function(target)

                Await semaphore.WaitAsync()

                Try
                    Await ProcessAsync(target)

                Finally
                    semaphore.Release()
                End Try

            End Function)

    Await Task.WhenAll(tasks)

End Function

この例では、最大5件まで同時に実行します。

並列度を決める観点

CPUコア数
DB接続プール
外部APIのレート制限
ネットワーク帯域
メモリ使用量
相手システムの処理能力

並列度は高ければ高いほど速いわけではありません。計測しながら最適な値を決めることが重要です。

8. Race Conditionと共有データ

並列処理では、複数スレッドが同じデータを変更するとRace Conditionが発生します。

危険な例

Dim successCount As Integer = 0

Parallel.ForEach(targets,
    Sub(target)

        ProcessTarget(target)

        successCount += 1

    End Sub)

successCount += 1は複数スレッドから同時に実行されるため、正しい件数にならない可能性があります。

Interlockedを使う例

Dim successCount As Integer = 0

Parallel.ForEach(targets,
    Sub(target)

        ProcessTarget(target)

        Threading.Interlocked.Increment(successCount)

    End Sub)

単純なカウンタ更新ではInterlockedを使うことで安全に更新できます。

共有状態を減らす

最も安全なのは、共有データそのものを減らすことです。

処理ごとに独立した変数を使う
結果を最後に集約する
共有Listへの直接追加を避ける
Concurrent系コレクションを利用する

9. チャンク処理

数十万件、数百万件のデータを一度に読み込むと、メモリ不足、長時間トランザクション、ロック競合などが発生します。

そのため、大量データは一定件数ごとに分割して処理します。これをチャンク処理と呼びます。

チャンク処理イメージ

総件数：100,000件

1～1,000
1,001～2,000
2,001～3,000
...
99,001～100,000

VB.NETでの例

Dim chunkSize As Integer = 1000

For offset As Integer = 0 To targets.Count - 1 Step chunkSize

    Dim chunk =
        targets.Skip(offset).
                Take(chunkSize).
                ToList()

    ProcessChunk(chunk)

Next

チャンク処理のメリット

メモリ使用量を抑えられる
トランザクションを短くできる
途中再開しやすい
進捗を管理しやすい
障害範囲を限定できる

10. チェックポイント設計

長時間バッチが途中で失敗した場合、最初からすべてやり直すのは非効率です。処理済み位置を記録し、途中から再開できるようにする仕組みをチェックポイントと呼びます。

チェックポイント管理例

BatchCheckpoint
- JobName
- ExecutionId
- LastProcessedId
- ProcessedCount
- UpdatedAt

処理例

Dim lastId As Long =
    _checkpointRepository.GetLastProcessedId("SalesImport")

Dim targets =
    _salesRepository.FindAfterId(lastId, 1000)

For Each target In targets

    Process(target)

    _checkpointRepository.Update(
        "SalesImport",
        target.Id)

Next

途中で障害が発生しても、最後に成功したIDから処理を再開できます。

チェックポイント設計のポイント

何を再開キーにするか決める
成功後にチェックポイントを更新する
処理とチェックポイント更新の整合性を考える
再実行時にも二重処理しないようにする

11. 再実行可能性

バッチ処理では、「一度失敗したら人間がDBを修正しないと再実行できない」という設計を避けるべきです。

再実行可能にするための設計

RequestIdや処理キーを持つ
処理済みデータを識別できるようにする
チェックポイントを保存する
冪等性を確保する
途中結果を管理する
失敗データだけ再処理できるようにする

ステータス管理例

ImportStatus
- Waiting
- Processing
- Completed
- Failed

再処理対象取得

SELECT *
FROM ImportQueue
WHERE Status IN ('Waiting', 'Failed')

失敗データを明確に管理すれば、成功済みデータを再処理せずに済みます。

12. リトライ設計

外部APIやDB通信では、一時的な障害が発生することがあります。この場合、一定回数のリトライが有効です。

指数バックオフの考え方

1回目：1秒後
2回目：2秒後
3回目：4秒後
4回目：8秒後

障害直後に連続リクエストするのではなく、徐々に待ち時間を増やします。

簡易実装例

Public Async Function ExecuteWithRetryAsync(
    action As Func(Of Task)) As Task

    Dim maxRetry As Integer = 4

    For retry As Integer = 0 To maxRetry - 1

        Try
            Await action()
            Return

        Catch ex As TimeoutException

            If retry = maxRetry - 1 Then
                Throw
            End If

            Dim waitSeconds As Integer =
                CInt(Math.Pow(2, retry))

            Await Task.Delay(
                TimeSpan.FromSeconds(waitSeconds))

        End Try

    Next

End Function

リトライ対象は、一時的な障害に限定する必要があります。入力エラーや業務エラーをリトライしても成功しません。

13. 大量CSV取込

大量CSV取込は、VB.NET業務システムで代表的なバッチ処理です。

推奨フロー

CSV
 ↓
ファイル検証
 ↓
行解析
 ↓
形式チェック
 ↓
一時テーブルへBulk Insert
 ↓
業務チェック
 ↓
本テーブルへ反映
 ↓
結果記録

一時テーブルを利用するメリット

Bulk Insertできる
DB側で一括検証できる
重複データを検出しやすい
エラー行を管理しやすい
本テーブルを直接汚さない

処理結果例

総件数     : 100,000
正常件数   : 99,742
エラー件数 : 258
処理時間   : 2分13秒

大量CSV取込で考慮すること

文字コード
CSVエスケープ
ファイルサイズ
行番号
重複チェック
Bulk Insert
途中エラー
再取込

14. 帳票一括生成

数百件、数千件のPDF帳票を一括生成する場合も、バッチアーキテクチャが重要です。

単純な実装

For Each invoice In invoices
    GeneratePdf(invoice)
Next

処理件数が少なければ問題ありませんが、大量データになると時間がかかります。

並列化の例

Dim options As New ParallelOptions With {
    .MaxDegreeOfParallelism = 4
}

Parallel.ForEach(
    invoices,
    options,
    Sub(invoice)

        GeneratePdf(invoice)

    End Sub)

ただし、利用するPDFライブラリがスレッドセーフか確認する必要があります。

帳票一括生成で注意すること

フォント
ファイル名重複
ディスク容量
同時生成数
PDFライブラリのスレッドセーフ性
失敗帳票だけ再作成できる設計

15. ジョブ排他制御

同じバッチが二重起動すると、二重登録や競合が発生する可能性があります。

ジョブロックテーブル例

BatchLocks
- JobName
- LockedAt
- LockedBy
- ExpiresAt

起動時チェック

If _batchLockRepository.IsLocked("DailySalesBatch") Then
    Throw New BusinessException(
        "DailySalesBatchは既に実行中です。")
End If

ロック設計で重要な点

異常終了時にロックが残らないようにする
有効期限を設定する
サーバー名やプロセスIDを記録する
運用担当者が解除できる仕組みを作る

単なるBooleanフラグだけでは、異常終了時に永久ロックになる危険があります。期限や所有者情報を持たせる方が安全です。

16. バッチ監視とログ

画面処理と違い、バッチ処理は人間が実行中の状態を直接見ていないことがあります。そのため、ログと監視は非常に重要です。

記録すべき情報

ジョブ名
実行ID
開始時刻
終了時刻
処理件数
成功件数
失敗件数
処理時間
リトライ回数
エラー内容

ログ例

[INFO]
Job=DailySalesBatch
ExecutionId=202609010001
Status=Start

[INFO]
Processed=10000
Success=9998
Error=2

[INFO]
Status=Completed
Elapsed=00:03:24

監視対象

予定時間になっても開始していない
想定時間を超えて実行中
Failedになっている
エラー件数が閾値を超えている
キュー滞留件数が増加している

「失敗したらログを見る」のではなく、「異常を自動検知できる」状態を目指すことが重要です。

17. パフォーマンス計測

並列化や非同期化を行う場合は、変更前後で実際の性能を計測します。

Stopwatchによる計測例

Dim sw As New Diagnostics.Stopwatch()

sw.Start()

Await ExecuteBatchAsync()

sw.Stop()

Logger.Info(
    "BatchElapsed=" &
    sw.Elapsed.ToString())

確認すべき指標

総処理時間
1件あたりの処理時間
CPU使用率
メモリ使用量
DB接続数
DB負荷
外部APIリクエスト数
エラー率

処理時間が半分になっても、DB負荷が10倍になっているのであれば良い改善とは限りません。システム全体を見て判断します。

18. 実務でありがちな失敗

何でもParallel.ForEachにする

並列化すると必ず速くなるわけではありません。 DBやAPIがボトルネックの場合、同時実行数を増やすことで逆に遅くなることがあります。

100万件を一度にメモリへ読み込む

大量データをListやDataTableへ一括ロードするとメモリを大量消費します。ページング、ストリーミング、チャンク処理を検討します。

途中再開できない

8時間かかるバッチが7時間59分で失敗し、最初から再実行しかできない設計は運用上非常に危険です。チェックポイントを設計します。

リトライで二重処理する

冪等性がない状態で再実行すると、登録、決済、メール、ポイントなどが二重処理される可能性があります。

ジョブの二重起動を考えていない

手動実行とスケジュール実行が重なるなど、実際の運用では二重起動が発生する可能性があります。ジョブ排他を設計する必要があります。

ログが開始と終了しかない

大量処理では、途中経過や成功件数、エラー対象を記録しないと障害調査が困難になります。

19. 演習課題

演習1：夜間バッチを設計する

毎日午前2時に売上データを集計するバッチを想定し、ジョブ管理、実行状態、ログ、異常終了時の処理を設計してください。

演習2：SemaphoreSlimで並列度を制御する

100件の外部API送信を最大5件ずつ並列実行する処理を実装してください。

演習3：チェックポイントを実装する

100万件のデータを1000件単位で処理し、障害発生後に最後に成功した位置から再開できるようにしてください。

演習4：冪等なキュー処理を設計する

同じQueueIdが複数回実行されても、実処理が1度だけ行われるようにしてください。

演習5：大量CSV取込を設計する

50万件のCSVファイルを対象に、解析、一時テーブル登録、業務チェック、本テーブル反映、エラー管理まで設計してください。

演習6：ジョブ二重起動を防止する

BatchLocksテーブルを利用し、同じバッチが同時実行されない仕組みを実装してください。異常終了によるロック残留も考慮してください。

演習7：性能比較を行う

逐次処理、並列度2、並列度5、並列度10で処理時間とDB負荷を計測し、最適な並列度を考察してください。

20. まとめ

本章では、VB.NET業務システムにおける非同期処理、並列処理、バッチ処理の実践設計について学習しました。

大量処理では、単に処理速度を上げるだけでは十分ではありません。途中障害、再実行、二重処理、ジョブ競合、外部システム障害など、本番運用で発生する問題を想定した設計が必要です。

バッチアーキテクチャでは、Scheduler、BatchRunner、Application Service、Repository、ジョブ状態管理、ログなどを分離し、処理本体と運用制御を切り離します。

キュー方式を利用すると、オンライン処理と重い後続処理を分離でき、失敗処理の再実行や並列度制御も行いやすくなります。

非同期処理では、I/O処理にはAsync / Awaitを活用し、 CPU負荷の高い処理では必要に応じてTask.Runや並列処理を利用します。

並列処理では、SemaphoreSlimなどを利用して同時実行数を制御します。最大並列度は、CPUだけでなくDB接続数、外部API制限、メモリ、ネットワークなどを考慮して決める必要があります。

大量データ処理では、チャンク処理とチェックポイント設計が重要です。途中で障害が発生しても、最後に正常処理した位置から再開できるようにすることで、運用負荷を大幅に下げられます。

また、リトライ処理を設計する場合は、必ず冪等性を考慮します。同じ処理が複数回実行されてもデータが壊れない仕組みを持つことが、本番バッチでは非常に重要です。

ジョブの二重起動、キュー滞留、処理時間超過なども監視対象です。バッチは「正常に動けば終わり」ではなく、失敗したときに迅速に検知・復旧できることまで含めて設計します。

本章のゴールは、大量処理を安全かつ高速に実行し、障害時にも再実行可能なバッチシステムを設計できるようになることです。非同期・並列処理の技術だけでなく、整合性、運用、監視、復旧まで含めて考えられることが、エンタープライズVB.NETエンジニアに求められる重要な能力です。