Reactor 파이프라인 설계기: External Task 200건 동시 처리 · ConnectableFlux 삽질

📑 목차

• 들어가며
• 환경
• 문제: 동기 처리 병목 + 락 만료
• 설계: Reactor 스트림 합성
• ConnectableFlux 기반 폴링
• flatMap(200) 동시성 제어
• 처리 스트림 + 락 갱신 스트림 합성
• Step 순차 실행 보장
• 에러 격리: onErrorResume의 함정
• 파이프라인 레벨 (정답)
• Step 레벨 (함정)
• 결과
• 배운 점

들어가며

Camunda BPM의 External Task 패턴은 프로세스 엔진과 외부 시스템 호출을 분리하는 강력한 아키텍처다. 하지만 ERP/RPA 같은 외부 호출이 수십 초씩 걸리는 환경에서 동기 처리는 전체 파이프라인을 막아버린다.

이전 고객사 BPM 프로젝트에서 Project Reactor를 활용하여 200건 동시 처리가 가능한 파이프라인을 설계한 과정을 공유한다.

환경

• Spring Boot 2.7 + Java 8
• Camunda BPM 7.13
• Project Reactor + RxJava 3
• External Task Client

---

문제: 동기 처리 병목 + 락 만료

문제	증상	영향
동기 병목	태스크를 순차 처리 (30초 + 45초 + 5초 = 80초)	10건이면 400초
락 만료	기본 60초 락, ERP 호출이 수분 걸리면 만료	다른 워커가 같은 태스크를 가져가서 중복 실행

설계: Reactor 스트림 합성

ConnectableFlux 기반 폴링

ConnectableFlux<List<LockedExternalTask>> taskStream = Flux
    .interval(Duration.ofSeconds(5))        // 5초마다 폴링
    .flatMap(tick -> fetchAndLock())         // Camunda에서 태스크 가져오기
    .publish();                             // Hot Stream으로 변환

taskStream.connect();                       // 구독 시작

publish()로 Hot Stream을 만들면 여러 구독자가 동일한 폴링 스트림을 공유한다. Cold Stream이면 구독자마다 새로운 폴링이 시작되어 Camunda에 불필요한 부하가 걸린다.

flatMap(200) 동시성 제어

taskStream
    .flatMap(tasks -> Flux.fromIterable(tasks))
    .flatMap(task -> processTask(task), 200)  // ← 최대 200건 동시 처리
    .subscribe();

처리 스트림 + 락 갱신 스트림 합성

설계의 하이라이트다. 60초 락의 75% 시점(45초)에서 자동으로 락을 연장한다.

private Mono<Void> processTask(LockedExternalTask task) {
    Mono<Void> processing = executeSteps(task);           // 실제 처리

    Flux<Void> lockRenewal = Flux.interval(Duration.ofSeconds(45))
        .flatMap(tick -> extendLock(task, 60_000));        // 45초마다 갱신

    return processing
        .mergeWith(lockRenewal)  // ← 두 스트림을 병렬로 합성
        .then();                 // processing 완료 시 lockRenewal도 자동 종료
}

Step 순차 실행 보장

Operator	동작	용도
flatMap(200)	병렬	태스크 간 동시 처리
concatMap	순차	태스크 내 Step 순서 보장

private Mono<Void> executeSteps(LockedExternalTask task) {
    return Flux.fromIterable(task.getSteps())
        .concatMap(step -> executeStep(step))  // concatMap = 순차 실행
        .then();
}

---

에러 격리: onErrorResume의 함정

파이프라인 레벨 (정답)

taskStream
    .flatMap(task -> processTask(task)
        .onErrorResume(e -> {
            handleFailure(task, e);    // Camunda에 실패 보고
            return Mono.empty();       // 파이프라인 유지
        }), 200)
    .subscribe();

Step 레벨 (함정)

운영 중 API 호출이 실패했는데 태스크가 성공으로 완료되는 버그가 발생했다.

// Before: 에러를 삼켜버리는 코드
return executeApiCall(context)
    .map(apiResult -> StepProcessingResult.builder()
        .status(SUCCESS).outputs(apiResult).build())
    .onErrorResume(err -> Mono.just(
        ProcessingResultFactory.stepFailureFrom(err)  // ← 실패를 "성공 결과"로 변환!
    ));

// After: onErrorResume 제거 — 에러를 상위로 전파
return executeApiCall(context)
    .map(apiResult -> StepProcessingResult.builder()
        .status(SUCCESS).outputs(apiResult).build());
    // onErrorResume 없음 → 에러가 상위 파이프라인으로 전파

onErrorResume은 양날의 검이다. 최상위에서는 격리 역할을 하지만, Step 레벨에서는 에러를 삼켜버리는 함정이 된다.

---

결과

지표	Before	After
동시 처리	1건	200건
락 만료	빈번	0건
에러 감지	삼켜짐	정상 전파
핸들러 구조	개별 구현	템플릿 패턴 (6개 통일)

배운 점

1. Reactor의 진가는 스트림 합성에 있다 — 처리 + 락갱신을 mergeWith로 합치는 발상이 핵심이었다
2. onErrorResume은 "어디서" 쓰느냐가 중요하다 — 위치에 따라 격리가 되기도, 에러를 삼키기도 한다
3. flatMap의 concurrency 파라미터를 항상 명시하자 — 기본값은 무제한이다. 프로덕션에서 무제한 병렬은 위험하다