Kotlin으로 Spring Batch 만들기


개요

Kotlin 언어를 이용하여 간단한 Spring Batch 프로젝트를 만들어 보겠다.

프로젝트 생성

아래와 같이 curl 명령어을 사용하여 Spring Boot 초기 프로젝트를 생성한다.

curl https://start.spring.io/starter.tgz \
-d bootVersion=2.5.5 \
-d dependencies=batch,h2 \
-d baseDir=kotlin-spring-batch \
-d groupId=com.devkuma \
-d artifactId=kotlin-spring-batch \
-d packageName=com.devkuma.batch \
-d applicationName=BatchApplication \
-d packaging=jar \
-d language=kotlin \
-d jvmVersion=11 \
-d type=gradle-project | tar -xzvf -

위 명령어를 실행하게 되면 Spring Batch, H2 Database를 추가하였다.

생성된 프로젝트의 파일 구조는 아래와 같이 구성된다.

.
├── HELP.md
├── build.gradle.kts
├── gradle
│   └── wrapper
│       ├── gradle-wrapper.jar
│       └── gradle-wrapper.properties
├── gradlew
├── gradlew.bat
├── settings.gradle.kts
└── src
    ├── main
    │   ├── kotlin
    │   │   └── com
    │   │       └── devkuma
    │   │           └── batch
    │   │               └── BatchApplication.kt
    │   └── resources
    │       └── application.properties
    └── test
        └── kotlin
            └── com
                └── devkuma
                    └── batch
                        └── BatchApplicationTests.kt

Spring Batch 설정

/src/main/kotlin/com/devkuma/elasticsearch/classElasticsearchApplication.kt

package com.devkuma.batch

import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing
import org.springframework.boot.autoconfigure.SpringBootApplication
import org.springframework.boot.runApplication

@EnableBatchProcessing
@SpringBootApplication
class BatchApplication

fun main(args: Array<String>) {
	runApplication<BatchApplication>(*args)
}

@EnableBatchProcessing 어노테이션을 추가하여 배치 기능을 활성화 한다. 이 어노테이션을 추가 하므로써 Spring Batch 기능들을 사용할 수 있게 된다.

로깅 라이브러리 추가

/build.gradle.kts

// ... 생략 ...

dependencies {
	implementation("org.springframework.boot:spring-boot-starter-batch")
	implementation("org.jetbrains.kotlin:kotlin-reflect")
	implementation("org.jetbrains.kotlin:kotlin-stdlib-jdk8")
	runtimeOnly("com.h2database:h2")
	testImplementation("org.springframework.boot:spring-boot-starter-test")
	testImplementation("org.springframework.batch:spring-batch-test")

	// https://mvnrepository.com/artifact/io.github.microutils/kotlin-logging
	implementation("io.github.microutils:kotlin-logging:2.0.11")
}

// ... 생략 ...

파일 내용 중에 의존성 라이브러리를 확인해 보면 Spring 배치(spring-boot-starter-batch), 코틀린용 H2 Database(com.h2database:h2)가 추가되어 있는 것을 볼수 있다.

그리고, 신규로 코틀린 로깅(kotlin-logging)을 추가 하였다.

Tasklet 처리 방식

단일 스탭 구성 구현

단일 스탭 일감의 설정을 추가한다.

/src/main/kotlin/com/devkuma/batch/config/SingleStepJobConfig.kt

package com.devkuma.batch.config

import mu.KotlinLogging
import org.springframework.batch.core.Job
import org.springframework.batch.core.Step
import org.springframework.batch.core.StepContribution
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory
import org.springframework.batch.core.scope.context.ChunkContext
import org.springframework.batch.repeat.RepeatStatus
import org.springframework.context.annotation.Bean
import org.springframework.context.annotation.Configuration

private val log = KotlinLogging.logger {}

