[Java/Spring] 파일 기반 재처리 아키텍처 - BlockingQueue로 데이터 유실 막기

배경

 멀티스레드 배치 시스템에서 벌크 INSERT가 간헐적으로 타임아웃으로 실패했다. 기존 코드는 예외를 로그로만 남겼기 때문에 실패한 수천 건이 그대로 유실됐다. 게다가 타임아웃이 발생하는 시점은 DB 부하가 높은 구간이다. 실패가 연달아 수 천건씩 N번 발생했다.

 

 

 

시도했던 방식

 단순히 예외를 잡아 재시도하는 방식은 배치 스레드를 지연시키고, 실패가 연달아 발생하면 재시도 로직도 함께 실패한다.

 

 

 타임아웃이 발생하는 시점은 DB 부하가 높은 구간이다. 재시도를 즉시 실행하면 같은 부하 구간에서 다시 실패할 가능성이 높고, 그 사이 다른 배치도 같은 스레드풀을 기다린다. 실패가 연달아 수 천건씩 N번 발생하는 패턴에서 인라인 재시도는 수십 번 연속으로 쌓여 배치 전체를 마비시킬 수 있다.

 

 

 

 대신 실패 데이터를 파일로 보존한 뒤 별도 스케줄러로 재처리하는 구조를 선택했다. 배치 스레드는 실패를 큐에 넣는 것만 하고 즉시 다음 chunk로 넘어간다. DB 부하가 해소된 뒤, 별도 스케줄러가 단건 INSERT로 재처리하면 타임아웃이 발생하지 않는다.

 

 

 

 

---

아키텍처 설계

 

 

 전체 흐름은 실패 감지 → 큐에 삽입 → 파일에 쓰기 → 재처리 네 단계로 나뉜다. 각 단계는 서로 다른 스레드에서 독립적으로 동작한다.

 

 배치 스레드는 실패를 큐에 넣는 순간 할 일이 끝난다. 큐 뒤에서 무슨 일이 일어나는지 알 필요가 없다. FailFileWriter는 큐에서 데이터를 꺼내 파일에 쓰는 일만 한다. FailScheduler는 파일이 언제 만들어졌는지 신경 쓰지 않고 정해진 시각에 파일을 읽어 재처리한다.

 

 각 단계가 다음 단계를 직접 호출하지 않고, 큐와 파일이라는 중간 저장소를 통해서만 데이터를 넘긴다. 각 컴포넌트가 독립적으로 동작하기 때문에 재처리 로직이 배치 성능에 영향을 주지 않는다.

 

 

 

 

 

---

구현 상세

DAO 레이어 - INSERT 실패 감지

public int collectProductData(User user, List<Product> products, ...) {
    try (SqlSession sqlSession = ...) {
        sqlSession.insert("productMapper.insertBulk", products);
        return products.size();
    } catch (Exception e) {
        failQueue.offerAll(user, products, ...);  // 실패 → 큐 삽입
        return 0;
    }
}

 

 DAO 레이어에서 실패를 감지하면 예외를 상위로 던지지 않고 `FailQueue.offerAll()`을 호출해 실패 데이터를 큐에 넣는다. 배치 스레드 입장에서 실패 처리는 여기서 끝나고, 이후 파일 기록과 재처리는 별도 컴포넌트가 담당한다.

 

 큐 삽입이 블로킹되지 않기 때문에 배치 스레드는 즉시 다음 chunk로 넘어간다.

 

 

 

FailQueue - 공유 큐

private final BlockingQueue<Object> queue = new LinkedBlockingQueue<>(200_000);

public void offerAll(UserInfo user, List<Product> products) {
    for (Product product : products) {
        queue.offer(new FailedProduct(user, product));
    }
}

public boolean offer(CloseRequest closeRequest) {
    return queue.offer(closeRequest);
}

 

 FailQueue는 LinkedBlockingQueue로 구현된 공유 버퍼다. DAO 레이어(producer)와 FailFileWriter(consumer) 사이를 연결한다. 배치 스레드가 여러 개여도 큐는 하나이기 때문에 동시성 처리는 BlockingQueue가 내부적으로 보장한다. 

 

 FailedProduct(실패 데이터)와 CloseRequest(배치 완료 신호) 두 타입을 같은 큐로 관리해 consumer가 순서를 보장받는다.

 

•  offer vs put: put은 큐가 꽉 차면 공간이 생길 때까지 블로킹한다. 배치 스레드가 큐 삽입 때문에 멈추면 안 되므로 블로킹하지 않는 offer를 사용했다. 큐가 꽉 차서 offer가 false를 반환하는 상황은 200,000건이 모두 쌓인 극단적인 경우로, 실운영에서는 발생하지 않을 것으로 판단했다.

 

 

 

