HIDDENLIGHTYOUTH
Articles

AI 호출 서른 번을 병렬로 돌리고 일부 실패까지 다루기

가상 사용자 서른 명의 의견을 하나씩 받으면 몇 분이 걸려요. 요청은 바로 응답하고, 뒤에서 열 개씩 동시에 호출하면서 시간 초과와 부분 실패를 어떻게 처리했는지 정리했어요.

DropThePitch에서 가장 오래 걸리는 단계는 의견 수집이에요. 페르소나 서른 명이 각자 자료를 읽고 의견을 남기는데, 한 명당 AI 호출이 한 번씩 필요해요. 호출 하나가 수십 초 걸리니, 차례로 부르면 십 분이 넘을 수도 있어요.

이 단계를 어떻게 빠르고 안전하게 만들었는지 정리했어요.

요청은 바로 돌려보내요

의견 수집 요청이 오면 서버는 결과를 기다리지 않고 바로 응답해요. HTTP 상태 코드도 200 대신 202 Accepted를 써요. "요청은 받았고, 처리는 아직"이라는 뜻이에요.

@PostMapping("/collect")
@ResponseStatus(HttpStatus.ACCEPTED)
@Operation(summary = "프로젝트 페르소나 의견 수집 시작")
public GlobalResponse<Integer> collectAllOpinions(@PathVariable Long projectId) {
    return GlobalResponse.ok(opinionCollectionService.collectAllOpinions(SecurityUtil.getUsername(), projectId));
}

응답에는 수집을 시작한 페르소나 수만 담아요. 화면은 그 뒤로 수집 상태를 주기적으로 물어보다가, 완료되면 의견 목록을 불러와요.

다만 바로 응답하기 전에 반드시 끝내야 하는 일이 있어요. 이미 수집 중인지 확인하는 것과 크레딧을 차감하는 것이에요. 이 두 가지는 요청 안에서 동기로 처리하고, 실패하면 오류로 돌려보내요. 크레딧이 모자란데 수집이 시작되면 안 되니까요.

같은 요청이 두 번 와도 한 번만

버튼을 빠르게 두 번 누르면 수집 요청이 두 번 와요. 둘 다 "아직 수집 전"이라고 판단하면 AI를 예순 번 부르고 크레딧도 두 번 빠져요.

처음에는 상태를 읽고, 확인하고, 바꾸는 순서로 생각했어요. 하지만 읽는 것과 바꾸는 것 사이에 다른 요청이 끼어들 수 있어요. 그래서 확인과 변경을 쿼리 하나로 합쳤어요.

@Modifying(clearAutomatically = true)
@Query("""
        update Project p
           set p.opinionCollectionStatus = :next
         where p.id = :projectId
           and p.opinionCollectionStatus in :expected
        """)
int updateOpinionCollectionStatus(@Param("projectId") Long projectId,
                                  @Param("expected") Collection<OpinionCollectionStatus> expected,
                                  @Param("next") OpinionCollectionStatus next);

default boolean startOpinionCollection(Long projectId) {
    return updateOpinionCollectionStatus(projectId,
            List.of(OpinionCollectionStatus.NOT_STARTED, OpinionCollectionStatus.FAILED),
            OpinionCollectionStatus.IN_PROGRESS) == 1;
}

"수집 전이거나 실패한 상태라면 수집 중으로 바꿔라"를 한 번에 실행해요. 데이터베이스는 같은 행을 동시에 고치지 못하게 막아 주기 때문에, 두 요청 중 하나만 1행을 바꾸고 나머지는 0행을 바꿔요. 0이 돌아온 요청은 이미 수집 중이라는 오류를 받아요.

실패한 상태에서도 다시 시작할 수 있게 한 것은 의도한 것이에요. 일부 의견이 실패했을 때 사용자가 다시 요청하면, 아직 의견이 없는 페르소나만 다시 모아요.

크레딧 차감은 잔액 행을 잠그고 처리해요. 크레딧은 의견 수집 말고 파일 분석에서도 빠지기 때문에, 두 곳에서 동시에 빼다가 잔액이 어긋나지 않게 하려는 것이에요.

열 개씩 동시에

서른 번의 호출을 한꺼번에 다 보내면 빠르겠지만, AI API에는 분당 요청 한도가 있어요. 여러 사용자가 동시에 수집을 요청하면 한도를 금방 넘어요. 그래서 의견 수집만을 위한 스레드 풀을 따로 만들고, 동시에 열 개까지만 돌게 했어요.

@Bean(OPINION_COLLECTION_EXECUTOR)
public Executor opinionCollectionExecutor() {
    ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
    executor.setCorePoolSize(OPINION_CONCURRENCY);
    executor.setMaxPoolSize(OPINION_CONCURRENCY);
    executor.setQueueCapacity(Integer.MAX_VALUE);
    executor.setThreadNamePrefix("opinion-ai-");
    executor.setWaitForTasksToCompleteOnShutdown(true);
    executor.setAwaitTerminationSeconds(120);
    executor.initialize();
    return executor;
}

