Spring Batch 멀티스레드 배치 중지 처리 2편

— 동적 Deadline 계산, 파티션 구조 적용, REST API 강제 중지

1편에서 ChunkListener + setTerminateOnly()로 배치 중지 문제를 해결했다.
2편에서는 고정 시간이 아닌 데이터 건수 기반 동적 Deadline 계산과
파티션 멀티스레드 구조에서의 적용, REST API를 통한 강제 중지까지 다룬다.

---

목차

1. 고정 Deadline의 한계
2. 동적 Deadline 계산 설계
3. 배치 처리 방식별 Deadline 계산
4. 파티션 구조에서의 Deadline 계산
5. secondsPerItem 설정 주의사항
6. REST API를 통한 배치 강제 중지
7. Processor 단계에서의 중지 처리
8. 멀티스레드 환경에서 스텝 정보 확인
9. 전체 흐름 요약

---

1. 고정 Deadline의 한계

1편에서 JobDeadlineConfig에 고정 시간으로 deadline을 관리했다.

@Component
public class JobDeadlineConfig {

    private final Map<String, LocalTime> deadlineMap = new ConcurrentHashMap<>();

    public JobDeadlineConfig() {
        deadlineMap.put("tsboScrapJob", LocalTime.of(23, 0));  // 고정 시간
    }
}

고정 시간 방식의 문제는 명확하다.

데이터 10건 → 실제 10분이면 처리 완료
but 고정 deadline이 23:00이면 → 23:00까지 대기 후 종료

외부 API 호출 시간이 제한되어 있는 상황이라면, 남은 데이터를 처리할 수 있는 시간까지만 돌도록 deadline을 동적으로 계산하는 것이 훨씬 효율적이다.

---

2. 동적 Deadline 계산 설계

계산 흐름

배치 시작
  → 데이터 건수 조회
  → 건당 처리시간 × 데이터 수 = 예상 소요시간
  → 시작시간 + 예상 소요시간 = 예상 종료시간
  → min(예상 종료시간, API 마감 하드리밋) → deadlineMap 업데이트
  → 배치 Job 실행

DeadlineCalculator 공통 구조

@Slf4j
public class DeadlineCalculator {

    private static final int BUFFER_MINUTES = 5;

    private final LocalTime hardDeadline;   // API 절대 마감시간 (초과 불가)
    private final long secondsPerItem;      // 건당 평균 처리시간(초)

    public DeadlineCalculator(LocalTime hardDeadline, long secondsPerItem) {
        this.hardDeadline = hardDeadline;
        this.secondsPerItem = secondsPerItem;
    }

    // 하드리밋 초과 시 클램핑
    private LocalTime clampToHardDeadline(LocalTime estimatedEnd) {
        if (estimatedEnd.isAfter(hardDeadline)) {
            log.warn("[DeadlineCalculator] 예상종료({})가 하드리밋({}) 초과 → 하드리밋으로 설정",
                    estimatedEnd, hardDeadline);
            return hardDeadline;
        }
        return estimatedEnd;
    }
}

---

3. 배치 처리 방식별 Deadline 계산

Spring Batch에는 크게 3가지 처리 방식이 있으며 각각 계산 방법이 다르다.

처리 방식 비교

방식	구조	병렬 여부	계산 기준
Sequential	단일 Step	❌ 순차	총 청크 수 × 청크당 처리시간
TaskExecutor	단일 Step + TaskExecutor	✅ 청크단위 병렬	스레드당 청크 수 × 청크당 처리시간
Partitioner	MasterStep + Partitioner	✅ 파티션단위 병렬	파티션당 청크 수 × 청크당 처리시간

TaskExecutor vs Partitioner 병렬 처리 차이

TaskExecutor — 청크 단위 동적 분배

전체 10,000건
├── 스레드1 → 청크1 처리 완료 → 다음 청크 가져감 (동적)
├── 스레드2 → 청크2 처리 완료 → 다음 청크 가져감 (동적)

