본문 바로가기
개발 기록/미국 증시 분석 자동화 시스템 운영기

[운영기 A-1] Spring Boot 파이프라인을 시간 기반에서 이벤트 기반으로 바꾼 이유 — PipelineOrchestrator 설계

Sari91 2026. 5. 18. 13:00

이 시리즈는 증시 자동화 구축기 8편의 후속입니다.
구축기를 먼저 읽지 않아도 이해할 수 있지만, 함께 읽으면 흐름이 더 자연스럽습니다.

이전 글: (운영기 시작)
다음 글: 운영기 A-2 — 독립 파이프라인들: 실적·경제지표·ETF 배당 자동화

구축기에서 솔직하게 인정한 것

구축기 3편에서 이런 내용을 썼다.

"현재는 스케줄 간격을 넉넉하게 잡아서 시간으로 해결한다."

당시엔 종목이 10개 수준이었고, 단계 사이에 여유를 충분히 두면 실제로 문제가 없었다.
그런데 종목이 늘고 파이프라인이 복잡해지면서 이 구조의 한계가 드러나기 시작했다.
이 글은 그 한계를 직접 겪고 이벤트 기반 파이프라인으로 전환한 과정의 기록이다.
 

시간 기반 구조의 4가지 문제

변경 전 구조는 단순했다. 10개 서비스가 각자의 @Scheduled로 정해진 시간에 독립 실행된다.

06:20  MarketIndexService.collectMarketIndex()
06:30  ExchangeRateService.collectExchangeRates()
06:50  ChatGptService.evaluatePreviousPredictions()
07:00  NewsCollectService.collectNews()
07:20  NewsCollectService.collectGlobalNews()
07:35  NewsCollectService.collectThemeNews()
07:40  GeminiService.refineNews()
08:30  ChatGptService.analyzeNews()
09:00  MailService.sendDailyReport()
09:10  BlogService.createBlogDraft()

 

각 단계는 서로를 모른다.

[이미지 1 — 변경 전: 10개 독립 @Scheduled 구조]

 
 

이 구조의 문제는 네 가지였다.
 
문제 1 — 이전 단계 완료 여부를 알 수 없다
 
07:40에 Gemini가 시작할 때, 07:35에 시작한 테마 뉴스 수집이 실제로 끝났는지 알 방법이 없다. 뉴스 수집이 예상보다 오래 걸리면 Gemini는 미완성 데이터를 정제하게 된다.

07:35 테마 뉴스 수집 시작 (예상 5분)
07:40 Gemini 시작 → 뉴스가 아직 DB에 없을 수 있음

 
문제 2 — 실패해도 파이프라인이 계속 진행된다
 
뉴스 수집이 0건으로 실패해도 Gemini, ChatGPT, BlogService는 각자 정해진 시간에 그냥 실행된다. 결과물은 비어있거나 전날 데이터를 재사용하게 된다. 아침에 Notion을 열었다가 초안이 이상하다는 것을 발견하고서야 알게 된다.
 
문제 3 — 수집 단계가 불필요하게 직렬로 실행된다
 
시장 지수(06:20), 환율(06:30), 뉴스(07:00 - 07:35)는 서로 의존성이 전혀 없다.
그러나 10분 ~ 15분씩 간격을 두고 순차 실행된다.
병렬로 돌릴 수 있는 작업을 시간 기반 구조가 강제로 직렬화한 것이다.
 
문제 4 — 파이프라인 흐름이 10개 파일에 분산된다
전체 파이프라인 순서를 파악하려면 서비스 파일을 하나씩 열어봐야 한다.
단계를 하나 추가하거나 순서를 바꾸면 여러 파일을 동시에 수정해야 한다.

 

PipelineOrchestrator 설계

해결 방향은 단순했다. 파이프라인 조율 책임을 한 클래스에 집중시키고, 그룹 간 연결은 완료 이벤트로 처리한다.
 
파이프라인을 3그룹으로 나눴다.
 
그룹 1 (수집): 시장 지수 + 환율 + 뉴스 — 서로 의존성 없으므로 병렬
그룹 2 (분석): 예측 평가 → Gemini 정제 → ChatGPT 분석 — 순서가 중요하므로 순차
그룹 3 (콘텐츠): 이메일 발송 → 블로그 초안 생성 — 순차
 

그룹 간 연결은 Spring ApplicationEvent로 처리한다.

 

