リスナー

Overview

リスナーとは、ジョブやステップを実行する前後に処理を挿入するためのインターフェースである。

本機能は、チャンクモデルとタスクレットモデルとで使い方が異なるため、それぞれについて説明する。

リスナーには多くのインターフェースがあるため、それぞれの役割について説明する。その後に、設定および実装方法について説明をする。

リスナーの種類

Spring Batchでは、実に多くのリスナーインターフェースが定義されている。ここではそのすべてを説明するのではなく、利用頻度が高いものを中心に扱う。

まず、リスナーは2種類に大別される。

JobListener: ジョブの実行に対して処理を挟み込むためのインターフェース
StepListener: ステップの実行に対して処理を挟み込むためのインターフェース

JobListenerについて

Spring Batchには、JobListenerという名前のインターフェースは存在しない。 StepListenerとの対比のため、本ガイドラインでは便宜的に定義している。
Java Batch(jBatch)には、javax.batch.api.listener.JobListenerというインターフェースが存在するので、実装時には間違えないように注意すること。また、StepListenerもシグネチャが異なる同名インターフェース(javax.batch.api.listener.StepListener)が存在するので、同様に注意すること。

JobListener

JobListenerのインターフェースは、JobExecutionListenerの1つのみとなる。

JobExecutionListener: ジョブの開始前、終了後に処理を挟み込む。

JobExecutionListenerインターフェース

public interface JobExecutionListener {
  void beforeJob(JobExecution jobExecution);
  void afterJob(JobExecution jobExecution);
}

StepListener

StepListenerのインターフェースは以下のように多くの種類がある。

StepListener: 以降に紹介する各種リスナーのマーカーインターフェース。

StepExecutionListener: ステップ実行の開始前、終了後に処理を挟み込む。

StepExecutionListenerインターフェース

public interface StepExecutionListener extends StepListener {
  void beforeStep(StepExecution stepExecution);
  ExitStatus afterStep(StepExecution stepExecution);
}

ChunkListener: 1つのチャンクを処理する前後と、エラーが発生した場合に処理を挟み込む。

ChunkListenerインターフェース

public interface ChunkListener extends StepListener {
  static final String ROLLBACK_EXCEPTION_KEY = "sb_rollback_exception";
  void beforeChunk(ChunkContext context);
  void afterChunk(ChunkContext context);
  void afterChunkError(ChunkContext context);
}

ROLLBACK_EXCEPTION_KEYの用途

afterChunkErrorメソッドにて、発生した例外を取得したい場合に利用する。 Spring Batchはチャンク処理中にエラーが発生した場合、ChunkContextにsb_rollback_exceptionというキー名で例外を格納した上でChunkListenerを呼び出すため、以下の要領でアクセスできる。

使用例

public void afterChunkError(ChunkContext context) {
    logger.error("Exception occurred while chunk. [context:{}]", context,
            context.getAttribute(ChunkListener.ROLLBACK_EXCEPTION_KEY));
}

例外ハンドリングについては、ChunkListenerインターフェースによる例外ハンドリングを参照。

ItemReadListener: ItemReaderが1件のデータを取得する前後と、エラーが発生した場合に処理を挟み込む。

ItemReadListenerインターフェース

public interface ItemReadListener<T> extends StepListener {
  void beforeRead();
  void afterRead(T item);
  void onReadError(Exception ex);
}

ItemProcessListener: ItemProcessorが1件のデータを加工する前後と、エラーが発生した場合に処理を挟み込む。

ItemProcessListenerインターフェース

public interface ItemProcessListener<T, S> extends StepListener {
  void beforeProcess(T item);
  void afterProcess(T item, S result);
  void onProcessError(T item, Exception e);
}

ItemWriteListener: ItemWriterが1つのチャンクを出力する前後と、エラーが発生した場合に処理を挟み込む。

ItemWriteListenerインターフェース

public interface ItemWriteListener<S> extends StepListener {
  void beforeWrite(List<? extends S> items);
  void afterWrite(List<? extends S> items);
  void onWriteError(Exception exception, List<? extends S> items);
}

本ガイドラインでは、以下のリスナーについては説明をしない。

リトライ系リスナー
スキップ系リスナー

これらのリスナーは例外ハンドリングでの使用を想定したものであるが、本ガイドラインではこれらのリスナーを用いた例外ハンドリングは行わない方針である。詳細は、例外ハンドリングを参照。

How to use

リスナーの実装と設定方法について説明する。

リスナーの実装

リスナーの実装と設定方法について説明する。

リスナーインターフェースをimplementsして実装する。
コンポーネントにメソッドベースでアノテーションを付与して実装する。