Partitioner — 파티션 단위 고정 분배

전체 10,000건
├── 파티션1 (1~5000건)  → 스레드1이 전담 (고정)
└── 파티션2 (5001~10000건) → 스레드2가 전담 (고정)

방식별 계산 메서드

/**
 * [단일 스레드] 순차 처리 기준 deadline 계산
 */
public LocalTime calculateBySequential(long itemCount, int chunkSize) {

    long totalChunks = (long) Math.ceil((double) itemCount / chunkSize);
    long actualItemsPerChunk = Math.min(chunkSize, itemCount);
    long secondsPerChunk = actualItemsPerChunk * secondsPerItem;
    long totalSeconds = totalChunks * secondsPerChunk;
    long totalWithBuffer = totalSeconds + (BUFFER_MINUTES * 60L);

    LocalTime estimatedEnd = LocalTime.now().plusSeconds(totalWithBuffer);

    log.info("[DeadlineCalculator-Sequential] "
                    + "전체건수={}, 청크사이즈={}, 총청크={}, "
                    + "청크당건수={}, 청크당={}초, "
                    + "예상소요={}분, 예상종료={}, 하드리밋={}",
            itemCount, chunkSize, totalChunks,
            actualItemsPerChunk, secondsPerChunk,
            totalWithBuffer / 60, estimatedEnd, hardDeadline);

    return clampToHardDeadline(estimatedEnd);
}

/**
 * [TaskExecutor] 청크 단위 병렬처리 기준 deadline 계산
 * - 청크를 여러 스레드가 동적으로 나눠가져가는 방식
 */
public LocalTime calculateByChunk(long itemCount, int chunkSize, int threadCount) {

    long totalChunks = (long) Math.ceil((double) itemCount / chunkSize);
    long chunksPerThread = (long) Math.ceil((double) totalChunks / threadCount);
    long secondsPerChunk = Math.min(chunkSize, itemCount) * secondsPerItem;
    long totalSeconds = chunksPerThread * secondsPerChunk;
    long totalWithBuffer = totalSeconds + (BUFFER_MINUTES * 60L);

    LocalTime estimatedEnd = LocalTime.now().plusSeconds(totalWithBuffer);

    log.info("[DeadlineCalculator-Chunk] "
                    + "전체건수={}, 청크사이즈={}, 총청크={}, "
                    + "스레드={}, 스레드당청크={}, 청크당={}초, "
                    + "예상소요={}분, 예상종료={}, 하드리밋={}",
            itemCount, chunkSize, totalChunks,
            threadCount, chunksPerThread, secondsPerChunk,
            totalWithBuffer / 60, estimatedEnd, hardDeadline);

    return clampToHardDeadline(estimatedEnd);
}

/**
 * [Partitioner] 파티션 단위 병렬처리 기준 deadline 계산
 * - 데이터가 파티션으로 고정 분배된 후 청크 단위로 처리
 * - 병렬이므로 가장 오래 걸리는 파티션 기준으로 계산
 */
public LocalTime calculateByPartition(long itemCount, int chunkSize, int partitionSize) {

    long chunksPerPartition = (long) Math.ceil((double) partitionSize / chunkSize);
    long secondsPerChunk = Math.min(chunkSize, partitionSize) * secondsPerItem;
    long totalSeconds = chunksPerPartition * secondsPerChunk;

    // 버퍼를 동적으로 조정 (처리시간의 10%, 최소 30초 ~ 최대 5분)
    long bufferSeconds = Math.min(
            Math.max(totalSeconds / 10, 30),
            BUFFER_MINUTES * 60L
    );

    long totalWithBuffer = totalSeconds + bufferSeconds;

    LocalTime estimatedEnd = LocalTime.now().plusSeconds(totalWithBuffer);

    log.info("[DeadlineCalculator-Partition] "
                    + "전체건수={}, 청크사이즈={}, 파티션당건수={}, "
                    + "파티션당청크={}, 청크당={}초, 버퍼={}초, "
                    + "예상소요={}분, 예상종료={}, 하드리밋={}",
            itemCount, chunkSize, partitionSize,
            chunksPerPartition, secondsPerChunk, bufferSeconds,
            totalWithBuffer / 60, estimatedEnd, hardDeadline);

    return clampToHardDeadline(estimatedEnd);
}