그룹 1이 완료되면 CollectionGroupCompletedEvent를 발행하고,

그룹 2의 @EventListener가 이를 받아 즉시 시작한다.

 

// PipelineOrchestrator.java

// 그룹 1 — 유일한 @Scheduled 진입점
@Scheduled(cron = "0 20 6 * * MON-SAT", zone = "Asia/Seoul")
public void startCollectionGroup() {
    CompletableFuture<Void> marketFuture =
        CompletableFuture.runAsync(() -> marketIndexService.collectNow());
    CompletableFuture<Void> exchangeFuture =
        CompletableFuture.runAsync(() -> exchangeRateService.collectNow());
    CompletableFuture<Void> newsFuture =
        CompletableFuture.runAsync(() -> {
            newsCollectService.collectNow();
            newsCollectService.collectGlobalNow();
            newsCollectService.collectThemeNow();
        });

    CompletableFuture.allOf(marketFuture, exchangeFuture, newsFuture).join();

    // 완료 검증
    int newsCount = newsMapper.countByDate(LocalDate.now());
    if (newsCount < 30) {
        telegramService.sendError("뉴스 수집 " + newsCount + "건 — 임계값 미달, 파이프라인 중단");
        return;
    }

    eventPublisher.publishEvent(new CollectionGroupCompletedEvent(this, LocalDate.now(), newsCount));
}

// 그룹 2 — 그룹 1 완료 이벤트 수신 후 자동 시작
@EventListener
public void startAnalysisGroup(CollectionGroupCompletedEvent event) {
    chatGptService.evaluatePreviousPredictions();
    geminiService.refineNow();
    chatGptService.analyzeNow();

    // 완료 검증
    int refinedCount = newsSummaryMapper.countByDate(LocalDate.now());
    if (refinedCount < 20) {
        telegramService.sendError("정제 " + refinedCount + "건 — 임계값 미달, 파이프라인 중단");
        return;
    }

    eventPublisher.publishEvent(new AnalysisGroupCompletedEvent(this, LocalDate.now(), refinedCount));
}

// 그룹 3 — 그룹 2 완료 이벤트 수신 후 자동 시작
@EventListener
public void startContentGroup(AnalysisGroupCompletedEvent event) {
    mailService.sendReportNow();
    BlogDraftDto draft = blogService.writeDraftNow();
    telegramService.sendBlogSuccess(draft.getTitle());
}

 

각 서비스에서 @Scheduled를 제거하고 *Now() 메서드만 남겼다.
MarketIndexService.collectMarketIndex() → MarketIndexService.collectNow()로 이름도 정리했다.

 

[이미지 2 — 변경 후: PipelineOrchestrator 3그룹 이벤트 흐름도]

 

완료 검증 임계값

단순히 "끝났는지"만 확인하는 게 아니라, "충분한 결과가 나왔는지"도 검증한다.

그룹 1: 뉴스 수집30건 이상Telegram 오류 + 파이프라인 중단
그룹 2: Gemini 정제20건 이상Telegram 오류 + 파이프라인 중단
그룹 2: ChatGPT 분석1건 이상Telegram 오류 + 파이프라인 중단

 
임계값 30건은 보유 종목 15개 × 종목당 2건 최소 기대값에서 나왔다.

종목이 크게 늘면 PipelineOrchestrator의 상수값을 조정해야 한다.

 
시장 지수나 환율 수집 실패는 경고(Telegram)만 보내고 계속 진행한다.
뉴스가 없으면 분석 자체가 의미 없으므로 파이프라인을 중단하지만, 환율 정보가 없어도 뉴스 분석은 진행할 수 있기 때문이다.

 

수동 실행 시 HTTP 블로킹 문제

리팩토링 후 예상 못한 문제가 하나 생겼다.
 

@EventListener는 기본적으로 동기 실행이다.

 
그룹 1이 이벤트를 발행하면 그룹 2가 같은 스레드에서 바로 실행된다.

그리고 startCollectionGroup() 메서드는 전체 파이프라인이 끝날 때까지 반환되지 않는다.

 

이 상태는 자동 스케줄 실행(@Scheduled)에서는 문제가 없었다.

왜냐하면, 스케줄러 스레드가 블로킹돼도 다른 요청에 영향이 없기 때문이다.
 
문제는 수동 실행이었다.

