バッチ処理も実はAPI

環境作成

通常バッチ処理と聞いたら『Cron』が起動し『hoge.sh』が叩かれ、結果JAVAファイルが発火する様な流れです。
しかしJakartaの標準的な方法は『Cron』が起動し『API』を叩きます。
通常のAPIと違うのは『非同期』という点だけです。

○ pom.xml

まずはお約束の依存関係を整えます。

pom.xml

<dependency>
    <groupId>jakarta.platform</groupId>
    <artifactId>jakarta.jakartaee-api</artifactId>
    <version>10.0.0</version>
    <scope>provided</scope>
</dependency>

○ エンドポイント作成

APIなので当然エンドポイントが必要となります。
パッケージ『api.batch』に作ります。

src/main/java/api/batch/BatchResource.java

上記設定で、エンドポイントは『http://localhost:8080/sample-pj/api/batch/start』となります。
※Path『api』は《@ApplicationPath("/api")》で指定しています

流れとしては、まず『JobOperator』型のインスタンスを作成します。
そしてインスタンスに対し『start』メソッドを発行します。
非同期処理なので、即座にレスポンスが返却されます。

ここで『start』メソッドの第１引数に注目して下さい。
文字列『my-job』とは何でしょう？
実はこれ『src/main/resources/META-INF/batch-jobs/my-job.xml』の省略形何です。
※「my-job」以外は《パス》《拡張子》全て固定となります

○「my-job.xml」を作成する

では早速『src/main/resources/META-INF/batch-jobs/my-job.xml』作成します。

my-job.xml

<?xml version="1.0" encoding="UTF-8"?>
<job id="my-job" xmlns="https://jakarta.ee/xml/ns/jakartaee" version="2.0">
    <step id="batch_1">
        <batchlet ref="myBatchJobFirst"/>
    </step>
</job>

『ID属性名』は任意で良いです。
batchletタグの『ref属性』に指定したのが発火するクラス名となります。
サンプルでは上記XMLファイルを元に解説していきますが、実はこのXMLは柔軟な設定が可能です。

XMLファイルの詳細設定

先ほどの『my-job.xml』のサンプルは最も単純なベース的内容となります。
それではシチュエーションに合わせたサンプルをご紹介します。

○ 順番に指定クラスを発火させる

my-job.xml

<?xml version="1.0" encoding="UTF-8"?>
<job id="my-job" xmlns="https://jakarta.ee/xml/ns/jakartaee" version="2.0">
    
    <step id="step1" next="step2">
        <batchlet ref="downloadBatchlet"/>
    </step>
    
    <step id="step2" next="step3">
        <batchlet ref="processDataBatchlet"/>
    </step>
    
    <step id="step3">
        <batchlet ref="uploadBatchlet"/>
    </step>
</job>

『ID属性名』『next属性名』を使い、順番を設定できます。
この場合[downloadBatchlet] > [processDataBatchlet] > [uploadBatchlet]で処理クラスが順番に発火します。

○ 処理の状態によって発火するクラスを制御する

my-job.xml

<?xml version="1.0" encoding="UTF-8"?>
<job id="my-job" xmlns="https://jakarta.ee/xml/ns/jakartaee" version="2.0">
    
    <step id="downloadStep">
        <batchlet ref="downloadBatchlet"/>
        
        <next on="COMPLETED" to="processStep"/>
        
        <next on="FAILED" to="errorNotificationStep"/>
    </step>
    
    <step id="processStep">
        <batchlet ref="processDataBatchlet"/>
    </step>
    
    <step id="errorNotificationStep">
        <batchlet ref="sendErrorMailBatchlet"/>
    </step>
</job>

『downloadBatchlet』クラスの処理が正常終了なら『processDataBatchlet』が発火します。
また失敗の場合は『sendErrorMailBatchlet』クラスが発火します。

バッチ実行クラス作成

『batch-jobs/my-job.xml』ファイルに記載した『batchlet ref="myBatchJobFirst"』に従い作成します。
場所は何処でも良いですが、エンドポイントのパッケージを『api.batch』としたので、
実行クラスは『api.job』とし『MyBatchJobFirst』を作成します。

MyBatchJobFirst.java

package api.job;

import jakarta.batch.api.AbstractBatchlet;
import jakarta.batch.runtime.BatchStatus;
import jakarta.enterprise.context.Dependent;
import jakarta.inject.Named;

