[퍼옴] [Spring Batch] 스프링 배치 강좌 3. JobExecutionContext를 활용하여 스텝간 데이터 공유하기

목적

이전 글에서 배운Metadata와JobRepository를 활용해본다.

여러 개의 스텝을 만들고 스텝들이 어떻게JobExecutionContext를 사용하여 데이터를 공유하는지 알아보자.

common-context.xml

Job들이 공통으로 사용하는 빈들을 모아 앞으로 재사용 하도록 하자.

src/main/resource폴더에common-context.xml파일을 생성하자.

<?xml version="1.0" encoding="UTF-8"?>

<beans xmlns="http://www.springframework.org/schema/beans"
    xmlns:batch="http://www.springframework.org/schema/batch"
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xmlns:context="http://www.springframework.org/schema/context"
    xmlns:tx="http://www.springframework.org/schema/tx"
    xmlns:jdbc="http://www.springframework.org/schema/jdbc"
    xsi:schemaLocation="http://www.springframework.org/schema/batch http://www.springframework.org/schema/batch/spring-batch.xsd
  http://www.springframework.org/schema/jdbc http://www.springframework.org/schema/jdbc/spring-jdbc-4.3.xsd
  http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
  http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd
  http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx-4.3.xsd">

    <bean id="jobRepository"
        class="org.springframework.batch.core.repository.support.JobRepositoryFactoryBean">
        <property name="dataSource" ref="dataSource" />
        <property name="transactionManager"
            ref="transactionManager" />
    </bean>

    <bean id="jobLauncher"
        class="org.springframework.batch.core.launch.support.SimpleJobLauncher">
        <property name="jobRepository" ref="jobRepository" />
    </bean>

    <bean name="dataSource"
        class="org.apache.commons.dbcp2.BasicDataSource"
        destroy-method="close">
        <property name="driverClassName" value="org.h2.Driver" />
        <property name="url"
            value="jdbc:h2:mem:test;MODE=Oracle" />
        <property name="username" value="sa" />
        <property name="password" value="" />
    </bean>

    <bean name="sessionFactory"
        class="org.springframework.orm.hibernate5.LocalSessionFactoryBean">
        <property name="dataSource" ref="dataSource" />
    </bean>

    <bean id="transactionManager"
        class="org.springframework.orm.hibernate5.HibernateTransactionManager"
        lazy-init="true">
        <property name="sessionFactory" ref="sessionFactory" />
    </bean>

    <!-- This tells Spring to activate annotation-driven transactions -->
    <tx:annotation-driven
        transaction-manager="transactionManager" />

    <jdbc:initialize-database
        data-source="dataSource">
        <jdbc:script
            location="org/springframework/batch/core/schema-drop-h2.sql" />
        <jdbc:script
            location="org/springframework/batch/core/schema-h2.sql" />
    </jdbc:initialize-database>

    <bean id="jobLauncherTestUtils"
        class="org.springframework.batch.test.JobLauncherTestUtils" />

    <bean id="jdbcTemplate"
        class="org.springframework.jdbc.core.JdbcTemplate">
        <property name="dataSource" ref="dataSource" />
    </bean>

</beans>

tasklet

아래tasklet을src/main/java에com.fwantastic.example2패키지에 생성하자. 자세한 설명은 주석을 참조하자.

CounterIncrementerTasklet.java

package com.fwantastic.example2;

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.item.ExecutionContext;
import org.springframework.batch.repeat.RepeatStatus;

/**
 * JobExecutionContext에서 MY.COUNTER 값을 꺼내오고, 콘솔에 출력 한 다음, 카운터 + 1의 값을 저장한다.
 */
public class CounterIncrementerTasklet implements Tasklet {

    private static final String MY_COUNTER_KEY = "MY.COUNTER";
    private static final int DEFAULT_VALUE = 0;

    private Integer counter;

    public RepeatStatus execute(final StepContribution contribution, final ChunkContext chunkContext) throws Exception {

        // JobExecutionContext를 access 하는 방법.
        final ExecutionContext jobExecutionContext = chunkContext.getStepContext().getStepExecution().getJobExecution()
                .getExecutionContext();

        // setter를 통해 값이 지정되지 않은 경우 jobExecutionContext에서 값을 불러온다.
        // jobExecutionContext에 MY.COUNTER라는 키/페어가 없을 경우 0을 기본 값으로 가진다.
        if (counter == null) {
            counter = jobExecutionContext.getInt(MY_COUNTER_KEY, DEFAULT_VALUE);
        }

        // 콘솔에 카운터 값 출력.
        System.out.println("카운터: " + counter);

        // 카운터 + 1의 값을 jobExecutionContext에 저장.
        jobExecutionContext.put(MY_COUNTER_KEY, counter + 1);

        return RepeatStatus.FINISHED;
    }

    public void setCounter(final Integer counter) {
        this.counter = counter;
    }

}

job

**src/main/resources폴더에com/fwantastic/example2폴더를 만들고 아래의 xml을 추가하자.

job2.xml**

<?xml version="1.0" encoding="UTF-8"?>

