Java – Stream 스트림의 고급 사용법

Stream의 고급 기능

Java 8에서 도입된 Stream API는 데이터 처리 능력이 매우 뛰어납니다. 중간 연산, 최종 연산, 수집기(collector), 병렬 스트림 등의 고급 기능을 제공하여 매우 매력적인 도구입니다.

 

 

1. 중간 연산과 최종 연산

Java Stream의 중간 연산(예: filter, map)은 지연(lazy) 실행됩니다. 즉, 새로운 스트림을 반환하며 처리 파이프라인을 구성하지만 즉시 실행되지는 않습니다.

최종 연산(예: collect, forEach)을 호출해야만 실제 처리가 수행되며 결과가 생성됩니다. 중간 연산은 체이닝할 수 있고, 최종 연산은 스트림을 소비하여 다시 사용할 수 없습니다.

예시:

stream.map(...).filter(...).collect(...)

• map, filter → 중간 연산
• collect → 최종 연산

import java.util.Arrays;
import java.util.List;
import java.util.stream.Collectors;

public class StreamOperations {
    public static void main(String[] args) {
        List<String> names = Arrays.asList("Alice", "Bob", "Charlie", "David", "Eve", "Frank");

        // 체이닝된 스트림 연산: 필터링 -> 매핑 -> 정렬 -> 수집
        List<String> processedNames = names.stream()
                .filter(name -> name.length() > 3)        // 중간 연산: 길이가 3 초과인 이름 필터링
                .map(String::toUpperCase)                // 중간 연산: 대문자로 변환
                .sorted((a, b) -> b.compareTo(a))         // 중간 연산: 내림차순 정렬
                .collect(Collectors.toList());            // 최종 연산: 리스트로 수집

        System.out.println("Processed names: " + processedNames);

        // 최종 연산: forEach
        System.out.print("Printing names: ");
        names.stream()
             .limit(4)                                    // 중간 연산: 앞의 4개 요소 제한
             .forEach(name -> System.out.print(name + " ")); // 최종 연산: 각 요소 출력

        // 최종 연산: reduce
        String concatenated = names.stream()
                .reduce("", (a, b) -> a + "|" + b);       // 문자열을 '|' 기호로 연결
        System.out.println("\nConcatenated: " + concatenated);
    }
}

---

2. 고급 수집기 (Collectors)

Collectors 클래스는 분류(groupingBy), 분할(partitioningBy), 문자열 연결(joining) 등의 고급 수집 기능을 제공합니다.

예:

• groupingBy: 특정 속성 기준으로 분류
• partitioningBy: 조건(true/false)로 두 그룹 분할
• joining: 문자열 연결
• toMap: 스트림을 key-value 형태로 변환

또한, 하위 수집기 조합을 통해 다단계 그룹화 및 통계 처리가 가능합니다.

 

import java.util.*;
import java.util.stream.Collectors;

public class AdvancedCollectors {
    public static void main(String[] args) {
        List<Person> people = Arrays.asList(
                new Person("Alice", 25, "New York"),
                new Person("Bob", 30, "London"),
                new Person("Charlie", 20, "New York"),
                new Person("David", 35, "London"),
                new Person("Eve", 28, "Paris")
        );

        // 도시별로 그룹화
        Map<String, List<Person>> peopleByCity = people.stream()
                .collect(Collectors.groupingBy(Person::getCity));
        System.out.println("People by city: " + peopleByCity);

        // 도시별로 그룹화하고 각 그룹의 평균 나이 계산
        Map<String, Double> avgAgeByCity = people.stream()
                .collect(Collectors.groupingBy(
                        Person::getCity,
                        Collectors.averagingInt(Person::getAge)
                ));
        System.out.println("Average age by city: " + avgAgeByCity);

        // 파티셔닝: 나이가 30 이상인지 여부에 따라 두 그룹으로 나누기
        Map<Boolean, List<Person>> partitioned = people.stream()
                .collect(Collectors.partitioningBy(p -> p.getAge() >= 30));
        System.out.println("Partitioned by age >=30: " + partitioned);

        // 이름 문자열 연결
        String allNames = people.stream()
                .map(Person::getName)
                .collect(Collectors.joining(", ", "[", "]"));
        System.out.println("All names: " + allNames);
    }
}

class Person {
    private String name;
    private int age;
    private String city;

    // 생성자, getter, setter 생략...
}

---

3. 병렬 스트림과 성능 고려사항

parallelStream()은 멀티코어 프로세서를 활용하여 스트림 요소를 병렬로 처리합니다. 하지만 다음에 유의해야 합니다:

• 스레드 안전해야 하며,
• 공유 가변 상태는 피해야 합니다.
• 작업 오버헤드 및 데이터 특성에 따라 병렬이 더 느릴 수 있습니다.

특히 병렬 스트림의 성능은 Spliterator의 분할 효율성과도 관련이 있으므로 성능 테스트가 중요합니다.

 

import java.util.Arrays;
import java.util.List;
import java.util.concurrent.ForkJoinPool;

public class ParallelStreams {
    public static void main(String[] args) {
        List<Integer> numbers = Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);

        // 순차 스트림
        long start = System.currentTimeMillis();
        int sumSequential = numbers.stream()
                .mapToInt(ParallelStreams::slowSquare)  // 시간 소요가 큰 작업을 시뮬레이션
                .sum();
        long duration = System.currentTimeMillis() - start;
        System.out.println("순차 합계: " + sumSequential + " (걸린 시간: " + duration + "ms)");

        // 병렬 스트림
        start = System.currentTimeMillis();
        int sumParallel = numbers.parallelStream()
                .mapToInt(ParallelStreams::slowSquare)  // 시간 소요가 큰 작업을 시뮬레이션
                .sum();
        duration = System.currentTimeMillis() - start;
        System.out.println("병렬 합계: " + sumParallel + " (걸린 시간: " + duration + "ms)");

        // 사용자 정의 ForkJoinPool 사용
        ForkJoinPool customPool = new ForkJoinPool(4);  // 병렬 스레드 수 지정
        start = System.currentTimeMillis();
        int sumCustomParallel = customPool.submit(() -> 
            numbers.parallelStream()
                   .mapToInt(ParallelStreams::slowSquare)
                   .sum()
        ).get();  // submit은 Future를 반환하므로 get()으로 결과 받음
        duration = System.currentTimeMillis() - start;
        System.out.println("커스텀 풀 합계: " + sumCustomParallel + " (걸린 시간: " + duration + "ms)");
    }

    // 시간 소요가 큰 작업을 시뮬레이션하는 메서드
    private static int slowSquare(int n) {
        try {
            Thread.sleep(100);  // 100ms 지연
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return n * n;
    }
}

---

4. 기본형 스트림과 flatMap

• IntStream, DoubleStream 등 기본형 스트림은 오토박싱 비용을 줄일 수 있습니다.
• flatMap은 중첩된 구조(예: Stream<List<T>>)를 단일 스트림으로 평탄화합니다.

예시:

stream.flatMap(list -> list.stream())

• 색상과 사이즈를 조합해 데카르트 곱(조합) 생성도 flatMap으로 구현 가능합니다.

import java.util.Arrays;
import java.util.List;
import java.util.stream.IntStream;
import java.util.stream.Stream;
import java.util.stream.Collectors;

public class PrimitiveAndFlatMap {
    public static void main(String[] args) {
        // 원시 타입 스트림 (박싱 비용을 피할 수 있음)
        IntStream.rangeClosed(1, 5)  // 1부터 5까지의 IntStream 생성
                .map(n -> n * n)     // 각 숫자를 제곱
                .average()           // 평균 계산
                .ifPresent(avg -> System.out.println("평균: " + avg));

        // flatMap 예제: 여러 스트림을 하나로 평탄화(flatten)
        List<List<String>> listOfLists = Arrays.asList(
                Arrays.asList("a", "b"),
                Arrays.asList("c", "d", "e"),
                Arrays.asList("f")
        );

        // flatMap을 사용해 중첩된 리스트를 평탄화
        List<String> flatList = listOfLists.stream()
                .flatMap(List::stream)  // 각 List<String>을 Stream<String>으로 변환하여 연결
                .collect(Collectors.toList());
        System.out.println("평탄화된 리스트: " + flatList);

        // 더 복잡한 flatMap 예제: 데카르트 곱 생성
        List<String> colors = Arrays.asList("Red", "Green", "Blue");
        List<String> sizes = Arrays.asList("S", "M", "L", "XL");

        List<String> combinations = colors.stream()
                .flatMap(color -> sizes.stream()
                        .map(size -> color + "-" + size))  // 각 색상과 크기의 조합 생성
                .collect(Collectors.toList());
        System.out.println("색상-크기 조합: " + combinations);
    }
}

---

5. 사용자 정의 수집기

Collector<T, A, R> 인터페이스를 구현하여 수집 로직을 사용자 정의할 수 있습니다.

구성 요소:

• supplier(): 결과 컨테이너 제공
• accumulator(): 요소 누적 처리
• combiner(): 병렬 처리시 결과 병합
• finisher(): 최종 결과 반환
• characteristics(): 수집기 특성 정의 (예: CONCURRENT 등)

예:

• 접두어, 구분자, 접미어를 포함한 문자열 연결기
• 단어 통계 수집기 (최단/최장 단어, 평균 길이 등)

import java.util.*;
import java.util.function.*;
import java.util.stream.Collector;

public class CustomCollector {
    public static void main(String[] args) {
        List<String> words = Arrays.asList("hello", "world", "java", "stream", "collector");
        
        // 사용자 정의 수집기로 문자열 연결, 접두사와 접미사 추가
        String concatenated = words.stream()
                .collect(new StringConcatenator(", ", "[", "]"));
        System.out.println("Concatenated: " + concatenated);

        // 더 복잡한 사용자 정의 수집기: 통계 정보 수집
        Stats stats = words.stream()
                .collect(Collector.of(
                        Stats::new,            // 공급자 (Supplier)
                        Stats::accumulate,     // 누적기 (Accumulator)
                        Stats::combine,        // 병합기 (병렬 처리용) (Combiner)
                        Stats::finalize        // 완료자 (Finisher)
                ));
        System.out.println("Stats: " + stats);
    }
}

// 사용자 정의 문자열 연결 수집기
class StringConcatenator implements Collector<String, StringBuilder, String> {
    private final String delimiter;  // 구분자
    private final String prefix;     // 접두사
    private final String suffix;     // 접미사

    public StringConcatenator(String delimiter, String prefix, String suffix) {
        this.delimiter = delimiter;
        this.prefix = prefix;
        this.suffix = suffix;
    }

    @Override
    public Supplier<StringBuilder> supplier() {
        return StringBuilder::new;
    }

    @Override
    public BiConsumer<StringBuilder, String> accumulator() {
        return (sb, str) -> {
            if (sb.length() > 0) {
                sb.append(delimiter);
            }
            sb.append(str);
        };
    }

    @Override
    public BinaryOperator<StringBuilder> combiner() {
        return (sb1, sb2) -> {
            if (sb1.length() > 0 && sb2.length() > 0) {
                sb1.append(delimiter);
            }
            return sb1.append(sb2);
        };
    }

    @Override
    public Function<StringBuilder, String> finisher() {
        return sb -> prefix + sb.toString() + suffix;
    }

    @Override
    public Set<Characteristics> characteristics() {
        return Collections.emptySet();
    }
}

// 통계 정보를 수집하는 클래스
class Stats {
    private int count;           // 단어 개수
    private int totalLength;     // 전체 길이 합계
    private String shortest;     // 가장 짧은 단어
    private String longest;      // 가장 긴 단어

    public Stats() {
        shortest = "";
        longest = "";
    }

    // 누적기: 단어 정보를 누적
    public void accumulate(String word) {
        if (count == 0) {
            shortest = longest = word;
        } else {
            if (word.length() < shortest.length()) shortest = word;
            if (word.length() > longest.length()) longest = word;
        }
        count++;
        totalLength += word.length();
    }

    // 병합기: 병렬 처리 시 두 Stats 객체를 합침
    public Stats combine(Stats other) {
        if (this.count == 0) return other;
        if (other.count == 0) return this;

        this.count += other.count;
        this.totalLength += other.totalLength;
        this.shortest = this.shortest.length() <= other.shortest.length() 
                ? this.shortest : other.shortest;
        this.longest = this.longest.length() >= other.longest.length() 
                ? this.longest : other.longest;
        return this;
    }

    // 완료자: 최종 결과 반환 (여기서는 그대로 반환)
    public Stats finalize() {
        return this;
    }

    @Override
    public String toString() {
        return String.format(
                "count=%d, avgLen=%.2f, shortest='%s', longest='%s'",
                count, (double)totalLength/count, shortest, longest
        );
    }
}

---

요약

• 중간 연산은 지연 실행되며, 최종 연산이 있어야 스트림이 실행됩니다.
• Collectors.groupingBy 등의 고급 수집기를 사용하면 복잡한 그룹화나 통계를 쉽게 수행할 수 있습니다.
• 병렬 스트림은 멀티코어 CPU 활용에 유리하지만 상황에 따라 성능 저하가 있을 수 있으므로 주의가 필요합니다.
• flatMap은 중첩된 구조를 평탄화할 때 유용하며, 기본형 스트림은 성능 향상에 도움이 됩니다.
• 사용자 정의 수집기는 복잡한 로직이 필요한 경우에 강력하지만, 스레드 안전과 동시성에 주의해야 합니다.