---

4. 파티션 구조에서의 Deadline 계산

현재 파티션 구조

@Bean
public Step tsboMasterStep() {
    return stepBuilderFactory.get("tsboMasterStep")
            .partitioner("tsboWorkStep", new TsboScrapPartitioner(jobDeadlineConfig))
            .step(tsboWorkStep())
            .gridSize(2)
            .taskExecutor(taskExecutor)
            .build();
}

@Bean
public Step tsboWorkStep() {
    return stepBuilderFactory.get("tsboWorkStep")
            .<HashMap<String, Object>, List<TfResult>>chunk(2)
            .reader(tsboScrapItemReader)
            .processor(tsboScrapItemProcessor)
            .writer(tsboScrapItemWriter)
            .listener(stopAtDeadlineListener)  // 1편에서 구현한 리스너
            .build();
}

Partitioner에서 Deadline 계산하는 이유

파티셔너는 gridSize(스레드 수)와 전체 데이터 건수가 확정되는 시점이므로 여기서 계산하는 것이 가장 자연스럽다.

partition(gridSize) 호출 시점
  → rtsReqList.size() : 전체 건수 확정 ✅
  → gridSize          : 실제 스레드 수 확정 ✅
  → 두 값이 모두 확정된 이 시점에 deadline 계산

⚠️ TsboScrapPartitioner는 new로 생성하므로 DeadlineCalculator도 내부에서 new로 생성한다.
DeadlineCalculator는 외부 의존성 없는 순수 계산 클래스이므로 문제없다.

public class TsboScrapPartitioner implements Partitioner {

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

    private final JobDeadlineConfig jobDeadlineConfig;
    private final DeadlineCalculator deadlineCalculator;
    private final int chunkSize;

    private static final String JOB_NAME = "tsboScrapJob";

    public TsboScrapPartitioner(JobDeadlineConfig jobDeadlineConfig,
                                 int chunkSize,
                                 long secondsPerItem) {
        this.jobDeadlineConfig = jobDeadlineConfig;
        this.chunkSize = chunkSize;
        this.deadlineCalculator = new DeadlineCalculator(LocalTime.of(12, 0), secondsPerItem);
    }

    @Override
    public Map<String, ExecutionContext> partition(int gridSize) {

        List<HashMap<String, Object>> rtsReqList = TsboRequestHolder.fullList;

        int totalItems = rtsReqList.size();
        int partitionSize = totalItems / gridSize;  // 파티션당 실제 건수

        // ✅ 파티션당 실제 건수 기반으로 deadline 계산
        LocalTime deadline = deadlineCalculator.calculateByPartition(
                totalItems,
                chunkSize,
                partitionSize
        );
        jobDeadlineConfig.updateDeadline(JOB_NAME, deadline);

        logger.info("[Partitioning] gridSize={}, totalItems={}, partitionSize={}, deadline={}",
                gridSize, totalItems, partitionSize, deadline);

        // 파티션 분할
        Map<String, ExecutionContext> result = new HashMap<>();
        int len = rtsReqList.size();

        for (int i = 0; i < gridSize; i++) {
            int from = i * partitionSize;
            int to = (i == gridSize - 1) ? len : from + partitionSize;

            ExecutionContext ctx = new ExecutionContext();
            ctx.putInt("fromIndex", from);
            ctx.putInt("toIndex", to);
            result.put("partition" + i, ctx);

            logger.debug("[Partitioning] Created partition{} : from={}, to={}", i, from, to);
        }

        return result;
    }
}