<beans xmlns="http://www.springframework.org/schema/beans"
    xmlns:batch="http://www.springframework.org/schema/batch"
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xmlns:context="http://www.springframework.org/schema/context"
    xmlns:tx="http://www.springframework.org/schema/tx"
    xmlns:jdbc="http://www.springframework.org/schema/jdbc"
    xsi:schemaLocation="http://www.springframework.org/schema/batch http://www.springframework.org/schema/batch/spring-batch.xsd
  http://www.springframework.org/schema/jdbc http://www.springframework.org/schema/jdbc/spring-jdbc-4.3.xsd
  http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
  http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd
  http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx-4.3.xsd">

    <import resource="classpath:/common-context.xml" />

    <job id="myJob2"
        xmlns="http://www.springframework.org/schema/batch">
        <description>
            <![CDATA[
            여러 개의 스텝 사용과 JobExecutionContext를 사용해 스텝 간 데이터를 공유하는 예제.

            myStep1:
                jobExecutionContext에서 MY.COUNT 값을 불러옴. 
                처음엔 값이 없기 때문에 기본값인 0을 콘솔에 출력하고 1을 jobExecutionContext에 저장.

            myStep2:
                jobExecutionContext에서 MY.COUNT 값을 불러옴. 
                1을 콘솔에 출력하고 2을 jobExecutionContext에 저장.

            myStep3: 
                SPEL을 사용 해서 jobExecutionContext에서 MY.COUNT 값을 불러와 NoOpTasklet의 setter를 통해 넘겨줌. 
                2를 콘솔에 출력하고 3을 jobExecutionContext에 저장.

            JobExecutionContext는 Map<String , Object>와 같다고 생각하면 된다. 
            공유하고자 하는 객체를 맵에 넣으면 Job 안에 있는 모든 컴포넌트들이 사용 할 수 있다.
            ]]>
        </description>

        <step id="myStep1" next="myStep2">
            <tasklet ref="counterIncrementerTasklet" />
        </step>

        <step id="myStep2" next="myStep3">
            <tasklet ref="counterIncrementerTasklet" />
        </step>

        <step id="myStep3">
            <tasklet ref="counterIncrementerTasklet2" />
        </step>

    </job>

    <bean id="counterIncrementerTasklet"
        class="com.fwantastic.example2.CounterIncrementerTasklet"
        scope="step" />

    <bean id="counterIncrementerTasklet2"
        class="com.fwantastic.example2.CounterIncrementerTasklet"
        scope="step">
        <property name="counter"
            value="#{jobExecutionContext['MY.COUNTER']}" />
    </bean>

</beans>

테스트

**src/test/java에com.fwantastic.example2**패키지 만들고 아래 junit 클래스를 추가하자.

MyJobTest.java****

package com.fwantastic.example2;

import java.util.concurrent.atomic.AtomicBoolean;

import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.batch.core.ExitStatus;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.test.JobLauncherTestUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;

@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations = { "/com/fwantastic/example2/job2.xml" })
public class MyJobTest {

    @Autowired
    private JobLauncherTestUtils jobLauncherTestUtils;

    // AtomicBoolean을 사용해 한 테스트 케이스에 Job이 여러 번 실행되는 걸 방지한다.
    private static AtomicBoolean isLaunched = new AtomicBoolean(false);

    private JobExecution jobExecution;

    @Before
    public void setUp() throws Exception {
        if (!isLaunched.getAndSet(true)) {
            jobExecution = jobLauncherTestUtils.launchJob();
        }
    }

    @Test
    public void testExitCode() {
        // Job이 에러 없이 종료되었는지 확인
        Assert.assertEquals(ExitStatus.COMPLETED.getExitCode(), jobExecution.getExitStatus().getExitCode());
    }
}

실행결과

INFO: Job: [FlowJob: [name=myJob2]] launched with the following parameters: [{random=629547}]
Dec 13, 2019 5:22:00 PM org.springframework.batch.core.job.SimpleStepHandler handleStep
INFO: Executing step: [myStep1]
카운터: 0
Dec 13, 2019 5:22:00 PM org.springframework.batch.core.step.AbstractStep execute
INFO: Step: [myStep1] executed in 14ms
Dec 13, 2019 5:22:00 PM org.springframework.batch.core.job.SimpleStepHandler handleStep
INFO: Executing step: [myStep2]
카운터: 1
Dec 13, 2019 5:22:00 PM org.springframework.batch.core.step.AbstractStep execute
INFO: Step: [myStep2] executed in 4ms
Dec 13, 2019 5:22:00 PM org.springframework.batch.core.job.SimpleStepHandler handleStep
INFO: Executing step: [myStep3]
카운터: 2
Dec 13, 2019 5:22:00 PM org.springframework.batch.core.step.AbstractStep execute
INFO: Step: [myStep3] executed in 34ms
Dec 13, 2019 5:22:00 PM org.springframework.batch.core.launch.support.SimpleJobLauncher$1 run
INFO: Job: [FlowJob: [name=myJob2]] completed with the following parameters: [{random=629547}] and the following status: [COMPLETED] in 112ms

각 스텝이JobExecutionContext를 사용해 데이터를 공유하는 방법을 알아보았다. 다음은CSV (Comma Separated Values)파일을 읽어서디비에 저장하는 방법을 알아보자.