どちらで実装するかは、リスナーの役割に応じて選択する。基準は後述する。

インターフェースを実装する場合

各種リスナーインターフェースをimplementsして実装する。必要に応じて、複数のインターフェースを同時に実装してもよい。以下に実装例を示す。

JobExecutionListenerの実装例

@Component
public class JobExecutionLoggingListener implements JobExecutionListener { // (1)

    private static final Logger logger =
            LoggerFactory.getLogger(JobExecutionLoggingListener.class);

    @Override
    public void beforeJob(JobExecution jobExecution) { // (2)
        logger.info("job started. [JobName:{}]", jobExecution.getJobInstance().getJobName());
    }

    @Override
    public void afterJob(JobExecution jobExecution) { // (3)

        logger.info("job finished.[JobName:{}][ExitStatus:{}]", jobExecution.getJobInstance().getJobName(),
                jobExecution.getExitStatus().getExitCode());
    }
}

リスナーの設定例

<batch:job id="chunkJobWithListener" job-repository="jobRepository">
     <batch:step id="chunkJobWithListener.step01">
         <batch:tasklet transaction-manager="jobTransactionManager">
             <batch:chunk reader="reader" processor="processor"
                          writer="writer" commit-interval="10"/>
             <batch:listeners>
                 <batch:listener ref="loggingEachProcessInStepListener"/>
             </batch:listeners>
         </batch:tasklet>
     </batch:step>
     <batch:listeners>
         <batch:listener ref="jobExecutionLoggingListener"/> <!-- (4) -->
     </batch:listeners>
 </batch:job>

説明
項番	説明
(1)	`JobExecutionListener`を`implements`して実装する。
(2)	`JobExecutionListener`が定義している`beforeJob`メソッドを実装する。この例では、ジョブ開始ログを出力する。
(3)	`JobExecutionListener`が定義している`afterJob`メソッドを実装する。この例では、ジョブ終了ログを出力する。
(4)	Bean定義の`<listeners>`タグで、(1)で実装したリスナーを設定する。設定方法の詳細は、リスナーの設定で説明する。

リスナーのサポートクラス

複数のリスナーインターフェースをimplementsした場合、処理が不要な部分についても空実装をする必要がある。この作業を簡略化するため、あらかじめ空実装を施したサポートクラスがSpring Batchには用意されている。インターフェースではなく、サポートクラスを活用してもよいが、その場合implementsではなくextendsになるため注意すること。

サポートクラス

org.springframework.batch.core.listener.ItemListenerSupport
org.springframework.batch.core.listener.StepListenerSupport

アノテーションを付与する場合

各種リスナーインターフェースに対応したアノテーションを付与する。必要に応じて、複数のアノテーションを同時に実装してもよい。

リスナーインターフェースとの対応表
リスナーインターフェース	アノテーション
JobExecutionListener	`@beforeJob` `@afterJob`
StepExecutionListener	`@BeforeStep` `@AfterStep`
ChunkListener	`@BeforeChunk` `@AfterChunk` `@afterChunkError`
ItemReadListener	`@BeforeRead` `@AfterRead` `@OnReadError`
ItemProcessListener	`@beforeProcess` `@afterProcess` `@onProcessError`
ItemWriteListener	`@BeforeWrite` `@AfterWrite` `@OnWriteError`

これらアノテーションはコンポーネント化された実装のメソッドに付与することで目的のスコープで動作する。以下に実装例を示す。

アノテーションを付与したItemProcessorの実装例

@Component
public class AnnotationAmountCheckProcessor implements
        ItemProcessor<SalesPlanDetail, SalesPlanDetail> {

    private static final Logger logger =
            LoggerFactory.getLogger(AnnotationAmountCheckProcessor.class);

    @Override
    public SalesPlanDetail process(SalesPlanDetail item) throws Exception {
        if (item.getAmount().signum() == -1) {
            throw new IllegalArgumentException("amount is negative.");
        }
        return item;
    }

    // (1)
    /*
    @BeforeProcess
    public void beforeProcess(Object item) {
        logger.info("before process. [Item :{}]", item);
    }
    */

    // (2)
    @AfterProcess
    public void afterProcess(Object item, Object result) {
        logger.info("after process. [Result :{}]", result);
    }

    // (3)
    @OnProcessError
    public void onProcessError(Object item, Exception e) {
        logger.error("on process error.", e);
    }
}

リスナーの設定例

<batch:job id="chunkJobWithListenerAnnotation" job-repository="jobRepository">
    <batch:step id="chunkJobWithListenerAnnotation.step01">
        <batch:tasklet transaction-manager="jobTransactionManager">
            <batch:chunk reader="reader"
                         processor="annotationAmountCheckProcessor"
                         writer="writer" commit-interval="10"/>  <! -- (4) -->
        </batch:tasklet>
    </batch:step>
