C# バッチ処理の入口と処理本体を分けて変更しやすくする

毎日決まった時刻に古いデータを削除したり、通知を受け取って未処理のデータを処理したりする機能があります。

こうした、画面操作とは別に実行するまとまった処理を、ここではバッチ処理と呼びます。

今回は、バッチ処理の保守をしやすくするために、「いつ・どこから呼ぶか」と「何を処理するか」を分ける方法を紹介します。

題材の実装では、Azure Functionsで定期実行と通知受信を扱い、処理本体は別プロジェクトのC#クラスに置いています。Azure Functionsは、時刻やメッセージの到着などをきっかけにコードを実行する仕組みです。

この記事では、まず通常のC#クラスで考え方を説明します。掲載コードは.NET 8のコンソールアプリで動作を確認しています。コードは構成を学ぶための簡略例で、Azure Functionsの設定やデータベースへの接続処理は含めません。

入口にすべて書くと何が困るのか

定期実行されるメソッドの中へ、対象データの検索、処理、保存、ログ出力をまとめて書く方法は、処理が小さい間はわかりやすい構成です。

ただし、同じ処理を別の入口からも呼びたくなると、扱いにくくなります。

たとえば、普段は5分ごとに実行し、新しいデータが届いたときはすぐに実行する場合です。2つの入口へ処理をコピーすると、片方だけ修正して動作が食い違う可能性があります。

題材の実装では、定期実行とキュー通知の受信が、同じ処理本体を呼びます。キューは、別の処理から渡されたメッセージを受け取るための仕組みです。

定期実行の入口 ──┐
                ├── 共通の実行処理 ── 処理本体
通知受信の入口 ──┘

この例の通知は「処理を始めてほしい」という合図です。処理対象そのものはデータベースから取得するため、どちらの入口から呼ばれても同じ処理を使えます。

通知の本文にしか処理対象がないシステムなら、本文の読み取りと引数への変換が必要です。すべての通知処理を、引数なしで共通化できるわけではありません。

変更する理由ごとに置き場所を分ける

分ける目安は、次のようになります。

時刻の設定を変えるために業務ルールを読む必要がなくなり、業務ルールを変えるために通知受信の設定を触る必要もなくなります。

題材の実装でも、入口は処理本体、実行を識別するID、停止要求を渡す短いコードになっています。処理本体の公開メソッドは、Azure Functions固有の実行情報を引数に取りません。

処理本体を通常のC#クラスにする

ここでは、未処理データを順番に処理する例を考えます。

using System.Threading;
using System.Threading.Tasks;

public interface IPendingWorkProcessor
{
    Task<bool> ProcessNextAsync(CancellationToken cancellationToken);
}

public sealed class PendingWorkJob
{
    private readonly IPendingWorkProcessor processor;

    public PendingWorkJob(IPendingWorkProcessor processor)
    {
        this.processor = processor;
    }

    public async Task ExecuteAsync(CancellationToken cancellationToken)
    {
        while (true)
        {
            cancellationToken.ThrowIfCancellationRequested();

            var processed = await processor.ProcessNextAsync(cancellationToken);
            if (!processed)
            {
                return;
            }
        }
    }
}

ProcessNextAsyncは、1件処理したらtrue、処理対象がなければfalseを返す想定です。データベースの検索や処理済みへの更新は、その実装側が担当します。

CancellationTokenは、アプリの停止などを処理側へ知らせるものです。ループの途中でも停止要求を確認し、呼び出す処理にも渡しています。トークンを渡すだけで処理が強制停止されるわけではないため、受け取った側でも対応が必要です。

このクラスには、実行時刻やキュー名を書いていません。入口側からExecuteAsyncを呼ぶだけで処理できます。

この簡略例は対象がなくなるまで処理します。データが絶えず追加される場合などは、1回の処理件数や実行時間にも上限を設けます。

共通にするのは、すべての処理で意味が同じ部分

入口ごとにログ出力や例外処理を書くと、記録する情報や失敗の扱いが少しずつ変わりやすくなります。

題材の実装では、共通の実行クラスで、実行を識別するIDをログへ付け、処理本体を呼び出しています。そこで捕捉した通常の例外は、失敗をログに残した後に再び投げています。

一方、データごとの再試行や失敗状態の保存は、処理本体側が扱っています。

単にtryとcatchが似ているからという理由で、すべてのエラー処理を共通化しないことが大切です。

停止要求についても扱いを決めます。題材の共通実行クラスは、渡されたトークンの停止要求に対応する例外を捕捉して終了します。この扱いをそのまま別の処理へ移すのではなく、途中まで進んだ処理を次回どう再開するかと合わせて決めます。

Azure Functionsの停止要求の受け取り方は、Microsoft Learnの分離ワーカープロセスの説明で確認できます。

実行環境を起動せず、処理本体を確認する

処理本体が通常のクラスなら、テスト用の処理を渡して確認できます。Azure Functionsを起動しなくても、ループの終了や停止要求への対応を試せます。

以下は、2件処理すると対象がなくなるテスト用クラスです。

using System.Threading;
using System.Threading.Tasks;

public sealed class FakePendingWorkProcessor : IPendingWorkProcessor
{
    public int CallCount { get; private set; }

    public Task<bool> ProcessNextAsync(CancellationToken cancellationToken)
    {
        cancellationToken.ThrowIfCancellationRequested();
        CallCount++;
        return Task.FromResult(CallCount <= 2);
    }
}

次のコードをコンソールアプリのProgram.csへ書けば、呼び出し回数を確認できます。クラスはそれぞれ別ファイルへ置く想定です。

using System;
using System.Threading;

var processor = new FakePendingWorkProcessor();
var job = new PendingWorkJob(processor);

await job.ExecuteAsync(CancellationToken.None);

Console.WriteLine(processor.CallCount); // 3

2回の処理と、「もう対象がない」ことを確認する1回で、合計3回になります。

ここで確認できるのは処理本体の動作です。キュー名の設定、定期実行の時刻、DIの登録、実際のデータベース処理は、別途確認する必要があります。

分割しても重複実行は防げない

定期実行と通知受信が同じ処理本体を呼べることと、同時に呼ばれても正しく処理できることは別です。

2つの実行が同じ未処理データを取得すると、どちらも処理してしまう可能性があります。

題材の実装では、処理対象を取得するときに、一定時間そのデータを処理する権利を確保する仕組みも使っています。この記事のPendingWorkJobには、その機能は含めていません。

実際のアプリでは、対象を確保する処理や、同じ操作が再実行されても結果が重複しない仕組みを組み合わせます。

再送による重複処理を防ぐ考え方

小さな処理なら2つに分けるだけでもよい

題材の実装は複数のバッチで共通の実行クラスを使っていますが、最初から同じ構成をそろえる必要はありません。

バッチが1つだけなら、入口と処理本体の2つに分けるだけでも、処理本体をテストしやすくなります。複数の入口で同じログ処理を繰り返すようになってから、共通の実行クラスを追加する方法もあります。

反対に、数行で終わり、別の入口から呼ぶ予定もない処理なら、入口のメソッド内に書いた方が読みやすいこともあります。

判断の目安は、実行方法を変えるときにも、業務ルールを変えるときにも、同じ大きなメソッドを修正しているかです。変更理由が混ざり始めたら、入口と処理本体の境界を作ることを検討してみてください。