1. 파일을 처리할 때 여러 스레드를 어떻게 조율할까
큰 파일이나 복잡한 데이터 처리를 할 때 자주 등장하는 과제가 있습니다: 작업을 여러 스레드로 분할하는 것입니다. 예를 들어 한 스레드는 파일에서 줄을 읽고, 다른 스레드들은 그 줄을 처리합니다(단어를 찾거나 통계를 계산 등).
어떤 문제가 있을까요?
- 모든 스레드가 동일한 리소스(예: 파일이나 공유 컬렉션)를 함께 사용하면 쉽게 오류가 납니다. 어떤 스레드는 다른 스레드를 “앞지르며” 경쟁 상태가 발생하고, 데이터가 유실되거나 메모리가 과부하될 수 있습니다.
- 스레드가 많고 조율이 없다면 — 어떤 스레드는 놀고, 다른 스레드는 과부하됩니다.
필요한 해결책:
- 스레드 간 데이터를 안전하게 교환할 수 있어야 합니다.
- 처리가 읽기보다 느린 경우에도 메모리가 넘치지 않도록 해야 합니다.
- 데이터가 끝났을 때 모든 스레드의 작업을 쉽게 종료할 수 있어야 합니다.
2. 패턴 “Producer–Consumer”(공급자–소비자)
Producer–Consumer는 스레드들이 서로 방해하지 않고 조화롭게 동작하도록 돕는 고전적인 패턴입니다.
여기에는 두 가지 역할이 있습니다. Producer(공급자)는 데이터를 생성합니다 — 예를 들어 파일에서 줄을 읽거나 네트워크에서 메시지를 받아 공용 큐에 넣습니다. Consumer(소비자)는 그 큐에서 데이터를 가져와 처리합니다. 예를 들어 단어를 세거나, 데이터베이스에 저장하거나, 다른 파일에 씁니다.
핵심 아이디어는 이 두 유형의 스레드가 독립적으로 동작한다는 것입니다. 공급자는 소비자가 처리하는 속도보다 더 빨리 읽을 수도 있고 그 반대일 수도 있지만, 그럼에도 서로를 막지 않습니다. 그 사이에는 작업 속도를 평준화하는 버퍼 — 큐가 있습니다.
시각적 다이어그램
[Fayl] --(chitaet)--> [Producer] --(kladyot v ochered’)--> [BlockingQueue] --(beryot)--> [Consumer] --(obrabatyvaet)3. BlockingQueue로 구현
Java에서는 이런 교환을 위해 BlockingQueue 인터페이스(예: ArrayBlockingQueue)가 제격입니다.
BlockingQueue란?
BlockingQueue는 크기가 제한된 스레드 안전 큐로, 동기화를 스스로 처리합니다. 공급자가 원소를 추가하려는데 큐가 이미 가득 찼다면 해당 스레드는 자동으로 블로킹되어 공간이 생길 때까지 기다립니다. 마찬가지로 소비자가 원소를 가져가려는데 큐가 비어 있다면, 누군가 큐에 무언가를 넣을 때까지 기다립니다.
이런 메커니즘은 스레드 “경쟁”과 메모리 초과 문제를 자동으로 해결합니다. 공급자는 큐에 불필요한 원소를 과다하게 밀어 넣지 못하고, 소비자는 비어 있는 상태에서 일하려 하지 않습니다. 모든 스레드가 차분하고 조화롭게 작업을 수행합니다.
큐 생성 예시
import java.util.concurrent.*;
BlockingQueue<String> queue = new ArrayBlockingQueue<>(100); // 100개 요소 버퍼
100은 큐의 최대 원소 수입니다. producer가 더 빠르게 동작하면, consumer가 일부 데이터를 처리할 때까지 기다리게 됩니다.
4. 스레드 조율: backpressure와 종료
큐 크기 제한(backpressure)
Backpressure는 consumer가 데이터를 처리하지 못해 지연될 때 producer가 메모리를 “범람”시키지 못하도록 막는 메커니즘입니다.
- 큐가 가득 차면 producer는 자동으로 “감속”합니다(put() 메서드가 블록됨).
- 큐가 비어 있으면 consumer는 기다립니다(take() 메서드가 블록됨).
이로써 스레드 속도가 달라도 시스템이 안정적으로 동작합니다.
종료: “poison pill”
producer가 파일 읽기를 끝냈다면, 더 이상 데이터가 없고 이제 종료해야 함을 consumer들에게 알려야 합니다.
해결책:
- 큐에 특별한 객체 — “poison pill”을 넣습니다(예: 문자열 "__END__" 또는 다른 특수 값이지만 null은 아님).
- consumer는 “poison pill”을 받으면 종료할 때임을 이해합니다.
consumer가 여러 개라면, consumer 수만큼 “poison pill”을 넣어야 합니다!
5. 파이프라인 예제: 파일 읽기와 줄 처리
간단한 파이프라인을 구현해 봅시다:
- 하나의 스레드는 파일에서 줄을 읽어 큐에 넣습니다.
- 여러 스레드는 큐에서 줄을 꺼내 단어 수를 세고 결과를 출력합니다.
단계 1. Producer — 파일 리더
import java.io.*;
import java.util.concurrent.*;
public class FileProducer implements Runnable {
private final BlockingQueue<String> queue;
private final File file;
private final int consumerCount;
private final String POISON_PILL = "__END__";
public FileProducer(BlockingQueue<String> queue, File file, int consumerCount) {
this.queue = queue;
this.file = file;
this.consumerCount = consumerCount;
}
@Override
public void run() {
try (BufferedReader reader = new BufferedReader(new FileReader(file))) {
String line;
while ((line = reader.readLine()) != null) {
queue.put(line); // 큐가 가득 차면 — 대기
}
} catch (IOException | InterruptedException e) {
e.printStackTrace();
} finally {
// 각 consumer에 대해 poison pill 넣기
try {
for (int i = 0; i < consumerCount; i++) {
queue.put(POISON_PILL);
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
}
}
단계 2. Consumer — 줄 처리기
import java.util.concurrent.BlockingQueue;
public class LineConsumer implements Runnable {
private final BlockingQueue<String> queue;
private final String POISON_PILL = "__END__";
public LineConsumer(BlockingQueue<String> queue) {
this.queue = queue;
}
@Override
public void run() {
try {
while (true) {
String line = queue.take(); // 큐가 비어 있으면 — 대기
if (POISON_PILL.equals(line)) {
break; // 작업 종료
}
int wordCount = line.trim().isEmpty() ? 0 : line.trim().split("\\s+").length;
System.out.println(Thread.currentThread().getName() + ": " + wordCount + "개의 단어: " + line);
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
}
단계 3. Executors로 파이프라인 실행
import java.io.File;
import java.util.concurrent.*;
public class PipelineDemo {
public static void main(String[] args) throws Exception {
int consumerCount = 3;
BlockingQueue<String> queue = new ArrayBlockingQueue<>(100);
File file = new File("input.txt"); // 사용자 파일
// producer 시작
Thread producer = new Thread(new FileProducer(queue, file, consumerCount));
producer.start();
// ExecutorService로 consumers 시작
ExecutorService consumers = Executors.newFixedThreadPool(consumerCount);
for (int i = 0; i < consumerCount; i++) {
consumers.submit(new LineConsumer(queue));
}
// producer 종료 대기
producer.join();
// consumers 종료 대기
consumers.shutdown();
consumers.awaitTermination(1, TimeUnit.MINUTES);
System.out.println("처리가 완료되었습니다!");
}
}
6. 시각화: 파이프라인 동작 다이어그램
flowchart LR
A[Producer: 파일 읽기] -- put() --> Q[BlockingQueue]
Q -- take() --> C1[Consumer 1: 단어 수 세기]
Q -- take() --> C2[Consumer 2: 단어 수 세기]
Q -- take() --> C3[Consumer 3: 단어 수 세기]
A -.->|poison pill| Q
Q -.->|poison pill| C1
Q -.->|poison pill| C2
Q -.->|poison pill| C3
7. 유용한 포인트와 best practices
- 큐의 크기를 제한하세요 — 메모리 초과를 방지합니다.
- 종료에는 “poison pill”을 사용하세요 — 그렇지 않으면 consumer들이 영원히 대기할 수 있습니다.
- null을 “poison pill”로 사용하지 마세요. 큐에 실제 null 값이 들어갈 가능성이 있다면 더더욱 안 됩니다. 특수 문자열이나 객체를 사용하세요.
- InterruptedException을 처리하세요 — 스레드를 올바르게 종료하는 데 중요합니다.
- ExecutorService를 사용해 스레드 풀을 관리하세요 — 직접 스레드를 만드는 것보다 더 쉽고 안전합니다.
8. 파이프라인 구현 시 흔한 실수
오류 №1: 무제한 큐 → OutOfMemoryError. LinkedBlockingQueue를 크기 제한 없이 사용하면, consumer가 따라가지 못할 때 producer가 메모리를 “범람”시킬 수 있습니다.
오류 №2: “poison pill”을 넣지 않음 → consumer가 대기 상태로 멈춤. “독약 알약”을 넣는 것을 잊으면, 소비자 스레드는 영원히 새 데이터를 기다리게 됩니다.
오류 №3: “poison pill”을 하나만 넣었는데 consumer는 여러 개. 각 소비자 스레드마다 “알약”이 하나씩 필요합니다 — 그렇지 않으면 모두 종료되지 않습니다.
오류 №4: InterruptedException을 처리하지 않음. 스레드가 인터럽트되면 플래그를 복구하는 등 올바르게 종료해야 하며, 그렇지 않으면 스레드가 “멈춘” 상태로 남을 수 있습니다.
오류 №5: 동기화 없이 스레드 간에 공용 변수를 사용. “바퀴를 다시 발명”하려 하지 마세요 — 일반 리스트나 배열 대신 BlockingQueue를 사용하세요.
GO TO FULL VERSION