TimerTriggerによって動作するAzure Function上でSpring Batch(Taskletモデル)を利用してみた

Timer Triggerによって、一定時間が来たタイミングでAzure Functionsが動作するアプリケーションを生成できるが、そのバッチ処理内部で、Spring Batchを利用することもできる。

今回は、Azure Functions上でTimerTriggerによって動作するJavaアプリケーション(Spring Boot上)でSpring Batchを利用してみたので、そのサンプルプログラムを共有する。

なお、Spring BatchにはChunkモデルとTaskletモデルがあるが、今回はTaskletモデルを利用している。Spring Batchのアーキテクチャについては、以下のサイトを参照のこと。
https://terasoluna-batch.github.io/guideline/5.0.0.RELEASE/ja/Ch02_SpringBatchArchitecture.html

前提条件

下記記事のサンプルプログラムを作成済であること。

Azure FunctionsでJavaアプリケーション(Spring Boot上)のLogbackでのログを確認してみたこれまで、Azure Functionsの関数の「モニター」で何度かログを確認したことがあったが、この方法だと、Spring Bootプ...

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

前提条件の記事のサンプルプログラムに、Spring Batchの処理を追加する。なお、下記の赤枠は、前提条件のプログラムから大きく変更したり、追加したりしたプログラムである。

pom.xmlへの追加内容は以下の通りで、Spring Batchを利用するためのライブラリを追加している。

<!-- Spring Batchを利用するための設定 -->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-batch</artifactId>
</dependency>

<groupId>org.springframework.boot</groupId>

<artifactId>spring-boot-starter-batch</artifactId>

</dependency>

また、メインクラスの内容は以下の通りで、Spring Batch内で行われるデータソースの自動設定を除外している。

package com.example;

import java.util.function.Function;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.autoconfigure.jdbc.DataSourceAutoConfiguration;
import org.springframework.context.annotation.Bean;

import com.example.model.TimerTriggerParam;
import com.example.model.TimerTriggerResult;
import com.example.service.TimerTriggerService;

//Spring Batchを利用するが、今回はDBを使わないため、データソースの自動設定を除外する
@SpringBootApplication(exclude = { DataSourceAutoConfiguration.class })
public class DemoAzureFunction {
    
    /** タイマートリガーのテストを行うサービスクラスのオブジェクト */
    @Autowired
    private TimerTriggerService timerTriggerService;

    public static void main(String[] args) throws Exception {
        SpringApplication.run(DemoAzureFunction.class, args);
    }
    
    /**
     * タイマートリガーのテストを行い結果を返却する関数
     * @return タイマートリガーのテストを行うサービスクラスの呼出結果
     */
    @Bean
    public Function<TimerTriggerParam, TimerTriggerResult> timerTriggerTest(){
        return timerTriggerParam 
            -> timerTriggerService.timerTriggerTest(timerTriggerParam);
    }
    
}

package com.example;

import java.util.function.Function;

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

import org.springframework.boot.SpringApplication;

import org.springframework.boot.autoconfigure.SpringBootApplication;

import org.springframework.boot.autoconfigure.jdbc.DataSourceAutoConfiguration;

import org.springframework.context.annotation.Bean;

import com.example.model.TimerTriggerParam;

import com.example.model.TimerTriggerResult;

import com.example.service.TimerTriggerService;

//Spring Batchを利用するが、今回はDBを使わないため、データソースの自動設定を除外する

@SpringBootApplication(exclude = { DataSourceAutoConfiguration.class })