설정 화면에서 "파이프라인 시작" 버튼을 누르면 Controller가 startCollectionGroup()을 직접 호출한다.

 
파이프라인 전체가 완료될 때까지(수십 분) HTTP 응답이 반환되지 않는다. Nginx 타임아웃(300초)을 넘으면 502가 떨어진다.

해결은 수동 실행 전용 @Async 메서드를 분리하는 것이었다.

 

// @Scheduled 경로 — 변경 없음
@Scheduled(cron = "0 20 6 * * MON-SAT", zone = "Asia/Seoul")
public void startCollectionGroup() { ... }

// 수동 실행 경로 — @Async로 즉시 응답 반환
@Async
public void startCollectionGroupManual() {
    startCollectionGroup(); // 백그라운드에서 실행
}

@Async
public void startAnalysisGroupManual() { ... }

@Async
public void startContentGroupManual() { ... }

 

Controller는 startCollectionGroupManual()을 호출하고

즉시 "파이프라인 시작됨 — Telegram으로 완료 알림 수신" 응답을 반환한다.

 

@EnableAsync는 SchedulingConfig에 추가하였고,

실제 파이프 라인은 백그라운드에서 실행되게 된다.

변경 효과 정리

파이프라인 진입점10개 @Scheduled1개 @Scheduled
그룹 간 연결시간 기반 (기대)ApplicationEvent (보장)
수집 실행 방식순차 (10~15분 간격)병렬 (CompletableFuture)
실패 처리실패해도 계속 진행임계값 미달 시 중단 + Telegram
흐름 파악10개 파일 열어야 함PipelineOrchestrator 1개 파일
수동 실행Controller 직접 블로킹@Async 분리, 즉시 응답

 

[이미지 3 — 실패 처리 전략 비교표]

 

변경하지 않은 것

리팩토링의 목적은 파이프라인 조율 방식이지, 각 서비스의 비즈니스 로직이 아니다.
 

  • Rate Limit용 Thread.sleep(): Gemini 3초, Finnhub/Yahoo 500ms, OpenAI 1초 — 그대로 유지
  • 각 서비스의 *Now() 메서드: 수동 API 호출로도 계속 사용 가능
  • 독립 파이프라인: 실적, 경제지표, 주간/월간 블로그는 PipelineOrchestrator와 무관하게 자체 @Scheduled 유지
  • Telegram 알림 패턴: 각 오류는 즉시 Telegram 발송

주의할 점 하나

그룹 1의 병렬 실행은 기본 ForkJoinPool.commonPool()을 사용한다.

 
현재 보유 종목 수준에서는 문제없지만, 종목이 크게 늘어나면 pool 포화로 태스크가 큐에 쌓일 수 있다.

필요 시 PipelineOrchestrator에 전용 ThreadPoolTaskExecutor를 주입해서 스레드 수를 명시적으로 제한하여 관리할 에정이다.

마무리

시간 기반 파이프라인은 "단순하게 시작하기"에는 좋은 구조다.
종목이 적고 파이프라인이 짧을 때는 충분히 동작하여 간단한 프로젝트에서는 구현이 편하다.
하지만 단계가 늘어나고 종목이 많아질수록 "이전 단계가 끝났을 것이라는 기대"가 부담이 된다.
 
이벤트 기반으로 전환하면서 가장 크게 달라진 것은 코드가 아니라 운영 경험이다.
실패가 생겼을 때 Telegram으로 즉시 알림이 오고, 어느 단계에서 멈췄는지 명확하게 알 수 있다.
조용히 실패하는 파이프라인과 명확하게 실패를 알려주는 파이프라인은 운영 부담이 완전히 다르다.
 
다음 편에서는 이벤트 파이프라인과 별도로 돌아가는 독립 파이프라인들을 다룬다.
실적 캘린더, 경제지표, ETF 배당 — 각각 완전히 다른 주기와 트리거를 가진 파이프라인을 어떻게 설계했는지 기록한다.

저작자표시 비영리 변경금지 (새창열림)

'개발 기록 > 미국 증시 분석 자동화 시스템 운영기' 카테고리의 다른 글

[운영기 A-2] 이벤트 파이프라인 옆에서 돌아가는 독립 파이프라인들 — 실적·경제지표·ETF 배당 자동화  (0) 2026.05.20

'개발 기록/미국 증시 분석 자동화 시스템 운영기' Related Articles

티스토리툴바