</batch:job>

説明
項番	説明
(1)	アノテーションで実装する場合は、処理が必要なタイミングのアノテーションのみを付与すればよい。この例では、ItemProcessの処理前には何もする必要がないため、`@beforeProcess`を付与した実装は不要となる。
(2)	ItemProcessの処理後に行う処理を実装する。この例では処理結果をログを出力している。
(3)	ItemProcessでエラーが発生したときの処理を実装する。この例では発生した例外をログを出力している。
(4)	アノテーションでリスナー実装がされているItemProcessを`<chunk>`タグに設定する。リスナーインターフェースとは異なり、`<listener>`タグで設定しなくても、自動的にリスナーが登録される。

アノテーションを付与するメソッドの制約

アノテーションを付与するメソッドはどのようなメソッドでもよいわけではない。対応するリスナーインターフェースのメソッドと、シグネチャを一致させる必要がある。この点は、各アノテーションのjavadocに明記されている。

JobExecutionListenerをアノテーションで実装したときの注意

JobExecutionListenerは、他のリスナーとスコープが異なるため、上記の設定では自動的にリスナー登録がされない。そのため、<listener>タグで明示的に設定する必要がある。詳細は、リスナーの設定を参照。

Tasklet実装へのアノテーションによるリスナー実装

Tasklet実装へのアノテーションによるリスナー実装した場合、以下の設定では一切リスナーが起動しないため注意する。

Taskletの場合

<batch:job id="taskletJobWithListenerAnnotation" job-repository="jobRepository">
    <batch:step id="taskletJobWithListenerAnnotation.step01">
        <batch:tasklet transaction-manager="jobTransactionManager"
                       ref="annotationSalesPlanDetailRegisterTasklet"/>
    </batch:step>
</batch:job>

タスクレットモデルの場合は、インターフェースとアノテーションの使い分けに従ってリスナーインターフェースを利用するのがよい。

リスナーの設定

リスナーは、Bean定義の<listeners>.<listener>タグによって設定する。 XMLスキーマ定義では様々な箇所に記述できるが、インターフェースの種類によっては意図とおり動作しないものが存在するため、以下の位置に設定すること。

リスナーを設定する位置

<!-- for chunk mode -->
<batch:job id="chunkJob" job-repository="jobRepository">
    <batch:step id="chunkJob.step01">
        <batch:tasklet transaction-manager="jobTransactionManager">
            <batch:chunk reader="(1)"
                         processor="(1)"
                         writer="(1)" commit-interval="10"/>
            <batch:listeners>
                <batch:listener ref="(2)"/>
            </batch:listeners>
        </batch:tasklet>
    </batch:step>
    <batch:listeners>
        <batch:listener ref="(3)"/>
    </batch:listeners>
</batch:job>

<!-- for tasklet mode -->
<batch:job id="taskletJob" job-repository="jobRepository">
    <batch:step id="taskletJob.step01">
        <batch:tasklet transaction-manager="jobTransactionManager" ref="tasklet">
            <batch:listeners>
                <batch:listener ref="(2)"/>
            </batch:listeners>
        </batch:tasklet>
    </batch:step>
    <batch:listeners>
        <batch:listener ref="(3)"/>
    </batch:listeners>
</batch:job>

設定値の説明
項番	説明
(1)	StepListenerに属するアノテーションによる実装を含んだコンポーネントを設定する。アノテーションの場合、必然的にこの場所に設定することになる。
(2)	StepListenerに属するリスナーインターフェース実装を設定する。タスクレットモデルの場合、`ItemReadListener`,`ItemProcessListener`,`ItemWriteListener`は利用できない。
(3)	JobListenerに属するリスナーを設定する。インターフェースとアノテーション、どちらの実装でもここに設定する必要がある。

複数リスナーの設定

<batch:listeners>タグには複数のリスナーを設定することができる。

複数のリスナーを登録したときに、リスナーがどのような順番で起動されるかを以下に示す。

ItemProcessListener実装
- listenerA, listenerB
JobExecutionListener実装
- listenerC, listenerD

複数リスナーの設定例

<batch:job id="chunkJob" job-repository="jobRepository">
    <batch:step id="chunkJob.step01">
        <batch:tasklet transaction-manager="jobTransactionManager">
            <batch:chunk reader="reader"
                         processor="pocessor"
                         writer="writer" commit-interval="10"/>
            <batch:listeners>
                <batch:listener ref="listenerA"/>
                <batch:listener ref="listenerB"/>
            </batch:listeners>
        </batch:tasklet>
    </batch:step>
    <batch:listeners>
        <batch:listener ref="listenerC"/>
        <batch:listener ref="listenerD"/>
    </batch:listeners>