public class DemoAzureFunction {

/** タイマートリガーのテストを行うサービスクラスのオブジェクト */

@Autowired

private TimerTriggerService timerTriggerService;

public static void main(String[] args) throws Exception {

SpringApplication.run(DemoAzureFunction.class, args);

}

/**

* タイマートリガーのテストを行い結果を返却する関数

* @return タイマートリガーのテストを行うサービスクラスの呼出結果

@Bean

public Function<TimerTriggerParam, TimerTriggerResult> timerTriggerTest(){

return timerTriggerParam

-> timerTriggerService.timerTriggerTest(timerTriggerParam);

}

さらに、サービスクラスの内容は以下の通りで、Spring Batchのジョブを起動する設定を追加している。

package com.example.service;

import java.util.HashMap;
import java.util.Map;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.JobParameter;
import org.springframework.batch.core.JobParameters;
import org.springframework.batch.core.launch.JobLauncher;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import com.example.model.TimerTriggerParam;
import com.example.model.TimerTriggerResult;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;

@Service
public class TimerTriggerService {

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

    /** Spring Batchのジョブを起動するクラス */
    @Autowired
    JobLauncher jobLauncher;

    /** Spring Batchのジョブを定義するクラス */
    @Autowired
    Job job;

    /**
     * タイマートリガーのテストを行うサービス.
     * @param param TimerTrigger呼出用Param
     * @return タイマートリガーのテストを行うサービスクラスの呼出結果
     */
    public TimerTriggerResult timerTriggerTest(TimerTriggerParam param) {
        // タイマートリガーのテストを行うサービスが呼び出されたことをログ出力する
        LOGGER.info("TimerTriggerService timerTriggerTest triggered: " 
            + param.getTimerInfo());

        // Spring Batchのジョブを起動する
        // ジョブ起動時に生成したパラメータを指定する
        try {
            jobLauncher.run(job, createInitialJobParameterMap(param));
        } catch (Exception e) {
            throw new RuntimeException(e);
        }

        // タイマートリガーのテストを行うサービスクラスの呼出結果を返却する
        TimerTriggerResult result = new TimerTriggerResult();
        result.setResult("success");
        return result;
    }

    /**
     * ジョブを起動するためのパラメータを生成する.
     * @param TimerTriggerParamオブジェクト
     * @return ジョブを起動するためのパラメータ
     * @throws JsonProcessingException
     */
    private JobParameters createInitialJobParameterMap(TimerTriggerParam param)
            throws JsonProcessingException {
        Map<String, JobParameter> m = new HashMap<>();
        m.put("time", new JobParameter(System.currentTimeMillis()));
        m.put("timerTriggerParam"
           , new JobParameter(new ObjectMapper().writeValueAsString(param)));

        JobParameters p = new JobParameters(m);
        return p;
    }
}

package com.example.service;

import java.util.HashMap;

import java.util.Map;

import org.slf4j.Logger;

import org.slf4j.LoggerFactory;

import org.springframework.batch.core.Job;

import org.springframework.batch.core.JobParameter;

import org.springframework.batch.core.JobParameters;

import org.springframework.batch.core.launch.JobLauncher;

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

import org.springframework.stereotype.Service;

import com.example.model.TimerTriggerParam;

import com.example.model.TimerTriggerResult;

import com.fasterxml.jackson.core.JsonProcessingException;

import com.fasterxml.jackson.databind.ObjectMapper;

@Service