계산 예시

전체 19건, gridSize=2, chunk=2, secondsPerItem=300초(5분)

파티션당 건수 = 19 / 2 = 9건
파티션당 청크 = ceil(9 / 2) = 5개
청크당 처리시간 = min(2, 9) × 300초 = 600초 = 10분
파티션당 처리시간 = 5 × 600초 = 3000초 = 50분
버퍼 = max(3000 / 10, 30) = 300초 = 5분
예상소요 = 3000 + 300 = 3300초 = 55분
시작 09:00 → 예상 종료 09:55  ✅

---

5. secondsPerItem 설정 주의사항

⚠️ secondsPerItem은 실제 외부 API 처리시간을 기준으로 설정해야 한다.

잘못된 설정 예시

secondsPerItem = 4L  →  4초로 계산
실제 외부 API 처리시간  →  300초 (5분)

결과: 예상 소요시간이 실제보다 훨씬 짧게 계산됨 ❌

deadline이 너무 일찍 끝나거나, 반대로 너무 늦게 잡혀 API 마감 시간을 넘기는 문제가 생긴다.

실제 처리시간 측정 방법

@Override
public List<TfResult> process(HashMap<String, Object> item) throws Exception {
    long start = System.currentTimeMillis();

    List<TfResult> result = callExternalApi(item);

    long elapsed = (System.currentTimeMillis() - start) / 1000;
    log.info("[Processor] 건당 처리시간: {}초", elapsed);  // 실측값 확인

    return result;
}

실측 후 yml 반영

batch:
  tsbo:
    chunk-size: 2
    seconds-per-item: 300  # 실측 5분 = 300초

@Value("${batch.tsbo.seconds-per-item:300}")
private long secondsPerItem;

---

6. REST API를 통한 배치 강제 중지

운영 중 긴급하게 배치를 중지해야 할 때 REST API로 처리할 수 있다.

파티션 구조에서 AtomicBoolean이 안되는 이유

파티션 멀티스레드 구조에서는 각 파티션이 독립적인 StepExecution을 가진다.
따라서 AtomicBoolean 플래그를 세팅해도 각 파티션 스레드에 Step 중단이 전파되지 않는다.

AtomicBoolean.set(true)
    ↓
partition0 StepExecution → 감지 못함 ❌
partition1 StepExecution → 감지 못함 ❌

대신 각 파티션 StepExecution에 직접 setTerminateOnly()를 세팅해야 한다.

JobRegistry 등록

jobOperator.stop()을 사용하려면 Job이 JobRegistry에 등록되어 있어야 한다.

@Configuration
@RequiredArgsConstructor
public class BatchConfig {

    private final JobRegistry jobRegistry;

    @Bean
    public JobRegistryBeanPostProcessor jobRegistryBeanPostProcessor() {
        JobRegistryBeanPostProcessor postProcessor = new JobRegistryBeanPostProcessor();
        postProcessor.setJobRegistry(jobRegistry);
        return postProcessor;
    }

    @Bean
    public JobOperator jobOperator(JobLauncher jobLauncher,
                                   JobRepository jobRepository,
                                   JobExplorer jobExplorer) {
        SimpleJobOperator jobOperator = new SimpleJobOperator();
        jobOperator.setJobLauncher(jobLauncher);
        jobOperator.setJobRepository(jobRepository);
        jobOperator.setJobExplorer(jobExplorer);
        jobOperator.setJobRegistry(jobRegistry);
        return jobOperator;
    }
}

BatchControlController

@RestController
@RequestMapping("/api/batch")
@RequiredArgsConstructor
@Slf4j
public class BatchControlController {

    private final JobOperator jobOperator;
    private final JobExplorer jobExplorer;

