배치작업 병렬 처리

배치의 병렬처리는 스케줄러에서 동일한 배치 Job을 서로 다른 파라미터 값으로 호출하는 Job 병렬처리와 하나의 배치 Job 내에서 여러 개의 Step을 동시에 수행하는 Step 병렬 처리하는 방법이 있다.

1. 배치 Job 병렬 처리

스케줄러에서 동일한 Job을 서로 다른 파라미터 값으로 동시에 호출 처리하는 방법이다.

프레임워크에서는 배치 Job ID와 실행 파라미터(JobParameter)의 조합으로 Job Instance를 생성하여 동일한 배치 Job이 여러 개 실행 되도록 한다.

Figure 1. 배치 Job 병렬 처리

2. 하나의 배치 병렬Job 내에서 Step 병렬 처리

일부 단위 Step들을 순서와 상관없이 동시에 처리되도록 Step 플로우를 정의하여 병렬로 처리할 수 있다.

('4장. Step 실행 플로우' 참조)

Figure 2. Step 병렬 처리

3. 데이터 Partition 처리

처리 데이터를 분할하여 동시에 여러 쓰레드가 분할된 데이터를 병렬로 처리하는 방법이다.

Figure 3. 데이터 Partition 처리

배치 작업 xml에서는 Master Step은 partition 요소로 정의 되어야 하며 Slave Step은 Job 흐름도 밖에 위치하여 한다.

handler 요소의 grid-size 속성 값으로 병렬처리 할 개수를 설정한다. 이 개수 값은 Partitioner 클래스의 partition 메소드의 파라미터로 전달된다.

    <job id="sample-partitioning" xmlns="http://www.springframework.org/schema/batch">
        <step id="sample-partitioning-001">
            <partition partitioner="SmpPartitioningPartitioner" step="sample-partitioning-002">
                <handler grid-size="3"/>
            </partition>
        </step>
    </job>

    <step id="sample-partitioning-002" xmlns="http://www.springframework.org/schema/batch">
        <tasklet>
                <chunk reader="SmpPartitioningDbToFixedChunk" processor="SmpPartitioningDbToFixedChunk" writer="w-sample-partitioning-002" />
            </tasklet>
    </step>

Partitioner 클래스 : 분할 대상 데이터 확인 및 분배 로직을 구현한다.

(ExecutionContext를 이용해서 다음 Step에 분할 처리할 데이터를 전달한다.)

@Component("MSmpPartitioning")
@Scope("step")
@BxmCategory(logicalName = "MSmpPartitioning")
public class MSmpPartitioning implements Partitioner{

    private final Logger logger = LoggerFactory.getLogger(this.getClass());

    @Override
    @BxmCategory(logicalName = "MSmpPartitioning")
    public Map<String, ExecutionContext> partition(int gridSize)
    {
        /**
         * Partition으로 분배한 Step의 ExecutionContext 정보를 전달하기 위한 Map
         * - Map의 Size 만큼 병렬로 Step이 수행이 된다. (보통 전달받은 gridSize 값이 Map의 size로 정의된다.)
         */
        Map<String, ExecutionContext> executionContextMap = new HashMap<String, ExecutionContext>();

        /**
         * gridSize 만큼 ExecutionContext Map에 설정한다.
         */
        for(int ix = 1; ix <= gridSize; ix++)
        {
            /**
             * ExecutionContext 값 설정
             * - Partitioning으로 수행할 값을 설정처리하면 된다.
             */
            ExecutionContext executionContext = new ExecutionContext();
            executionContext.put("deptNo", ix * 10);

            /**
             * Partition Id를 key로 ExecutionContext값을 설정한다.
             */
            executionContextMap.put("id_" + ix, executionContext);
        }

        return executionContextMap;
    }
}

Slave Step에서는 전달 받은 ExecutionContext에서 대상범위를 확인한다.

@BxmCategory(logicalName = "open : 샘플용 직원정보 Iterator 처리")
public void open(ExecutionContext executionContext) throws ItemStreamException
{

    /**
     * 배치 입력파라미터 "deptNo" Get
     */
    int deptNo = executionContext.getInt("deptNo");

    /**
     * 입력받은 "deptNo" 에 대하여 샘플용 직원정보를 Iterator로 가져온다.
     */
    DSmpEmpTst001selectList01InDto inDto = new DSmpEmpTst001selectList01InDto();
    inDto.setFeduDeptNo(deptNo);
    iterator = dSmpEmpTst001.selectList01(inDto).iterator();
}