@Named("myBatchJobFirst")
@Dependent
public class MyBatchJobFirst extends AbstractBatchlet {
    @Override
    public String process() throws Exception {
        // ここにバッチのメイン処理を記述
        System.out.println("myBatchJobFirst の処理を実行中...");

        // 正常終了時は "COMPLETED"（または独自の終了ステータス文字列）を返す
        return BatchStatus.COMPLETED.toString();
    }
    @Override
    public void stop() throws Exception {
        // 非同期キャンセル要求時のクリーンアップ処理（任意）
        System.out.println("処理の中断要求を受け取りました。");
    }
}

作成出来たらPostMan等を使ってエンドポイントにPOSTリクエストしてみましょう。
URLは『http://localhost:8080/sample-pj/api/batch/start』

ポイント

URLを見て「おや」と思った方いませんか？
そう『api』なんて設定何処でと思っていませんか。
この設定は通常のAPIの章で作成した『RestApiService』クラスに付与した以下アノテーションです。

@ApplicationPath("/api")

またPostManが手元にない場合は、Windowsのターミナルから叩いても連携できますよ。
curl -X POST "http://localhost:8080/sample-pj/api/batch/start" -H "Content-Type: application/json" -d "{}"

パラメータを渡す方法

バッチと言っても『API』なので当然パラメータを渡す事は可能です。
でもちょっと待ってください。
折角『CRON』から直接APIを叩けるのに、動的にパラメータを付与するとなると『Shell』が必要になります。
本当にそのパラメータは必要ですか？

バッチは本来『夜間データ整理』等の単純な処理を担当するものです。
必要なパラメータってせいぜい日付くらいで、しかも日付は発火した後にJAVA側で普通に取得できます。

ただし、例えば大量のデータがあり、午前１時はDBの奇数行の処理、午前５時は偶数みたいな場合には、
それぞれの『CRON』にパラメータ『1』や『5』を付加するのはありかもしれませんね。

○ [KEY -> VALUE]で受取る方法

ターミナルだと以下の様なリクエストに対応させます。
curl -X POST "http://localhost:8080/sample-pj/api/batch/start?name_cd=foo"

BatchResource.java

package api.batch;

import java.util.Properties;
import jakarta.batch.operations.JobOperator;
import jakarta.batch.runtime.BatchRuntime;
import jakarta.enterprise.context.RequestScoped;
import jakarta.ws.rs.POST;
import jakarta.ws.rs.Path;
import jakarta.ws.rs.QueryParam;
import jakarta.ws.rs.core.Response;

@Path("/batch")
@RequestScoped
public class BatchResource {
    @POST
    @Path("/start")
    public Response startJob(@QueryParam("name_cd") String nameCd) {
        // JobOperator に渡す Properties オブジェクトを作成
        Properties jobParameters = new Properties();
        // パラメータが存在する場合のみ設定（未指定時のデフォルト動作を担保するため）
        if (nameCd != null && !nameCdnameCd.isEmpty()) {
            jobParameters.setProperty("name_cd", nameCd);
        }
        // バッチ操作用のオブジェクトを取得
        JobOperator jobOperator = BatchRuntime.getJobOperator();

        // my-job.xml で定義したバッチを非同期で開始
        long executionId = jobOperator.start("my-job", jobParameters);

        // ジョブIDを返して即座にレスポンス（同期実行でクライアントを待たせない）
        return Response.accepted("Job started with execution ID: " + executionId).build();
    }
}

『QueryParam("name_cd")』は、リクエスト側が設定するKEYの名前です。
『String nameCd』は、JAVAコード内で扱う変数名です。
『jobParameters』型のインスタンスを生成し、値をセットします。
バッチ実行メソッド『jobOperator.start』の第２引数に渡します。

○『my-job.xml』にもパラメータを登録

my-job.xml

<?xml version="1.0" encoding="UTF-8"?>
<job id="my-job" xmlns="https://jakarta.ee/xml/ns/jakartaee" version="2.0">
    <step id="batch_1">
        <batchlet ref="myBatchJobFirst">
            <properties>
                <property name="name_cd" value="#{jobParameters['name_cd']}"/>
            </properties>
        </batchlet>
    </step>
</job>

ちょっと面倒ですが、全てのバッチは全て『XML』ファイルを経由します。
したがって、該当ファイルにもパラメータをセットする必要があります。

○ バッチメイン処理からパラメータを取得する

今回の設定では、実際にバッチを処理するクラスは『MyBatchJobFirst』となります。
このクラスでパラメータを取得しなければ意味がありませんよね。

MyBatchJobFirst.java

package com.example.batch;