public class TimerTriggerService {

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

private static final Logger LOGGER

= LoggerFactory.getLogger(TimerTriggerService.class);

/** Spring Batchのジョブを起動するクラス */

@Autowired

JobLauncher jobLauncher;

/** Spring Batchのジョブを定義するクラス */

@Autowired

Job job;

/**

* タイマートリガーのテストを行うサービス.

* @param param TimerTrigger呼出用Param

* @return タイマートリガーのテストを行うサービスクラスの呼出結果

public TimerTriggerResult timerTriggerTest(TimerTriggerParam param) {

// タイマートリガーのテストを行うサービスが呼び出されたことをログ出力する

LOGGER.info("TimerTriggerService timerTriggerTest triggered: "

+ param.getTimerInfo());

// Spring Batchのジョブを起動する

// ジョブ起動時に生成したパラメータを指定する

try {

jobLauncher.run(job, createInitialJobParameterMap(param));

} catch (Exception e) {

throw new RuntimeException(e);

}

// タイマートリガーのテストを行うサービスクラスの呼出結果を返却する

TimerTriggerResult result = new TimerTriggerResult();

result.setResult("success");

return result;

}

/**

* ジョブを起動するためのパラメータを生成する.

* @param TimerTriggerParamオブジェクト

* @return ジョブを起動するためのパラメータ

* @throws JsonProcessingException

private JobParameters createInitialJobParameterMap(TimerTriggerParam param)

throws JsonProcessingException {

Map<String, JobParameter> m = new HashMap<>();

m.put("time", new JobParameter(System.currentTimeMillis()));

m.put("timerTriggerParam"

, new JobParameter(new ObjectMapper().writeValueAsString(param)));

JobParameters p = new JobParameters(m);

return p;

}

また、Spring Batchの起動定義は以下の通りで、バッチジョブに処理(Tasklet)とその前後処理(Listener)を追加している。

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 JobBuilderFactory jobBuilderFactory;

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

    /**
     * Spring Batchのジョブ内で指定する処理単位(ステップ)を定義する.
     * @return Spring Batchのジョブ内で指定する処理単位
     */
    @Bean
    public Step step() {
        // 生成するステップ内で実行処理(Tasklet)を指定する
        return stepBuilderFactory.get("step")
                .tasklet(demoTasklet)
                .build();
    }
    
