1. Stream API 복습
여러분은 이미 Stream API에 익숙할 것입니다 — 이는 컬렉션을 다루는 편리한 방법으로, 데이터 처리(필터링, 정렬, 집계 등)를 위한 간결하고 읽기 쉬운 코드를 작성할 수 있게 해줍니다.
전형적인 예시는 다음과 같습니다:
List<Integer> numbers = Arrays.asList(1, 2, 3, 4, 5);
int sum = numbers.stream()
.filter(n -> n % 2 == 0)
.mapToInt(n -> n)
.sum();
System.out.println(sum); // 6 (2 + 4)
이 예제에서 컬렉션은 스트림(stream())으로 바뀌고, 그중 짝수만 걸러낸 다음 int로 변환하여 sum() 호출로 합계를 구합니다.
Stream API는 코드를 더 짧고 표현력 있게 만들어 줍니다. 단계별로 어떻게 처리할지를 서술하는 대신, 무엇을 얻고 싶은지만 선언하면 됩니다. 필요할 때는 단 한 줄로 손쉽게 병렬 처리로 전환할 수도 있습니다.
2. 병렬 스트림: 문법과 동작 원리
스트림을 병렬로 만드는 방법
간단합니다. stream() 대신 parallelStream()을 사용하거나, 기존 스트림에 .parallel()을 호출하면 됩니다.
List<Integer> numbers = ...;
int sum = numbers.parallelStream()
.filter(n -> n % 2 == 0)
.mapToInt(n -> n)
.sum();
또는 이렇게:
numbers.stream()
.parallel() // 병렬 스트림으로 전환
.filter(...)
.map(...)
.sum();
내부에서는 무엇이 일어날까요?
- 컬렉션이 자동으로 여러 조각으로 분할됩니다.
- 각 조각은 별도의 스레드에서 처리됩니다(ForkJoinPool — 전용 스레드 풀 — 사용).
- 결과들이 합쳐져 최종 값이 됩니다.
즉, 멀티코어 프로세서라면 실제로 병렬로 처리됩니다. 예를 들어 필터링과 합계 계산이 여러 코어에서 동시에 수행될 수 있습니다.
어떤 경우에 특히 유용할까요?
- 큰 컬렉션 처리(수만 개 이상의 요소).
- 각 요소에 대한 계산이 무거운 경우.
- 처리 순서를 엄격히 유지할 필요가 없는 경우.
예: 순차 스트림과 병렬 스트림 비교
큰 배열을 처리하는 간단한 예제를 보겠습니다.
import java.util.*;
import java.util.stream.*;
public class ParallelStreamDemo {
public static void main(String[] args) {
List<Integer> numbers = IntStream.rangeClosed(1, 10_000_000)
.boxed()
.collect(Collectors.toList());
// 순차 스트림
long time1 = System.currentTimeMillis();
long count1 = numbers.stream()
.filter(n -> n % 2 == 0)
.count();
long time2 = System.currentTimeMillis();
System.out.println("순차: " + (time2 - time1) + " ms, 짝수: " + count1);
// 병렬 스트림
long time3 = System.currentTimeMillis();
long count2 = numbers.parallelStream()
.filter(n -> n % 2 == 0)
.count();
long time4 = System.currentTimeMillis();
System.out.println("병렬: " + (time4 - time3) + " ms, 짝수: " + count2);
}
}
이 코드를 직접 실행해 보세요 — 대개 병렬 스트림이 더 빠르게 처리할 것입니다(특히 멀티코어 CPU라면). 하지만 항상 그런 것은 아닙니다! 자세한 내용은 아래에서 다룹니다.
3. 내부 동작: ForkJoinPool과 자동 분할
병렬 스트림은 내부적으로 ForkJoinPool.commonPool()을 사용하며, 이 풀은 사용 가능한 프로세서 코어 수(일반적으로 코어 개수)에 맞춰 스레드 수를 자동으로 관리합니다.
개략적으로:
+-----------------------------+
| 컬렉션 |
+-----------------------------+
| 1 | 2 | 3 | ... | 10 million |
+----+----+----+-----+-----------+
| | | |
v v v v
[스레드1][스레드2]...[스레드N]
| | | |
+----+----+-----------+
|
[결과 병합]
각 스레드가 자신의 조각을 처리하고, 이후 결과를 합칩니다.
4. 제약과 함정
병렬 스트림은 ‘모두 가속’ 버튼이 아닙니다. 때로는 오히려 더 느려질 수도 있습니다!
병렬화가 불리한 경우:
- 컬렉션이 작을 때(약 1000개 이하).
- 요소별 연산이 매우 빠를 때(예: 단순한 n * 2).
- 처리 순서를 엄격히 유지해야 할 때(예: 파일에 순차적으로 기록).
왜일까요? 스레드 생성과 동기화에도 비용이 듭니다. 작업이 ‘가벼운’ 경우라면, 이러한 부가 비용이 병렬화의 이점을 상쇄할 수 있습니다.
부작용은 병렬성의 적
스트림 내부 연산이 외부 변수를 변경한다면 주의하세요!
나쁜 예:
List<Integer> numbers = Arrays.asList(1, 2, 3, 4, 5);
int[] sum = {0};
numbers.parallelStream().forEach(n -> sum[0] += n);
System.out.println(sum[0]); // ??? (15를 기대하지만 어떤 값이든 나올 수 있음)
왜일까요? 여러 스레드가 동시에 하나의 변수를 변경하면 race condition(경쟁 상태)이 발생하기 때문입니다. 최종 결과가 올바르지 않을 수 있습니다.
올바른 방법 — 결과를 반환하는 스트림 메서드를 사용하세요:
int sum = numbers.parallelStream().mapToInt(n -> n).sum();
모든 컬렉션이 동일하게 병렬화에 적합한 것은 아님
일반 ArrayList는 분할이 잘 되지만, LinkedList나 무한 스트림(예: Stream.generate(...))은 그렇지 않습니다.
5. 실습: 성능 비교
예: 최댓값 찾기
import java.util.*;
import java.util.stream.*;
public class ParallelMaxDemo {
public static void main(String[] args) {
List<Integer> numbers = IntStream.rangeClosed(1, 30_000_000)
.boxed()
.collect(Collectors.toList());
// 순차
long t1 = System.currentTimeMillis();
int max1 = numbers.stream().max(Integer::compareTo).get();
long t2 = System.currentTimeMillis();
System.out.println("순차: " + (t2 - t1) + " ms, max = " + max1);
// 병렬
long t3 = System.currentTimeMillis();
int max2 = numbers.parallelStream().max(Integer::compareTo).get();
long t4 = System.currentTimeMillis();
System.out.println("병렬: " + (t4 - t3) + " ms, max = " + max2);
}
}
무엇을 보게 될까요? 최신 멀티코어 프로세서에서는 보통 병렬 스트림이 더 빠릅니다. 하지만 30_000_000을 1000으로 바꾸면 차이가 거의 없거나, 때로는 병렬이 더 느릴 수도 있습니다!
6. 활용 예: 필터링, 집계, 정렬
필터링과 카운트
List<String> names = Arrays.asList("Anya", "Boris", "Vasya", "Grisha", "Dasha", "Egor", "Zhenya");
long count = names.parallelStream()
.filter(name -> name.length() == 4)
.count();
System.out.println("길이가 4인 이름 수: " + count);
그룹화
List<String> words = Arrays.asList("고양이", "고래", "고양이", "개", "고래", "고양이");
Map<String, Long> freq = words.parallelStream()
.collect(Collectors.groupingBy(
w -> w,
Collectors.counting()
));
System.out.println(freq); // {개=1, 고래=2, 고양이=3}
정렬 (여기서는 병렬화가 항상 이점을 주지는 않습니다!)
List<Integer> bigList = IntStream.rangeClosed(1, 5_000_000)
.boxed()
.collect(Collectors.toList());
long t1 = System.currentTimeMillis();
List<Integer> sorted = bigList.parallelStream()
.sorted()
.collect(Collectors.toList());
long t2 = System.currentTimeMillis();
System.out.println("병렬 정렬: " + (t2 - t1) + " ms");
7. 중요한 포인트와 권장 사항
parallelStream()을 사용할 때
- 컬렉션이 크다(수만 개 이상의 요소).
- 요소별 연산이 ‘무겁다’(복잡한 계산, 파일/네트워크 작업).
- 요소 순서에 의존하지 않는다.
- 부작용이 없다(외부 변수를 변경하지 않는다).
parallelStream()을 사용하지 말아야 할 때
- 컬렉션이 작다.
- 연산이 빠르다.
- 순서를 엄격히 보장해야 한다.
- 공유 변수에 접근한다(스레드 안전한 컬렉션이나 다른 접근 방식을 고려).
사용되는 스레드 수는 어떻게 알 수 있을까?
기본값은 프로세서 코어 개수입니다: Runtime.getRuntime().availableProcessors(). 이 동작은 시스템 속성으로 변경할 수 있습니다:
System.setProperty("java.util.concurrent.ForkJoinPool.common.parallelism", "8");
부작용을 이해할 때만 변경하세요 — 그렇지 않으면 CPU를 ‘가득’ 사용해 오히려 느려질 수 있습니다.
8. 병렬 스트림 사용 시 흔한 실수
오류 №1: forEach 내부의 부작용
많은 분들이 이렇게 생각합니다: “지금 병렬로 리스트를 채우면 되겠지!”
List<Integer> result = new ArrayList<>();
IntStream.range(0, 1_000)
.parallel()
.forEach(result::add); // 위험!
System.out.println(result.size()); // 결과 — 예측 불가!
왜 안 좋을까요? ArrayList는 스레드 안전하지 않아서 여러 스레드가 동시에 추가하면 결과가 예측 불가합니다. 누락, 중복, 예외가 발생할 수 있습니다.
해결책: 스트림의 수집 메서드(collect)를 사용해 안전하게 모으거나, 전용 스레드 안전 컬렉션을 사용하세요.
List<Integer> result = IntStream.range(0, 1_000)
.parallel()
.boxed()
.collect(Collectors.toList());
오류 №2: 작은 작업에서 가속을 기대
병렬화는 공짜가 아닙니다! 컬렉션이 작으면, 스케줄링과 동기화 오버헤드 때문에 병렬 스트림이 더 느릴 수 있습니다.
오류 №3: 순서 깨짐
요소 순서가 중요하다면(예: 파일에 순서대로 기록) 병렬 스트림을 사용하지 마세요 — 순서가 보장되지 않거나 크게 느려질 수 있습니다.
오류 №4: ‘부적절한’ 컬렉션 사용
일부 컬렉션(예: LinkedList, 비표준 구조)은 분할이 어려워 병렬화 효율이 낮습니다.
오류 №5: 결과 수집 시 thread-safety 무시
직접 결과를 모은다면(예: 리스트에 추가), 스레드 안전한 컬렉션(CopyOnWriteArrayList, ConcurrentLinkedQueue)이나 스트림 수집 메서드를 사용하세요.
GO TO FULL VERSION