    /**
     * 배치 강제 중지
     * POST /api/batch/{jobName}/stop
     */
    @PostMapping("/{jobName}/stop")
    public ResponseEntity<Map<String, Object>> forceStop(@PathVariable String jobName) {

        Set<JobExecution> runningJobs = jobExplorer.findRunningJobExecutions(jobName);
        if (runningJobs.isEmpty()) {
            return ResponseEntity.badRequest()
                    .body(result("NOT_RUNNING", jobName + " 실행 중인 배치 없음"));
        }

        runningJobs.forEach(jobExecution -> {

            // ① JobExecution STOPPING 처리
            try {
                jobOperator.stop(jobExecution.getId());
            } catch (NoSuchJobExecutionException | JobExecutionNotRunningException e) {
                log.warn("[BatchControl] jobOperator.stop() 실패: {}", e.getMessage());
            }

            // ② 각 파티션 StepExecution 직접 종료 처리
            // jobOperator.stop()만으로는 파티션 WorkerStep까지 전파가 안되는 경우가 있음
            jobExecution.getStepExecutions().forEach(stepExecution -> {
                if (stepExecution.getStatus() == BatchStatus.STARTED
                        || stepExecution.getStatus() == BatchStatus.STARTING) {
                    stepExecution.setTerminateOnly();
                    log.info("[BatchControl] StepExecution 중지: stepName={}, stepExecutionId={}",
                            stepExecution.getStepName(),
                            stepExecution.getId());
                }
            });
        });

        return ResponseEntity.ok(result("STOP_REQUESTED", "강제 중지 요청 완료"));
    }

    /**
     * 배치 상태 조회
     * GET /api/batch/{jobName}/status
     */
    @GetMapping("/{jobName}/status")
    public ResponseEntity<Map<String, Object>> status(@PathVariable String jobName) {

        Set<JobExecution> runningJobs = jobExplorer.findRunningJobExecutions(jobName);

        Map<String, Object> data = new HashMap<>();
        data.put("running", !runningJobs.isEmpty());
        data.put("jobExecutionIds", runningJobs.stream()
                .map(JobExecution::getId)
                .collect(Collectors.toList()));

        return ResponseEntity.ok(result("OK", "조회 성공", data));
    }

    private Map<String, Object> result(String code, String message) {
        return result(code, message, null);
    }

    private Map<String, Object> result(String code, String message, Object data) {
        Map<String, Object> map = new HashMap<>();
        map.put("code", code);
        map.put("message", message);
        map.put("data", data);
        return map;
    }
}

중지 요청 후 실제 중지 시점

⚠️ REST API로 중지를 요청해도 즉시 중지되지 않는다.

REST API 중지 요청
    ↓
각 파티션 StepExecution.setTerminateOnly() 세팅
    ↓
현재 처리 중인 청크 완료  ← 중지 불가 (계속 진행)
    ↓
beforeChunk() → 플래그 감지 → 다음 청크부터 중지 ✅

---

7. Processor 단계에서의 중지 처리

외부 API 호출처럼 건당 처리시간이 긴 경우 Processor에도 중지 체크를 추가하면 더 빠르게 중지할 수 있다.

beforeChunk()는 청크 시작 전에만 체크하지만, Processor에서 체크하면 청크 처리 중간에도 중지가 가능하다.

@StepScope + @BeforeStep 방식

파티션 멀티스레드 환경에서는 Processor가 싱글톤이면 StepExecution이 덮어써질 수 있으므로 반드시 @StepScope로 선언해야 한다.

@Component
@StepScope  // ← 파티션별 독립 인스턴스 생성을 위해 필수
@Slf4j
public class TsboScrapItemProcessor implements ItemProcessor<HashMap<String, Object>, List<TfResult>> {

    private StepExecution stepExecution;

    @BeforeStep
    public void beforeStep(StepExecution stepExecution) {
        this.stepExecution = stepExecution;
        log.info("[Processor] StepName={}, StepExecutionId={} 초기화",
                stepExecution.getStepName(),
                stepExecution.getId());
    }