import jakarta.batch.api.AbstractBatchlet;
import jakarta.batch.api.BatchProperty;
import jakarta.batch.runtime.BatchStatus;
import jakarta.enterprise.context.Dependent;
import jakarta.inject.Inject;
import jakarta.inject.Named;

@Named("myBatchJobFirst")
@Dependent
public class MyBatchJobFirst extends AbstractBatchlet {
    @Inject
    @BatchProperty(name = "name_cd")
    private String nameCd;

    @Override
    public String process() throws Exception {
        // ここにバッチのメイン処理を記述
        System.out.println("myBatchJobFirst の処理を実行中...");

        // 正常終了時は "COMPLETED"（または独自の終了ステータス文字列）を返す
        return BatchStatus.COMPLETED.toString();
    }

    @Override     public void stop() throws Exception {
        // 非同期キャンセル要求時のクリーンアップ処理（任意）
        System.out.println("処理の中断要求を受け取りました。");
    }
}

『@BatchProperty(name = "name_cd")』と『@Inject』を付与する事で、ようやくパラメータが取得できます。

裏技

本来のJakarta EEの思想からは外れますが『XML』ファイルにパラメータを記述しない方法もあります。
先ずは『@Inject』の部分を以下の様に変更します。

@Inject
private JobContext jobContext;

次にメソッド内を以下の様に修正する事で直接値を参照できます。

long executionId = jobContext.getExecutionId();
JobOperator jobOperator = BatchRuntime.getJobOperator();
Properties jobParams = jobOperator.getParameters(executionId);
String nameCd = jobParams.getProperty("name_cd");

○ JSONで受取る方法

JSONで受取るには、まずはクラスでJSONを宣言し『Map型』で受取ります。

BatchResource.java