스레드는 열 개로 고정하고, 나머지 요청은 대기열에서 차례를 기다려요. 서버를 다시 배포할 때 진행 중이던 호출이 끊기지 않도록, 종료 전에 최대 2분까지 기다려 주는 설정도 넣었어요.

한 명씩 끝나는 대로 저장해요

호출 하나는 CompletableFuture 하나로 만들어요.

private CompletableFuture<Boolean> collectOne(Long projectId, OpinionCollectionTarget target, String analysisBrief) {
    return CompletableFuture
            .supplyAsync(() -> opinionAiClient.collectOpinion(target.personaId(), analysisBrief),
                    opinionCollectionExecutor)
            .orTimeout(CALL_TIMEOUT_SECONDS, TimeUnit.SECONDS)
            .thenApply(callResult -> {
                transactionTemplate.executeWithoutResult(status ->
                        applyResult(target.opinionId(), callResult.result()));
                saveAiUsageQuietly(projectId, callResult);
                return true;
            })
            .exceptionally(throwable -> {
                log.error("[collectOne] 의견 수집 실패: opinionId={}, personaId={}",
                        target.opinionId(), target.personaId(), throwable);
                return false;
            });
}

흐름은 이래요.

  1. 전용 스레드 풀에서 AI를 불러요.
  2. 7분 안에 답이 없으면 시간 초과로 실패 처리해요.
  3. 답이 오면 그 의견을 바로 저장해요. 서른 명이 다 끝날 때까지 기다리지 않아요.
  4. 어떤 이유로든 실패하면 예외를 던지지 않고 false를 돌려줘요.

3번 덕분에 중간에 서버가 멈춰도 이미 받은 의견은 남아요. 다시 요청하면 의견이 없는 페르소나만 다시 모으면 돼요. 4번이 특히 중요해요. 한 명이 실패했다고 나머지 스물아홉 명의 결과까지 버리면 안 돼요. 실패를 값으로 바꿔 두면 다른 호출에 영향을 주지 않아요.

AI 사용량을 기록하는 부분도 조용히 실패하게 했어요. 사용량 기록은 운영을 위한 것이라, 기록에 실패했다고 사용자가 받은 의견을 무를 이유는 없어요.

모두 끝나면 결과를 판정해요

서른 개의 CompletableFuture가 모두 끝나면 성공한 수를 세요.

CompletableFuture.allOf(futures.toArray(CompletableFuture[]::new))
        .whenComplete((ignored, throwable) -> finish(projectId, futures, startedAt));

OpinionCollectionStatus status = succeeded == futures.size()
        ? OpinionCollectionStatus.COMPLETED
        : OpinionCollectionStatus.FAILED;

projectRepository.finishOpinionCollection(projectId, status);

if (status == OpinionCollectionStatus.COMPLETED) {
    eventPublisher.publishEvent(new OpinionCollectionCompletedEvent(projectId));
}

모두 성공했을 때만 리포트를 만들어요. 스물다섯 명의 의견으로 만든 리포트도 만들 수는 있지만, 연령대별로 다섯 명씩 고른 구성이 깨져서 연령대 비교가 틀어져요. 그래서 하나라도 빠지면 실패 상태로 두고, 사용자가 다시 요청하면 빠진 사람만 채우게 했어요.

스레드 풀을 만들었더니 다른 곳이 느려졌어요

전용 스레드 풀을 만든 뒤 이상한 일이 생겼어요. 의견 수집과 상관없는 파일 분석이 느려진 거예요.

원인은 스프링 부트의 기본 동작이었어요. 스프링 부트는 Executor 빈이 하나도 없을 때만 @Async용 기본 실행기를 만들어요. 의견 수집용 풀을 빈으로 등록하자 기본 실행기가 만들어지지 않았고, 이름을 지정하지 않은 @Async가 모두 의견 수집용 풀로 들어갔어요. 파일 분석이 의견 수집과 스레드 열 개를 나눠 쓰고 있었던 거예요.

그래서 분석용 실행기를 따로 만들고, 분석 쪽 @Async에는 이름을 명시했어요.

@Bean(ANALYSIS_EXECUTOR)
public Executor analysisExecutor() {
    return Executors.newVirtualThreadPerTaskExecutor();
}

분석과 썸네일 생성은 대부분 네트워크 응답을 기다리는 일이라 가상 스레드를 썼어요. 의견 수집은 동시에 몇 개를 보낼지 직접 정해야 해서 고정된 풀을 그대로 뒀어요.

정리

  • 오래 걸리는 작업은 202 Accepted로 바로 응답하고 뒤에서 처리해요. 단, 크레딧 차감과 중복 확인은 응답 전에 끝내요.
  • 상태 확인과 변경을 쿼리 하나로 합치면 같은 요청이 두 번 와도 한 번만 실행돼요.
  • 전용 스레드 풀로 동시 호출 수를 제한하면 AI API 한도를 지킬 수 있어요.
  • 호출마다 시간 제한을 두고, 실패는 예외 대신 값으로 바꿔서 다른 호출에 번지지 않게 해요.
  • Executor 빈을 하나 만들면 스프링 부트의 기본 실행기가 사라지니, @Async에는 실행기 이름을 적어 줘요.
다음 글 가상 사용자 600명 중 서른 명을 고르는 방법