FailFileWriter - 전용 consumer 스레드

필드 구성

private final FailQueue failQueue;
private final Map<String, BufferedWriter> writerMap = new ConcurrentHashMap<>();
private final ObjectMapper objectMapper = new ObjectMapper();
private volatile boolean running = true;

 

 

필드 설명

필드	설명
writerMap	키로 Writer 관리
running	volatile 선언으로 @PreDestroy와 consumer 스레드 간 가시성 보장
objectMapper	consumer 스레드가 1개이므로 인스턴스 공유 안전

 

• writerMap : userId#...#...#date 형태의 키로 Writer 인스턴스를 관리한다. 배치 유형, 날짜 등 조합마다 파일이 하나씩 열리기 때문에, 키만 보면 어떤 배치의 실패 데이터가 어느 파일에 기록되고 있는지 알 수 있다. ConcurrentHashMap을 쓴 이유는 consumeLoop(consumer 스레드)와 shutdown()(@PreDestroy, 메인 스레드)이 동시에 맵에 접근할 수 있어서다. consumer 스레드가 하나뿐이더라도 종료 시점에 두 스레드가 맵을 동시에 읽고 쓸 수 있으므로 일반 HashMap은 안전하지 않다.

 

• running 플래그는 volatile로 선언한다. @PreDestroy가 실행되는 메인 스레드에서 false로 변경한 값이 consumer 스레드에 즉시 보여야 하기 때문이다. volatile이 없으면 JVM이 레지스터나 CPU 캐시에 값을 들고 있을 수 있어 consumer 스레드가 변경을 인식하지 못할 수 있다.

 

 

voliatile에 대한 자세한 내용은 이전에 다룬 적이 있다.

[Java] volatile - 멀티스레드 가시성 문제 트러블슈팅

 

 

 

 

스레드 생명주기

@PostConstruct
public void start() {
    Thread t = new Thread(this::consumeLoop, "fail-file-writer");
    t.start();
}

@PreDestroy
public void shutdown() {
    running = false;
    for (BufferedWriter writer : writerMap.values()) {
        try { writer.close(); }
        catch (IOException e) { log.error("shutdown: 파일 닫기 실패"); }
    }
}

 

애플리케이션 기동 시 @PostConstruct가 consumer 스레드를 하나 띄운다. 스레드 이름을 fail-file-writer로 명시한 것은 스레드 모니터링에서 역할을 식별하기 위해서다.

 

@PreDestroy에서는 running = false로 루프를 빠져나오게 한 뒤 열려있는 Writer를 모두 닫는다.

 

 

 

consumeLoop

private void consumeLoop() {
    try {
        while (running) {
            Object object = failQueue.take(); // 큐가 빌 때까지 블로킹

            if (object instanceof FailedProduct(UserInfo user, Product product)) {
                String key = user.getId() + "#" + .. + "#" + LocalDate.now();

                BufferedWriter writer = writerMap.computeIfAbsent(key, k -> createWriter(k));
                writer.write(objectMapper.writeValueAsString(product));
                writer.newLine();
                writer.flush(); // 매 건마다 flush — 프로세스 종료 시 유실 방지
            }

            if (object instanceof CloseRequest(String userId, ..)) {
                closeWriterInternal(userId + "#" + ..);
            }
        }
    } catch (Exception e) {
        log.error("consumeLoop 비정상 종료", e);
    }
}

 

 take()는 큐가 빌 때까지 블로킹한다. poll()처럼 타임아웃을 걸어 주기적으로 루프를 돌게 할 수도 있지만, 데이터가 없을 때 CPU를 낭비하지 않으려면 take()로 블로킹하는 것이 낫다. 배치가 오랫동안 실행되지 않는 시간대에도 스레드가 CPU를 점유하지 않는다.

 

 FailedProduct를 꺼내면, 키로 Writer를 찾거나 새로 생성해 JSON 직렬화 결과를 한 줄로 기록하고 즉시 flush()한다. BufferedWriter는 내부 버퍼가 차야 디스크에 쓰기 때문에, flush() 없이 프로세스가 비정상 종료되면 버퍼에 남은 데이터가 사라진다. 재처리 대상을 보존하는 역할이므로 I/O 횟수보다 데이터 안전을 우선했다.

 

 CloseRequest를 꺼내면 해당 배치의 Writer를 flush/close한다. close 이후 같은 키로 데이터가 다시 들어오면 computeIfAbsent가 Writer를 새로 생성하기 때문에 APPEND 옵션이 이 경우에도 중요하다.

 

 

 

 

