リスナー

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));
}

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)
        // do nothing.
    }

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

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

        // per step execution
        // (5)
        jobExecution.getStepExecutions().forEach(stepExecution -> {
            Object errorItem = stepExecution.getExecutionContext().get("ERROR_ITEM");
            if (errorItem != null) {
                logger.error("detected error on this item processing. " +
                        "[step:{}] [item:{}]", stepExecution.getStepName(),
                        errorItem);
            }
        });
    }
}

リスナーの設定例

<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"/> <!-- (6) -->
     </batch:listeners>
 </batch:job>

説明
項番	説明
(1)	`JobExecutionListener`を`implements`して実装する。
(2)	`JobExecutionListener`が定義している`beforeJob`メソッドを実装する。この例では、ジョブ開始前には何もしない。
(3)	`JobExecutionListener`が定義している`afteJob`メソッドを実装する。この例では、ジョブ終了時にジョブの最終状態と例外情報をログ出力する。
(4)	ジョブ名と終了コードをINFOログ出力する。必要な情報は、引数の`JobExecution`からそれぞれ取得する。
(5)	ステップごとに発生した例外をログ出力する。引数の`JobExecution`から紐づく`StepExecution`を取得し実現する。ここでは、例外の起因となった入力データを`ExecutionContext`から`ERROR_ITEM`というキーで取得している。 `ExecutionContext`に設定する例はジョブ単位の例外ハンドリングを参照。
(6)	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に属するリスナーインターフェース実装を設定する。
(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に対してのみ実装する。