    /**
     * Spring Batchのジョブを定義する.
     * @param step1 実行するステップ
     * @return Spring Batchのジョブ
     */
    @Bean
    public Job job(Step step) {
        // 生成するジョブ内で、実行前後の処理(リスナ)と処理単位(ステップ)を指定する
        return jobBuilderFactory.get("job")
                .incrementer(new RunIdIncrementer())
                .listener(listener())
                .start(step)
                .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 JobBuilderFactory jobBuilderFactory;

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

@Autowired

private StepBuilderFactory stepBuilderFactory;

/**

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

* @return Spring Batchのジョブ内で指定する処理単位

@Bean

public Step step() {

// 生成するステップ内で実行処理(Tasklet)を指定する

return stepBuilderFactory.get("step")

.tasklet(demoTasklet)

.build();

}

/**

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

* @param step1 実行するステップ

* @return Spring Batchのジョブ

@Bean

public Job job(Step step) {

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

return jobBuilderFactory.get("job")

.incrementer(new RunIdIncrementer())

.listener(listener())

.start(step)

.build();

}

/**

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

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

@Bean

public DemoJobListener listener() {

return new DemoJobListener();

}

さらに、Spring Batchのジョブ内での処理(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.repeat.RepeatStatus;
import org.springframework.stereotype.Component;

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

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

@Component
public class DemoTasklet implements Tasklet {

    /* Spring Bootでログ出力するためのLogbackのクラスを生成 */
    private static final Logger LOGGER 
         = LoggerFactory.getLogger(DemoTasklet.class);
    
    /**
     * Spring Batchのジョブ内での処理を定義する.
     */
    @Override
    public RepeatStatus execute(StepContribution contribution
           , ChunkContext chunkContext) throws Exception {
        // Spring Batchのジョブ内での処理が呼び出されたことをログ出力する
        String paramStr = chunkContext.getStepContext()
                .getStepExecution().getJobParameters()
                .getString("timerTriggerParam");
        if(paramStr != null) {
            TimerTriggerParam param = new ObjectMapper().readValue(
                 paramStr, new TypeReference<TimerTriggerParam>() {});
            LOGGER.info("DemoTasklet 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.repeat.RepeatStatus;

import org.springframework.stereotype.Component;

import com.example.model.TimerTriggerParam;

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

import com.fasterxml.jackson.databind.ObjectMapper;

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

@Component

public class DemoTasklet implements Tasklet {

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

private static final Logger LOGGER

= LoggerFactory.getLogger(DemoTasklet.class);

/**

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

@Override

public RepeatStatus execute(StepContribution contribution

, ChunkContext chunkContext) throws Exception {

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

String paramStr = chunkContext.getStepContext()

.getStepExecution().getJobParameters()

.getString("timerTriggerParam");

if(paramStr != null) {

TimerTriggerParam param = new ObjectMapper().readValue(

paramStr, new TypeReference<TimerTriggerParam>() {});

LOGGER.info("DemoTasklet execute " + " triggered: "

+ param.getTimerInfo());

}

return RepeatStatus.FINISHED;

}

また、Spring Batchのジョブの前後で実施する処理(Listener)を定義を定義しているクラスの内容は、以下の通り。

package com.example.batch;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.listener.JobExecutionListenerSupport;

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

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

    /**
     * Spring Batchのジョブ実行前の処理を定義する.
     */
    @Override
    public void beforeJob(JobExecution jobExecution) {
        super.beforeJob(jobExecution);
        
        // Spring Batchのジョブ実行前の処理が呼び出されたことをログ出力する
        printLog(jobExecution, "beforeJob");
    }
    
    /**
     * Spring Batchのジョブ実行後の処理を定義する.
     */
    @Override
    public void afterJob(JobExecution jobExecution) {
        super.afterJob(jobExecution);
        
        // Spring Batchのジョブ実行後の処理が呼び出されたことをログ出力する
        printLog(jobExecution, "afterJob");
    }
    
    /**
     * ログ出力を行う.
     * @param jobExecution ジョブ実行時の定義オブジェクト
     * @param methodName メソッド名
     */
    private void printLog(JobExecution jobExecution, String methodName) {
        try {
            String paramStr = jobExecution.getJobParameters()
                                  .getString("timerTriggerParam");            
            if(paramStr != null) {
                TimerTriggerParam param = new ObjectMapper().readValue(
                    paramStr, new TypeReference<TimerTriggerParam>() {});
                LOGGER.info("DemoJobListener " + methodName + " triggered: " 
                    + param.getTimerInfo());
            }
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
    
}

package com.example.batch;

import org.slf4j.Logger;

import org.slf4j.LoggerFactory;

import org.springframework.batch.core.JobExecution;

import org.springframework.batch.core.listener.JobExecutionListenerSupport;

import com.example.model.TimerTriggerParam;

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

import com.fasterxml.jackson.databind.ObjectMapper;

public class DemoJobListener extends JobExecutionListenerSupport {

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

private static final Logger LOGGER

= LoggerFactory.getLogger(DemoJobListener.class);

/**

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

@Override

public void beforeJob(JobExecution jobExecution) {

super.beforeJob(jobExecution);

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

printLog(jobExecution, "beforeJob");

}

/**

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

@Override

public void afterJob(JobExecution jobExecution) {

super.afterJob(jobExecution);

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

printLog(jobExecution, "afterJob");

}

/**

* ログ出力を行う.

* @param jobExecution ジョブ実行時の定義オブジェクト

* @param methodName メソッド名

private void printLog(JobExecution jobExecution, String methodName) {

try {

String paramStr = jobExecution.getJobParameters()

.getString("timerTriggerParam");

if(paramStr != null) {

TimerTriggerParam param = new ObjectMapper().readValue(

paramStr, new TypeReference<TimerTriggerParam>() {});

LOGGER.info("DemoJobListener " + methodName + " triggered: "

+ param.getTimerInfo());

}

} catch (Exception e) {

throw new RuntimeException(e);

}

その他のソースコード内容は、以下のサイトを参照のこと。ただし、TimerTriggerParamからLoggerのBean定義を削除している。
https://github.com/purin-it/azure/tree/master/azure-functions-spring-batch-tasklet/demoAzureFunc

また、その後のビルドとデプロイ手順については、前提条件の記事を参照のこと。