파일 생성 - APPEND 옵션

private BufferedWriter createWriter(String key) {
    try {
        Path dir = Paths.get(FAIL_PATH);
        Files.createDirectories(dir);
        Path file = dir.resolve(key + ".txt");
        return Files.newBufferedWriter(file,
            StandardOpenOption.CREATE,
            StandardOpenOption.APPEND // 재시작 후에도 기존 파일에 이어쓰기
        );
    } catch (IOException e) {
        log.error("[{}] 파일 생성 실패", key);
        return null;
    }
}

 

 APPEND 옵션이 없으면 Files.newBufferedWriter의 기본값인 TRUNCATE_EXISTING이 적용돼 기존 파일 내용이 지워진다. 배치 실행 중에 앱이 재시작되면 이전 실패 데이터가 파일에 남아있는 상태에서 새 실패 데이터가 추가될 수 있는데, APPEND 덕분에 이전 데이터를 덮어쓰지 않고 뒤에 이어쓴다. 스케줄러는 파일 전체를 순서대로 처리하므로 재시작 전후 데이터가 같은 파일에 섞여 있어도 문제없다.

 

 

 

Writer 종료 - prefix 매칭

private void closeWriterInternal(String requestKey) {
    writerMap.entrySet().removeIf(entry -> {
        if (entry.getKey().startsWith(requestKey + "#")) {
            try {
                entry.getValue().flush();
                entry.getValue().close();
            } catch (IOException e) {
                log.error("{} writer 닫기 실패", entry.getKey(), e);
            }
            return true;
        }
        return false;
    });
}

 

 

 

 

 

CloseRequest를 큐에 넣은 이유

 처음에는 closeWriter()를 직접 호출했다. 그런데 큐에 아직 남아있는 FailedProduct가 처리되기 전에 close가 실행되는 문제가 있었다.

 

 이를 해결하려고 closeRequestIds Set을 두고 consumeLoop에서 FailedProduct 처리 직전에 체크하는 방식을 시도했다. 하지만 이 방식은 마지막 FailedProduct 이후 큐가 비면 take()에서 블로킹 → close 신호가 Set에 있어도 영원히 실행되지 않는 문제가 있었다.

 

 최종 해결책은 CloseRequest를 FailedProduct와 같은 큐에 넣는 것이다. 단일 consumer 스레드가 큐를 순서대로 소비하므로, CloseRequest 이전의 모든 FailedProduct가 반드시 먼저 처리된다.

 

 

 

 

 

FailServiceImpl - 재처리 로직

public void retryInsert() {
    List<Path> failFiles = Files.list(failDir)
        .filter(p -> p.getFileName().toString().endsWith(".txt"))
        .toList();

    for (Path failFile : failFiles) {
        String fileName = failFile.getFileName().toString();
        String userId   = extractUserId(fileName);
        ...

        UserInfo user = userDao.findUser(userId, ..);

        try (BufferedReader reader = Files.newBufferedReader(failFile)) {
            String line;
            while ((line = reader.readLine()) != null) {
                Product product = objectMapper.readValue(line, Product.class);
                insertProduct(user, product); // 단건 INSERT
            }
        }

        Files.delete(failFile);
    }
}

 

 

왜 단건 INSERT인가

 벌크 INSERT가 타임아웃으로 실패했다는 것은 DB에 수천 건을 한 번에 밀어넣는 과정에서 max_statement_time을 초과했다는 의미다. 재처리를 다시 벌크로 하면 같은 이유로 또 실패할 가능성이 있다. 단건 INSERT는 한 건씩 순차 처리하기 때문에 쿼리 하나의 실행 시간이 짧아 타임아웃이 발생하지 않는다. 스케줄러가 배치 부하가 낮은 시간대에 실행되므로 단건이어도 전체 처리 시간은 문제가 되지 않는다.

 

 

 

 

FailScheduler

@Scheduled(cron = "0 0 16 * * *")
public void retry() {
    failService.retryInsert();
}

 

 

 

---

정리

 핵심은 실패 감지 → 큐 → 파일 → 재처리 흐름을 배치 로직과 완전히 분리한 것이다. DAO 레이어는 실패를 큐에 넣기만 하고, 파일 쓰기와 재처리는 각자 독립된 컴포넌트가 담당한다. 덕분에 재처리 로직이 배치 성능에 영향을 주지 않고, 컴포넌트별로 독립적인 테스트와 교체가 가능하다.