    @Override
    public List<TfResult> process(HashMap<String, Object> item) throws Exception {

        // 중지 플래그 체크
        if (stepExecution.isTerminateOnly()) {
            log.info("[Processor] 중지 요청 감지 → 처리 스킵");
            return null;  // null 반환 시 writer로 넘어가지 않음
        }

        // 실제 처리 로직 (외부 API 호출)
        return callExternalApi(item);
    }
}

@BeforeStep이 안될 경우 — JobExecution 상태 체크

@Override
public List<TfResult> process(HashMap<String, Object> item) throws Exception {

    StepContext stepContext = StepSynchronizationManager.getContext();

    if (stepContext == null) {
        log.warn("[Processor] StepContext 없음");
        return null;
    }

    BatchStatus jobStatus = stepContext.getStepExecution()
            .getJobExecution()
            .getStatus();

    if (jobStatus == BatchStatus.STOPPING) {
        log.info("[Processor] Job STOPPING 감지 → 처리 스킵");
        return null;
    }

    return callExternalApi(item);
}

중지 감지 포인트 비교

beforeChunk() ── 감지 ✅ (기본, 청크 시작 전)
    ↓
read()
    ↓
process() ── 감지 추가 시 ✅ (청크 중간에서도 중지 가능)
    ↓
write()
    ↓
afterChunk()

---

8. 멀티스레드 환경에서 스텝 정보 확인

파티션 멀티스레드 환경에서 현재 어떤 파티션/스레드에서 처리 중인지 확인하는 방법이다.

@BeforeStep
public void beforeStep(StepExecution stepExecution) {
    log.info("[Processor] StepName={}, StepExecutionId={}, ThreadName={}",
            stepExecution.getStepName(),        // tsboWorkStep:partition0, partition1
            stepExecution.getId(),              // StepExecution ID
            Thread.currentThread().getName()); // tsbo-batch-1, tsbo-batch-2
}

출력 예시

[Processor] StepName=tsboWorkStep:partition0, StepExecutionId=1, ThreadName=tsbo-batch-1
[Processor] StepName=tsboWorkStep:partition1, StepExecutionId=2, ThreadName=tsbo-batch-2

TaskExecutor 스레드명 Prefix 설정

@Bean
public TaskExecutor taskExecutor() {
    ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
    executor.setCorePoolSize(2);
    executor.setMaxPoolSize(2);
    executor.setThreadNamePrefix("tsbo-batch-");  // 스레드명 prefix 설정
    executor.initialize();
    return executor;
}

---

9. 전체 흐름 요약

스케줄러 실행
    ↓
TsboScrapPartitioner.partition(gridSize)
    ├── 전체 건수 확정
    ├── 실제 gridSize 확정
    └── DeadlineCalculator.calculateByPartition() → deadline 계산
    ↓
JobDeadlineConfig.updateDeadline() → deadline 세팅
    ↓
각 파티션 WorkerStep 병렬 실행
    ├── beforeChunk() → deadline 체크 → setTerminateOnly()
    └── process() → isTerminateOnly() 체크 (빠른 중지)
    ↓
현재 청크 완료 후 중지

중지 케이스 정리

중지 원인	감지 위치	처리 방식
Deadline 초과	StopAtDeadlineListener.beforeChunk()	setTerminateOnly()
REST API 강제 중지	StopAtDeadlineListener.beforeChunk()	jobOperator.stop() + setTerminateOnly()
Processor 중간 중지	ItemProcessor.process()	isTerminateOnly() → null 반환

DeadlineCalculator 메서드 선택 기준

배치 구조	사용 메서드
단일 Step, 순차 처리	calculateBySequential()
단일 Step + TaskExecutor	calculateByChunk()
MasterStep + Partitioner	calculateByPartition()

---

✅ 1편의 ChunkListener + setTerminateOnly() + 2편의 동적 Deadline 계산을 조합하면,
데이터 건수에 맞게 배치 종료 시간을 자동으로 계산하고 안전하게 중단할 수 있다.