import jakarta.ws.rs.Consumes;
import jakarta.ws.rs.core.MediaType;
～省略～
@POST
@Path("/start")
@Consumes(MediaType.APPLICATION_JSON)
public Response startJob(Map<String, String> requestParams) {
    Properties jobParameters = new Properties();
    // パラメータが存在する場合のみ設定（未指定時のデフォルト動作を担保するため）
    if (requestParams != null ) {
        requestParams.forEach((key, value) -> {
            if (value != null) {
                jobParameters.setProperty(key, value);
            }
        });
    }
～省略～

その後の設定方法は同じですので『正攻法』『裏技』どちらでも値を取得できます。

Batchlet方式によるバッチ処理

バッチには実は２種類の処理方法があります。
後述する『Chunk方式』とは違い、Batchlet方式は１つのタスクを単一の処理として実行する方式です。
因みに前述のXMLファイルの設定方法は『Batchlet方式』となります。
サンプルコードは以下となります。

MyBatchJobFirst.java

package api.job;

import jakarta.batch.api.AbstractBatchlet;
import jakarta.batch.runtime.BatchStatus;
import jakarta.enterprise.context.Dependent;
import jakarta.inject.Named;

@Named("myBatchJobFirst")
@Dependent
public class MyBatchJobFirst extends AbstractBatchlet {
    private boolean stoped = false;
    @Override
    public String process() throws Exception {
        // 正常終了時は "COMPLETED"（または独自の終了ステータス文字列）を返す
        return this.stoped ? BatchStatus.STOPPED.toString() : BatchStatus.COMPLETED.toString();
     }

    @Override
    public void stop() throws Exception {
        // ジョブの中断要求があった場合の処理（必要に応じて実装）
        this.stoped = true;
    }
}

処理は『process()』メソッドから始まります。
『stop()』メソッドは任意ですが、設置する場合はフラグで状態を判断できる様にします。
サンプルでは、フラグが立っている場合は《STOPPED》を返却する様にしています。

ポイント

[COMPLETED][STOPPED]の他に、以下定数が用意されています。

・BatchStatus.STARTING.toString()
・BatchStatus.STARTED.toString()
・BatchStatus.STOPPING.toString()
・BatchStatus.COMPLETED.toString()
・BatchStatus.FAILED.toString()
・BatchStatus.ABANDONED.toString()
・BatchStatus.FAILING.toString()

Chunk方式によるバッチ処理

大量のデータを一定の件数ごとに区切って「読込 → 変換・加工 → 書込」を繰り返す処理です。
全件を一括処理するのではなく、設定した件数（例:1000件）毎にトランザクションをコミットするため、
メモリの消費を抑えつつ、万が一途中でエラーが発生しても途中経過を保持・リカバリしやすい特徴があります。
JAVAコード的な特徴としては、３つのクラスを作成してバッチ処理します。

基本的な構造（[エンドポイント] > [XMLファイル] > [Java処理]）は同じですが、少し勝手が違うのでエンドポイントからの説明となります。

BatchResource.java

package api.batch;

import jakarta.batch.operations.JobOperator;
import jakarta.batch.runtime.BatchRuntime;
import jakarta.enterprise.context.RequestScoped;
import jakarta.ws.rs.POST;
import jakarta.ws.rs.Path;
import jakarta.ws.rs.core.Response;

@Path("/batch")
@RequestScoped
public class BatchResource {
    @POST
    @Path("/chunk")
    public Response chunkJob() {
        // バッチ操作用のオブジェクトを取得
        JobOperator jobOperator = BatchRuntime.getJobOperator();
        // chunk-job.xml で定義したバッチを非同期で開始
        long executionId = jobOperator.start("chunk-job", null);

        // ジョブIDを返して即座にレスポンス（同期実行でクライアントを待たせない）
        return Response.accepted("Job started with execution ID: " + executionId).build();
    }
}

エンドポイントに関しては『Batchlet方式』と同じです。
次にXMLファイルを作成します。

chunk-job.xml

<?xml version="1.0" encoding="UTF-8"?>
<job id="chunk-job" xmlns="https://jakarta.ee/xml/ns/jakartaee" version="2.0">
    <step id="chunkStep">
        
        <chunk item-count="3">
            <reader ref="myReader" />
            <processor ref="myProcessor" />
            <writer ref="myWriter" />
        </chunk>
    </step>
</job>

先ほどの説明通り、コミットするまでの間隔を設定できます（３回）
また以下役割を持つクラスを宣言します。

・reader                      読込担当クラス                      myReader
・processor               処理担当クラス                      myProcessor
・writer                       書き込み担当クラス             myWriter

myReader.java

package api.job;

import jakarta.batch.api.chunk.AbstractItemReader;
import jakarta.enterprise.context.Dependent;
import jakarta.inject.Named;

@Named("myReader")
@Dependent
class MyReader extends AbstractItemReader {
    private int count = 0;
    @Override
    public Object readItem() throws Exception {
        count++;
        // 例: 10件まで読み込む
        if (count <= 10) {
            System.out.println("1件読込");
            return "Item" + count;
        }
        // これ以上データがない場合は null を返す
        return null;
    }
}

ここでの処理は『データの読込み』となります。
1件単位で読込み、次のクラスへ処理を渡します。
※本来はここでDB一覧のデータを読込んで『for文』で回すイメージです。

MyProcessor.java

package api.job;

import jakarta.batch.api.chunk.ItemProcessor;
import jakarta.enterprise.context.Dependent;
import jakarta.inject.Named;

@Named("myProcessor")
@Dependent
class MyProcessor implements ItemProcessor {
    @Override
    public Object processItem(Object item) throws Exception {
        String input = (String) item;
        // データを加工して返す
        System.out.println("データを加工して返す");
        return input.toUpperCase();
    }
}

主にデータの加工を担当するクラスです。
サンプルでは単純に渡されたデータを大文字変換しています。

さてここで問題です。
『return』はどのクラスに返却されるでしょうか？
正解は『３件』溜まるまでは『myReader』へ返却されます。
そしてこの『３』は、XMLファイルで設定した『chunk item-count="3"』となります。

MyWriter.java

package api.job;

import java.util.List;
import jakarta.batch.api.chunk.AbstractItemWriter;
import jakarta.enterprise.context.Dependent;
import jakarta.inject.Named;

@Named("myWriter")
@Dependent
class MyWriter extends AbstractItemWriter {
    @Override
    public void writeItems(List<Object> items) throws Exception {
        // まとめて一括書き込み（例: DBバッチインサートやファイル追記）
        System.out.println("データ数：" + items.size());
        for (Object item : items) {
            System.out.println("Writing: " + item);
        }
    }
}

３件溜まると本クラスに処理が移行します。
ここでは最終的な結果を処理します。

具体的には、DB登録や更新等を処理します。
※『for文』内で登録処理を記述します

トランザクション管理について

Batchlet方式による管理は、以下の様な通常の設定となるので特に意識せずにコードを記述できます。
@Transactional(value = Transactional.TxType.REQUIRED, rollbackOn = {Exception.class})

しかし、Chunk方式の場合はアノテーションによる上記管理を記述するとエラーとなります。
Chunkの場合は『Chunkコンテナ』が自動的にトランザクションを発行するためバッティングします。
手動でロールバックするには、単純に『RuntimeException』や『Exception』を投げます。