@Configuration
class SingleStepJobConfig(
    private val jobBuilderFactory: JobBuilderFactory,
    private val stepBuilderFactory: StepBuilderFactory
) {

    @Bean
    fun singleStepJob(): Job {
        return jobBuilderFactory["singleStepJob"]
            .start(singleStep())
            .build()
    }

    @Bean
    fun singleStep(): Step {
        return stepBuilderFactory["singleStep"]
            .tasklet { _: StepContribution, _: ChunkContext ->
                log.info { "Single Step!!" }
                RepeatStatus.FINISHED
            }
            .build()
    }
}
  • jobBuilderFactory["singleStepJob"]
    • simpleJob이란 이름으로 Batch Job을 생성한다.
    • Job 이름은 별도로 지정하지 않고, 이렇게 Builder를 통해 지정한다.
  • stepBuilderFactory["singleStep"]
    • simpleStep1이란 이름으로 Batch Step을 생성한다.
    • 위에 Job 이름과 동일하게 Builder를 통해 이름을 지정한다.
  • .tasklet { _: StepContribution, _: ChunkContext
    • Step에서 수행될 기능들을 명시한다.
    • Tasklet은 Step에서 단일로 수행될 커스텀한 기능들을 선언할때 사용한다.
    • 여기서는 Batch가 수행되면 log.info { "Single Step!!" } 구문으로 의해 로그가 출력된다.

실행 결과는 아래와 같다.

// ... 생략 ...

2021-10-02 01:51:51.746  INFO 72551 --- [           main] o.s.b.c.l.support.SimpleJobLauncher      : Job: [SimpleJob: [name=singleStepJob]] launched with the following parameters: [{}]
2021-10-02 01:51:51.769  INFO 72551 --- [           main] o.s.batch.core.job.SimpleStepHandler     : Executing step: [singleStep]
2021-10-02 01:51:51.778  INFO 72551 --- [           main] c.d.batch.config.SingleStepJobConfig     : Single Step!!
2021-10-02 01:51:51.782  INFO 72551 --- [           main] o.s.batch.core.step.AbstractStep         : Step: [singleStep] executed in 13ms
2021-10-02 01:51:51.786  INFO 72551 --- [           main] o.s.b.c.l.support.SimpleJobLauncher      : Job: [SimpleJob: [name=singleStepJob]] completed with the following parameters: [{}] and the following status: [COMPLETED] in 28ms

// ... 생략 ...

위에 로그를 확인해 보면 “Single Step!!“이 출력된 것을 확인 할 수 있을 것이다.

다중 Step 구성 구현

다중 스탭 일감의 설정을 추가한다.

/src/main/kotlin/com/devkuma/batch/config/SingleStepJobConfig.kt

package com.devkuma.batch.config

import mu.KotlinLogging
import org.springframework.batch.core.Job
import org.springframework.batch.core.Step
import org.springframework.batch.core.StepContribution
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory
import org.springframework.batch.core.scope.context.ChunkContext
import org.springframework.batch.repeat.RepeatStatus
import org.springframework.context.annotation.Bean
import org.springframework.context.annotation.Configuration

private val log = KotlinLogging.logger {}

@Configuration
class MultipleStepJobConfig(
    private val jobBuilderFactory: JobBuilderFactory,
    private val stepBuilderFactory: StepBuilderFactory
) {

    @Bean
    fun multipleStepJob(): Job {
        return jobBuilderFactory["multipleStepJob"]
            .start(startStep())
            .next(nextStep())
            .next(lastStep())
            .build()
    }

    @Bean
    fun startStep(): Step {
        return stepBuilderFactory["startStep"]
            .tasklet { _: StepContribution, _: ChunkContext ->
                log.info { "Start Step!!" }
                RepeatStatus.FINISHED
            }
            .build()
    }

    @Bean
    fun nextStep(): Step {
        return stepBuilderFactory["nextStep"]
            .tasklet { _: StepContribution, _: ChunkContext ->
                log.info { "Next Step!!" }
                RepeatStatus.FINISHED
            }
            .build()
    }

    @Bean
    fun lastStep(): Step {
        return stepBuilderFactory["lastStep"]
            .tasklet { _: StepContribution, _: ChunkContext ->
                log.info { "Last Step!!" }
                RepeatStatus.FINISHED
            }
            .build()
    }
}
  • jobBuilderFactory["multipleStepJob"]
    • multipleStepJob이란 이름으로 Batch Job을 생성한다.
  • stepBuilderFactory["startStep"], stepBuilderFactory["nextStep"], stepBuilderFactory["lastStep"]
    • 각각 startStep, nextStep, lastStep 이름으로 Batch Step을 각각 생성한다.

실행 결과는 아래와 같다.

// ... 생략 ...

2021-10-02 01:54:56.544  INFO 72707 --- [           main] o.s.b.c.l.support.SimpleJobLauncher      : Job: [SimpleJob: [name=multipleStepJob]] launched with the following parameters: [{}]
2021-10-02 01:54:56.573  INFO 72707 --- [           main] o.s.batch.core.job.SimpleStepHandler     : Executing step: [startStep]
2021-10-02 01:54:56.585  INFO 72707 --- [           main] c.d.batch.config.MultipleStepJobConfig   : Start Step!!
2021-10-02 01:54:56.593  INFO 72707 --- [           main] o.s.batch.core.step.AbstractStep         : Step: [startStep] executed in 20ms
2021-10-02 01:54:56.603  INFO 72707 --- [           main] o.s.batch.core.job.SimpleStepHandler     : Executing step: [nextStep]
2021-10-02 01:54:56.607  INFO 72707 --- [           main] c.d.batch.config.MultipleStepJobConfig   : Next Step!!
2021-10-02 01:54:56.613  INFO 72707 --- [           main] o.s.batch.core.step.AbstractStep         : Step: [nextStep] executed in 10ms
2021-10-02 01:54:56.619  INFO 72707 --- [           main] o.s.batch.core.job.SimpleStepHandler     : Executing step: [lastStep]
2021-10-02 01:54:56.622  INFO 72707 --- [           main] c.d.batch.config.MultipleStepJobConfig   : Last Step!!
2021-10-02 01:54:56.627  INFO 72707 --- [           main] o.s.batch.core.step.AbstractStep         : Step: [lastStep] executed in 8ms

// ... 생략 ...

위에 로그를 확인해 보면 스탭별로 “Start Step!!”, “Next Step!!”, “Last Step!!“이 출력된 것을 확인 할 수 있을 것이다.

Flow를 통한 Step 구성 구현

package com.devkuma.batch.config

import mu.KotlinLogging
import org.springframework.batch.core.ExitStatus
import org.springframework.batch.core.Job
import org.springframework.batch.core.Step
import org.springframework.batch.core.StepContribution
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory
import org.springframework.batch.core.scope.context.ChunkContext
import org.springframework.batch.repeat.RepeatStatus
import org.springframework.context.annotation.Bean
import org.springframework.context.annotation.Configuration

private val log = KotlinLogging.logger {}

@Configuration
class FlowStepJobConfig(
    private val jobBuilderFactory: JobBuilderFactory,
    private val stepBuilderFactory: StepBuilderFactory
) {

    @Bean
    fun flowStepJob(): Job {
        return jobBuilderFactory["flowStepJob"]

            .start(flowStartStep())
            .on(ExitStatus.COMPLETED.exitCode) // "flowStartStep"의 "ExitStatus"가 "COMPLETED"인 경우
            .to(flowProcessStep()) // "flowProcessStep"을 실행 시킨다
            .on("*") // "flowProcessStep"의 결과와 상관없이
            .to(flowWriteStep()) // "writeStep"을 실행 시킨다.
            .on("*") // "flowWriteStep"의 결과와 상관없이
            .end() // "Flow"를 종료 시킨다.

            .from(flowStartStep())
            .on(ExitStatus.FAILED.exitCode) // "flowStartStep"의 "ExitStatus"가 "FAILED"일 경우
            .to(flowFailOverStep()) // "flowFailOverStep"을 실행 시킨다.
            .on("*") // "flowFailOverStep"의 결과와 상관없이
            .to(flowWriteStep()) //  "flowWriteStep"을 실행 시킨다.
            .on("*") //  // "flowWriteStep"의 결과와 상관없이
            .end() // "Flow"를 종료시킨다.

            .from(flowStartStep())
            .on("*") // "flowStartStep"의 "ExitStatus"가 "FAILED", "COMPLETED"가 아닌 모든 경우
            .to(flowWriteStep()) // "flowWriteStep"을 실행시킨다.
            .on("*") // "flowWriteStep"의 결과와 상관없이
            .end() // "Flow"를 종료시킨다.

            .end()
            .build()
    }

    @Bean
    fun flowStartStep(): Step {
        return stepBuilderFactory["flowStartStep"]
            .tasklet { contribution: StepContribution, _: ChunkContext ->
                log.info("Flow Start Step!")

                val result = "COMPLETED"
                // val result = "FAIL";
                // val result = "UNKNOWN";

                // "Flow"에서 "on"은 "RepeatStatus"가 아닌 "ExitStatus"를 바라본다.
                if (result == "COMPLETED") {
                    contribution.exitStatus = ExitStatus.COMPLETED
                } else if (result == "FAIL") {
                    contribution.exitStatus = ExitStatus.FAILED
                } else if (result == "UNKNOWN") {
                    contribution.exitStatus = ExitStatus.UNKNOWN
                }
                RepeatStatus.FINISHED
            }
            .build()
    }

    @Bean
    fun flowProcessStep(): Step {
        return stepBuilderFactory["flowProcessStep"]
            .tasklet { _: StepContribution?, _: ChunkContext? ->
                log.info("Flow Process Step!")
                RepeatStatus.FINISHED
            }
            .build()
    }

    @Bean
    fun flowFailOverStep(): Step {
        return stepBuilderFactory.get("flowFailOverStep")
            .tasklet { _: StepContribution, _: ChunkContext ->
                log.info { "Flow FailOver Step!!" }
                RepeatStatus.FINISHED
            }
            .build()
    }

    @Bean
    fun flowWriteStep(): Step {
        return stepBuilderFactory["flowWriteStep"]
            .tasklet { _: StepContribution?, _: ChunkContext? ->
                log.info("Flow Write Step!")
                RepeatStatus.FINISHED
            }
            .build()
    }
}
  • jobBuilderFactory["flowStepJob"]
    • 주석에 내용을 따라 flowStartStepExitStatus 값에 따라 동작을 다르게 하게 된다.
      • ExitStatusCOMPLETED인 경우는 flowProcessStep, flowWriteStep를 실행 시킨다.
      • ExitStatusFAILED인 경우는 flowFailOverStep, flowWriteStep를 실행 시킨다.
      • ExitStatusFAILED, “COMPLETED"가 둘다 아닌 경우는 flowWriteStep를 실행 시킨다.

실행 결과는 아래와 같다.

// ... 생략 ...

:19:50.268  INFO 73719 --- [           main] o.s.b.c.l.support.SimpleJobLauncher      : Job: [FlowJob: [name=flowStepJob]] launched with the following parameters: [{}]
2021-10-02 02:19:50.295  INFO 73719 --- [           main] o.s.batch.core.job.SimpleStepHandler     : Executing step: [flowStartStep]
2021-10-02 02:19:50.303  INFO 73719 --- [           main] c.d.batch.config.FlowStepJobConfig       : Flow Start Step!
2021-10-02 02:19:50.309  INFO 73719 --- [           main] o.s.batch.core.step.AbstractStep         : Step: [flowStartStep] executed in 14ms
2021-10-02 02:19:50.314  INFO 73719 --- [           main] o.s.batch.core.job.SimpleStepHandler     : Executing step: [flowProcessStep]
2021-10-02 02:19:50.315  INFO 73719 --- [           main] c.d.batch.config.FlowStepJobConfig       : Flow Process Step!
2021-10-02 02:19:50.317  INFO 73719 --- [           main] o.s.batch.core.step.AbstractStep         : Step: [flowProcessStep] executed in 3ms
2021-10-02 02:19:50.320  INFO 73719 --- [           main] o.s.batch.core.job.SimpleStepHandler     : Executing step: [flowWriteStep]
2021-10-02 02:19:50.322  INFO 73719 --- [           main] c.d.batch.config.FlowStepJobConfig       : Flow Write Step!
2021-10-02 02:19:50.324  INFO 73719 --- [           main] o.s.batch.core.step.AbstractStep         : Step: [flowWriteStep] executed in 3ms
2021-10-02 02:19:50.327  INFO 73719 --- [           main] o.s.b.c.l.support.SimpleJobLauncher      : Job: [FlowJob: [name=flowStepJob]] completed with the following parameters: [{}] and the following status: [COMPLETED] in 47ms
2021-10-02 02:19:50.330  INFO 73719 --- [           main] o.s.b.c.l.support.SimpleJobLauncher      : Job: [SimpleJob: [name=multipleStepJob]] launched with the following parameters: [{}]

// ... 생략 ...

위에 로그를 확인해 보면 flowStartStepcontribution.exitStatuscontribution.exitStatus = ExitStatus.COMPLETED 이어서 flowProcessStepFlow Process Step!가 출력되는 것을 확인 할 수 있다.

flowStartStepcontribution.exitStatus 값을 변경해 보면서 로그는 확인해 보도록 하자.

참고