Azure Functions上でSpring Batchを利用しているプログラムで処理単位(Tasklet)を複数つないでみた

これまでこのブログでは、1つの処理単位(Tasklet)のみを扱ってきたが、複数の処理単位を順次実行することもできる。

今回は、EventGridTriggerを利用して、BlobにCSVファイルが作成されたタイミングで、そのCSVファイルをDBに書き込む処理を実行した後で、ログ出力する処理単位(Tasklet)を実行してみたので、そのサンプルプログラムを共有する。

前提条件

下記記事のサンプルプログラムを作成済で、EventGridTriggerTestHandler.javaは、EventGridのCSVをDBに書き込む処理を残した状態になっていること。

EventGridTriggerを利用したプログラムをローカル環境で動かしてみた前回は、EventGridTriggerを利用して、BlobにCSVファイルが作成されたタイミングで、そのCSVファイルをDBに書き込む...

また、以下の記事のPostmanをインストール済であること。

POSTメソッドでリクエストされるAzure FunctionsのAPIをPostmanによって呼び出してみたAzure App Serviceを利用せずに、HTTPトリガーによって呼び出されるAzure Functionsの動作確認を行うには、...

作成したサンプルプログラムの修正

前提条件の記事のサンプルプログラムに、ログ出力する処理単位(Tasklet)を追加する。なお、下記の赤枠は、前提条件のプログラムから変更・追加したプログラムである。

新しく追加したログ出力する処理単位(Tasklet)を定義したクラスは、以下の通り。

package com.example.batch;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.batch.core.StepContribution;
import org.springframework.batch.core.scope.context.ChunkContext;
import org.springframework.batch.core.step.tasklet.Tasklet;
import org.springframework.batch.repeat.RepeatStatus;
import org.springframework.stereotype.Component;

import com.example.model.EventGridTriggerParam;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;

@Component
public class DemoTaskletLog implements Tasklet {

    /* Spring Bootでログ出力するためのLogbackのクラスを生成 */
    private static final Logger LOGGER 
        = LoggerFactory.getLogger(DemoTaskletLog.class);

    /**
     * Spring Batchのジョブ内でのログ出力処理を定義する.
     */
    @Override
    public RepeatStatus execute(StepContribution contribution
             , ChunkContext chunkContext) throws Exception {
        // Spring Batchのジョブ内での処理が呼び出されたことをログ出力する
        String paramStr = chunkContext.getStepContext().getStepExecution()
                .getJobParameters().getString("eventGridTriggerParam");

        EventGridTriggerParam param = null;
        if (paramStr != null) {
            param = new ObjectMapper().readValue(paramStr
                , new TypeReference<EventGridTriggerParam>() {});
            LOGGER.info("DemoTaskletLog execute " 
                + " triggered: " + param.getTimerInfo());
        }

        // ジョブが終了したことを返す
        return RepeatStatus.FINISHED;
    }

}

package com.example.batch;

import org.slf4j.Logger;

import org.slf4j.LoggerFactory;

import org.springframework.batch.core.StepContribution;

import org.springframework.batch.core.scope.context.ChunkContext;

import org.springframework.batch.core.step.tasklet.Tasklet;

import org.springframework.batch.repeat.RepeatStatus;

import org.springframework.stereotype.Component;

import com.example.model.EventGridTriggerParam;

import com.fasterxml.jackson.core.type.TypeReference;

import com.fasterxml.jackson.databind.ObjectMapper;

@Component

public class DemoTaskletLog implements Tasklet {

/* Spring Bootでログ出力するためのLogbackのクラスを生成 */

private static final Logger LOGGER

= LoggerFactory.getLogger(DemoTaskletLog.class);

/**

* Spring Batchのジョブ内でのログ出力処理を定義する.

@Override

public RepeatStatus execute(StepContribution contribution

, ChunkContext chunkContext) throws Exception {

// Spring Batchのジョブ内での処理が呼び出されたことをログ出力する

String paramStr = chunkContext.getStepContext().getStepExecution()

.getJobParameters().getString("eventGridTriggerParam");

EventGridTriggerParam param = null;

if (paramStr != null) {

param = new ObjectMapper().readValue(paramStr

, new TypeReference<EventGridTriggerParam>() {});

LOGGER.info("DemoTaskletLog execute "

+ " triggered: " + param.getTimerInfo());

}

// ジョブが終了したことを返す

return RepeatStatus.FINISHED;

}

また、新しく追加したログ出力する処理単位(Tasklet)を追加したジョブ定義クラスは以下の通りで、ジョブを定義するjobメソッドで、CSVファイルをDBに書き込む処理単位(step)を実行した後で、ログ出力する処理単位(stepLog)を呼び出すようにしている。

package com.example.batch;

import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.core.launch.support.RunIdIncrementer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
@EnableBatchProcessing
public class DemoTaskletConfig {

    /** Spring Batchのジョブ内で指定する実行処理を定義するクラス */
    @Autowired
    private DemoTasklet demoTasklet;
    
    /** Spring Batchのジョブ内で指定するログ出力処理を定義するクラス */
    @Autowired
    private DemoTaskletLog demoTaskletLog;

    /** Spring Batchのジョブを生成するクラス */
    @Autowired
    private JobBuilderFactory jobBuilderFactory;

    /** Spring Batchのステップを生成するクラス */
    @Autowired
    private StepBuilderFactory stepBuilderFactory;