</batch:job>

リスナーの起動順序

前処理に該当する処理は、リスナーの登録順に起動される。
後処理またはエラー処理に該当する処理は、リスナー登録の逆順に起動される。

インターフェースとアノテーションの使い分け

リスナーインターフェースとアノテーションによるリスナーの使い分けを説明する。

リスナーインターフェース: job、step、chunkにおいて共通する横断的な処理の場合に利用する。
アノテーション: ビジネスロジック固有の処理を行いたい場合に利用する。
原則として、ItemProcessorに対してのみ実装する。

StepExecutionListenerでの前処理における例外発生

前処理(beforeStepメソッド)で例外が発生した場合、モデルによりリソースのオープン/クローズの実行有無が変わる。それぞれのモデルにおいて前処理で例外が発生した場合について説明する。

チャンクモデル: リソースのオープン前に前処理が実行されるため、リソースのオープンは行われない。
リソースのクローズは、リソースのオープンがされていない場合でも実行されるため、ItemReader/ItemWriterを実装する場合にはこのことに注意する必要がある。
タスクレットモデル: タスクレットモデルでは、executeメソッド内で明示的にリソースのオープン/クローズを行う。
前処理で例外が発生すると、executeメソッドは実行されないため、当然リソースのオープン/クローズも行われない。

前処理(StepExecutionListener#beforeStep())でのジョブの打ち切り

ジョブを実行する条件が整っていない場合、ジョブを実行する前に処理を打ち切りたい場合がある。

そのような場合は、前処理(beforeStepメソッド)にて例外をスローすることでジョブ実行前に処理を打ち切ることができる。
ここでは以下の要件を実装する場合を例に説明する。

StepExecutionListenerが定義しているbeforeStepメソッドで入力ファイルと出力ファイルの起動パラメータの妥当性検証を行う。
起動パラメータのいずれかが未指定の場合、例外をスローする。

しかし、TERASOLUNA Batch 5.xでは起動パラメータの妥当性検証は、JobParametersValidatorの使用を推奨している。ここでは、あくまでも前処理中断のサンプルとしてわかりやすい妥当性検証を利用しているため、実際に起動パラメータの妥当性検証を行う場合は"パラメータの妥当性検証"を参照すること。

以下に実装例を示す。

起動パラメータの妥当性検証を行うStepExecutionListenerの実装例

@Component
@Scope("step")
public class CheckingJobParameterErrorStepExecutionListener implements StepExecutionListener {

    @Value("#{jobParameters['inputFile']}") // (1)
    private File inputFile;

    @Value("#{jobParameters['outputFile']}") // (1)
    private File outputFile;

    @Override
    public void beforeStep(StepExecution stepExecution) {
        if (inputFile == null) {
            throw new BeforeStepException("The input file must be not null."); // (2)
        }
        else if (outputFile == null) {
            throw new BeforeStepException("The output file must be not null."); // (2)
        }
    }

    @Override
    public ExitStatus afterStep(StepExecution stepExecution) {
        // omitted.
    }
}

リスナーの設定例

<bean id="reader" class="org.terasoluna.batch.functionaltest.ch04.listener.LoggingReader" scope="step"
      p:resource="file:#{jobParameters['inputFile']}"/> <!-- (3) -->
<bean id="writer" class="org.terasoluna.batch.functionaltest.ch04.listener.LoggingWriter" scope="step"
      p:resource="file:#{jobParameters['outputFile']}"/> <!-- (3) -->

<batch:job id="chunkJobWithAbortListener" job-repository="jobRepository">
    <batch:step id="chunkJobWithAbortListener.step01">
        <batch:tasklet transaction-manager="jobTransactionManager">
            <batch:chunk reader="reader" writer="writer" commit-interval="10"/>
        </batch:tasklet>
        <batch:listeners>
            <batch:listener ref="checkingJobParameterErrorStepExecutionListener"/> <!-- (4) -->
        </batch:listeners>
    </batch:step>
</batch:job>

説明
項番	説明
(1)	@Valueアノテーションを使用して参照するパラメータを指定する。
(2)	例外をスローする。この例では`RuntimeException`クラスを継承し、自作した例外クラスを使用している。
(3)	参照するパラメータを指定する。パラメータをセットしているクラスはそれぞれ`ItemStreamReader`、`ItemStreamWriter`を実装した独自クラスである。
(4)	リスナーインターフェース実装を設定する。