대용량 처리를 위한 멀티스레드 기반의 병행처리 예제

<job id="parallelJob" xmlns="http://www.springframework.org/schema/batch">
	<step id="staging" next="loading">
		<tasklet>
			<chunk reader="fileItemReader" processor="validatingProcessor" writer="stagingItemWriter" commit-interval="2"/>
		</tasklet>
	</step>
	<step id="loading">
		<tasklet task-executor="taskExecutor">
			<chunk reader="stagingReader" processor="stagingProcessor" writer="tradeWriter" commit-interval="3"/>
		</tasklet>
	</step>
</job>
 
<bean id="taskExecutor" class="org.springframework.core.task.SimpleAsyncTaskExecutor" />

@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations = { "/egovframework/batch/simple-job-launcher-context.xml", "/egovframework/batch/jobs/parallelJob.xml",
		"/egovframework/batch/job-runner-context.xml"  })
public class EgovParallelJobFunctionalTests {
 
	//배치작업을  test하기 위한 JobLauncherTestUtils
	@Autowired
	private JobLauncherTestUtils jobLauncherTestUtils;
 
	//DB 사용을 위한  SimpleJdbcTemplate
	private SimpleJdbcTemplate jdbcTemplate;
 
	...
 
	@Test
	public void testLaunchJob() throws Exception {
		int before = SimpleJdbcTestUtils.countRowsInTable(jdbcTemplate, "BATCH_STAGING");
		JobExecution execution = jobLauncherTestUtils.launchJob();
		int after = SimpleJdbcTestUtils.countRowsInTable(jdbcTemplate, "BATCH_STAGING");
		assertEquals(BatchStatus.COMPLETED, execution.getStatus());
		assertEquals(after - before, execution.getStepExecutions().iterator().next().getReadCount());
	}
 
}

<job id="parallelStep" xmlns="http://www.springframework.org/schema/batch">
	<split id="split1" task-executor="taskExecutor" next="step4">
		<flow>
			<step id="step1" next="step2">
				<tasklet>
					<chunk reader="itemReader" writer="itemWriter1" commit-interval="1" />
				</tasklet>
			</step>
			<step id="step2">
				<tasklet>
					<chunk reader="itemReader" writer="itemWriter2" commit-interval="2" />
				</tasklet>
			</step>
		</flow>
		<flow>
			<step id="step3">
				<tasklet>
					<chunk reader="itemReader" writer="itemWriter3" commit-interval="2" />
				</tasklet>
			</step>
		</flow>
	</split>
        <step id="step4">
		<tasklet>
			<chunk reader="itemReader" writer="itemWriter4" commit-interval="2" />
		</tasklet>
	</step>
</job>
 
<bean id="taskExecutor" class="org.springframework.core.task.SimpleAsyncTaskExecutor" />

@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations ={"/egovframework/batch/simple-job-launcher-context.xml", "/egovframework/batch/jobs/parallelStep.xml","/egovframework/batch/job-runner-context.xml"})
public class EgovParallelStepFunctionalTests{
 
	//배치작업을  test하기 위한 JobLauncherTestUtils
	@Autowired
	private JobLauncherTestUtils jobLauncherTestUtils;
 
	/**
	 * 배치작업 테스트
	 */
	@Test
	public void testLaunchJob() throws Exception {
		JobExecution jobExecution = jobLauncherTestUtils.launchJob(this.getUniqueJobParameters());
		assertEquals(BatchStatus.COMPLETED, jobExecution.getStatus());
		// /target/test-outputs/parallelStep폴더의 output파일 확인
	}
 
	/**
     * 잡파라미터를 설정하기 위한 메소드 
     * @return jobParameters
     */
	protected JobParameters getUniqueJobParameters() {
		return new JobParametersBuilder().addString("inputFile","/egovframework/data/input/delimited.csv")
		.addString("outputFile1","file:./target/test-outputs/parallelStep/delimitedOutput1.csv")
		.addString("outputFile2","file:./target/test-outputs/parallelStep/delimitedOutput2.csv")
		.addString("outputFile3","file:./target/test-outputs/parallelStep/delimitedOutput3.csv")
		.addString("outputFile4","file:./target/test-outputs/parallelStep/delimitedOutput4.csv")
		.addParameter("timestamp", new JobParameter(new Date().getTime())).toJobParameters();
	}
}