    /**
     * Spring Batchのジョブ内で指定するCSVファイル取込処理(ステップ)を定義する.
     * @return Spring Batchのジョブ内で指定するCSVファイル取込処理
     */
    @Bean
    public Step step() {
        // 生成するステップ内でCSVファイル取込処理(Tasklet)を指定する
        return stepBuilderFactory
                .get("step")
                .tasklet(demoTasklet)
                .build();
    }
    
    /**
     * Spring Batchのジョブ内で指定するログ出力処理(ステップ)を定義する.
     * @return Spring Batchのジョブ内で指定するログ出力処理
     */
    @Bean
    public Step stepLog() {
        // 生成するステップ内でログ出力する処理(Tasklet)を指定する
        return stepBuilderFactory
                .get("stepLog")
                .tasklet(demoTaskletLog)
                .build();
    }

    /**
     * Spring Batchのジョブを定義する.
     * @param step CSVファイル取込処理
     * @param stepLog ログ出力処理
     * @return Spring Batchのジョブ
     */
    @Bean
    public Job job(Step step, Step stepLog) {
        // 生成するジョブ内で、実行前後の処理(リスナ)と処理単位(ステップ)を指定する
        return jobBuilderFactory
                .get("job")
                .incrementer(new RunIdIncrementer())
                .listener(listener())
                .start(step)
                .next(stepLog)
                .build();
    }

    /**
     * Spring Batchのジョブの実行前後の処理を定義する.
     * @return Spring Batchのジョブの実行前後の処理
     */
    @Bean
    public DemoJobListener listener() {
        return new DemoJobListener();
    }
}

package com.example.batch;

import org.springframework.batch.core.Job;

import org.springframework.batch.core.Step;

import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;

import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;

import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;

import org.springframework.batch.core.launch.support.RunIdIncrementer;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.context.annotation.Bean;

import org.springframework.context.annotation.Configuration;

@Configuration

@EnableBatchProcessing

public class DemoTaskletConfig {

/** Spring Batchのジョブ内で指定する実行処理を定義するクラス */

@Autowired

private DemoTasklet demoTasklet;

/** Spring Batchのジョブ内で指定するログ出力処理を定義するクラス */

@Autowired

private DemoTaskletLog demoTaskletLog;

/** Spring Batchのジョブを生成するクラス */

@Autowired

private JobBuilderFactory jobBuilderFactory;

/** Spring Batchのステップを生成するクラス */

@Autowired

private StepBuilderFactory stepBuilderFactory;

/**

* Spring Batchのジョブ内で指定するCSVファイル取込処理(ステップ)を定義する.

* @return Spring Batchのジョブ内で指定するCSVファイル取込処理

@Bean

public Step step() {

// 生成するステップ内でCSVファイル取込処理(Tasklet)を指定する

return stepBuilderFactory

.get("step")

.tasklet(demoTasklet)

.build();

}

/**

* Spring Batchのジョブ内で指定するログ出力処理(ステップ)を定義する.

* @return Spring Batchのジョブ内で指定するログ出力処理

@Bean

public Step stepLog() {

// 生成するステップ内でログ出力する処理(Tasklet)を指定する

return stepBuilderFactory

.get("stepLog")

.tasklet(demoTaskletLog)

.build();

}

/**

* Spring Batchのジョブを定義する.

* @param step CSVファイル取込処理

* @param stepLog ログ出力処理

* @return Spring Batchのジョブ

@Bean

public Job job(Step step, Step stepLog) {

// 生成するジョブ内で、実行前後の処理(リスナ)と処理単位(ステップ)を指定する

return jobBuilderFactory

.get("job")

.incrementer(new RunIdIncrementer())

.listener(listener())

.start(step)

.next(stepLog)

.build();

}

/**

* Spring Batchのジョブの実行前後の処理を定義する.

* @return Spring Batchのジョブの実行前後の処理

@Bean

public DemoJobListener listener() {

return new DemoJobListener();

}

その他のソースコード内容は、以下のサイトを参照のこと。
https://github.com/purin-it/azure/tree/master/event-grid-trigger-multi-tasklet/demoAzureFunc

サンプルプログラムの実行結果

以下の記事の「EventGridのローカル環境での実行」に記載した通りに実行する。

実行した結果のログは以下の通りで、CSVファイルをDBに書き込むDemoTaskletクラスの処理を実行した後で、ログ出力するDemoTaskletLogクラスの処理が実行されたことが確認できる。

要点まとめ

Spring Batchを利用しているプログラムで処理単位(Tasklet)を複数つなぐ処理は、Spring Batchのジョブ定義クラスで定義することができる。

前提条件

作成したサンプルプログラムの修正

サンプルプログラムの実行結果

要点まとめ

Azure Functions上でSpring Batchを利用しているプログラムをEventGridTriggerによって動かしてみた(ソースコード編)

Azure Functions上でSpring Batchを利用しているプログラムをEventGridTriggerによって動かしてみた(Azure設定編)

EventGridTriggerを利用したプログラムをローカル環境で動かしてみた

Spring Batchを利用しているプログラムの処理単位(Tasklet)を@StepScopeアノテーションで並行に実行してみた

Azure Functions上でSpring Batchを利用しているプログラムで処理単位(Tasklet)を並列に実行してみた

Azure Functions上でSpring Batchを利用しているプログラムの処理単位(Tasklet)間でデータ受け渡しをしてみた