[Java] Stream API

Stream API 란? 

Java에서 데이터를 저장하기 위해 배열이나 Collection을 사용하는데, 저장된 데이터에 접근하기 위해서 반복문이나 반복자를 사용하여 코드를 작성하는데, 이렇게 작성한 코드는 길고, 가독성이 떨어지며 재사용성도 거의 불가능하다. 

이를 해결하기 위해 Java SE 8부터 Stream API를 도입했다.

Stream API는 데이터를 추상화하여 다루어 저장된 데이터를 여러 방식으로 공통된 방법을 제공한다. 

Stream 

데이터 흐름(연속적 요소)를 의미

병렬 처리를 지원해 성능 최적화가 가능하다. 

Stream API 특징 

내부 반복(internal iteration)을 통해 작업을 수행

재사용이 가능한 컬렉션과는 달리 단 한 번만 사용 가능

원본 데이터 변경 X

지연 연산을 사용해 지연(lazy) 연산을 통해 성능 최적화

parallelStream() 메소드 → 병렬 처리 지원

Stream API 기본 구조 

1️⃣ 데이터 소스(Collection, 배열, 파일 등)에서 스트림 생성
2️⃣ 중간 연산(Intermediate Operation)으로 데이터 변환 및 가공
3️⃣ 최종 연산(Terminal Operation)으로 결과 생성

Stream 동작 방식

1. 스트림의 생성
2. 스트림의 중개 연산 (스트림의 변환)
3. 스트림의 최종 연산 (스트림의 사용)

Stream API 동작 흐름 (출처 : TCPSchool)

Stream 중개 연산 (intermediate operation)

Stream API에 의해 초기 Stream은 중개 연산 통해 또 다른 Stream으로 변환

중개 연산은 연속으로 연결해서 사용 가능 

filter-map 기반의 API 사용하여 지연(lazy) 연산을 통해 최적화 가능 

결과를 즉시 반환하지 않고, 최종 연산이 실행될 때까지 대기

Stream 필터링

filter : 조건에 맞는 요소로 새로운 스트림 반환 

distinct : 중복된 요소가 제거된 새로운 Stream 반환, 내부적으로 Object 클래스의 equals 메서드 사용해 중복 비교

Stream 변환

map : 주어진 함수에 인수로 전달해 반환값들로 이루어진 새로운 Stream 반환

flatMap : 해당 Stream 요소가 배열일 때, 각 배열의 각 요소의 반환값을 하나로 합친 새로운 Stream 얻을 수 있음 

Stream 제한 

limit : 첫 번째 요소부터 전달된 개수만큼의 요소만으로 이루어진 새로운 스트림 반환

skip : 첫 번째 요소부터 전달된 개수만큼의 요소를 제외한 나머지 요소만으로 이루어진 새로운 스트림 반환

Stream 정렬 

sorted : 비교자(comparator)를 이용하여 정렬, 비교자가 없을 때, 기본적으로 사전 편찬 순(natural order)으로 정렬

Stream 연산 결과 확인 

peek : 결과 스트림으로부터 요소를 소모하여 추가로 명시된 동작 수행, 원본 스트림에서 요소를 소모X, 연산과 연산 사이에 결과를 확인할 때 사용, 디버깅 용도로 주로 사용됨 

메소드	연산
Stream<T> filter(Predicate<? super T> predicate)	해당 스트림에서 주어진 조건(predicate)에 맞는 요소만으로 구성된 새로운 스트림을 반환함.
<R> Stream<R> map(Functoin<? super T, ? extends R> mapper)	해당 스트림의 요소들을 주어진 함수에 인수로 전달하여, 그 반환값으로 이루어진 새로운 스트림을 반환함.
<R> Stream<R> flatMap(Functoin<? super T, ? extends Stream<? extends R>> mapper)	해당 스트림의 요소가 배열일 경우, 배열의 각 요소를 주어진 함수에 인수로 전달하여, 그 반환값으로 이루어진 새로운 스트림을 반환함.
Stream<T> distinct()	해당 스트림에서 중복된 요소가 제거된 새로운 스트림을 반환함. 내부적으로 Object 클래스의 equals() 메소드를 사용함.
Stream<T> limit(long maxSize)	해당 스트림에서 전달된 개수만큼의 요소만으로 이루어진 새로운 스트림을 반환함.
Stream<T> peek(Consumer<? super T> action)	결과 스트림으로부터 각 요소를 소모하여 추가로 명시된 동작(action)을 수행하여 새로운 스트림을 생성하여 반환함.
Stream<T> skip(long n)	해당 스트림의 첫 번째 요소부터 전달된 개수만큼의 요소를 제외한 나머지 요소만으로 이루어진 새로운 스트림을 반환함.
Stream<T> sorted() Stream<T> sorted(Comparator<? super T> comparator)	해당 스트림을 주어진 비교자(comparator)를 이용하여 정렬함. 비교자를 전달하지 않으면 영문사전 순(natural order)으로 정렬함.

Stream 최종 연산

최종 연산이 호출되면 Stream은 종료됨 (1회만 사용 가능)

지연(lazy)되었던 모든 중개 연산들이 최종 연산 시에 모두 수행

최종 연산 시에 모든 요소를 소모한 해당 스트림은 더는 사용할 수 없게 된다. 

요소의 출력

forEach : 스트림의 각 요소를 소모하여 명시된 동작 수행, 스트림의 모든 요소를 출력하는 용도로 주로 사용

요소의 소모

reduce : 첫 번째와 두 번째 요소를 가지고 연산을 수행한 뒤, 그 결과와 세 번째 요소를 가지고 또다시 연산을 수행하여 모든 요소를 소모하여 연산을 수행해 결과 반환. 

요소의 검색

findFirst

findAny

요소의 검사

anyMatch : 해당 스트림의 일부 요소가 특정 조건을 만족할 경우에 true를 반환

allMatch : 해당 스트림의 모든 요소가 특정 조건을 만족할 경우에 true를 반환 

noneMatch : 해당 스트림의 모든 요소가 특정 조건을 만족하지 않을 경우에 true를 반환

요소의 통계

count :  해당 스트림의 요소의 총 개수를 long 타입의 값 

min : 스트림의 요소 중에서 가장 작은 값을 가지는 요소를 참조하는 Optional 객체를 얻을 수 있다.

max : 스트림의 요소 중에서 가장 큰 값을 가지는 요소를 참조하는 Optional 객체를 얻을 수 있다.

요소의 연산

sum : 해당 스트림의 모든 요소에 대해 합

average : 해당 스트림의 모든 요소에 대해 평균, Optional 객체를 반환

요소의 수집

collect :

Collectors 객체에 구현된 방법대로 스트림의 요소를 수집

Collectors 클래스에는 미리 정의된 다양한 방법이 클래스 메소드로 정의

그 외

1. 스트림을 배열이나 컬렉션으로 변환 : toArray(), toCollection(), toList(), toSet(), toMap()
2. 요소의 통계와 연산 메소드와 같은 동작을 수행 : counting(), maxBy(), minBy(), summingInt(), averagingInt() 등
3. 요소의 소모와 같은 동작을 수행 : reducing(), joining()
4. 요소의 그룹화와 분할 : groupingBy(), partitioningBy()

 

메소드	설명
void forEach(Consumer<? super T> action)	스트림의 각 요소에 대해 해당 요소를 소모하여 명시된 동작을 수행함.
Optional<T> reduce(BinaryOperator<T> accumulator) T reduce(T identity, BinaryOperator<T> accumulator)	처음 두 요소를 가지고 연산을 수행한 뒤, 그 결과와 다음 요소를 가지고 또다시 연산을 수행함. 이런 식으로 해당 스트림의 모든 요소를 소모하여 연산을 수행하고, 그 결과를 반환함.
Optional<T> findFirst() Optional<T> findAny()	해당 스트림에서 첫 번째 요소를 참조하는 Optional 객체를 반환함. (findAny() 메소드는 병렬 스트림일 때 사용함)
boolean anyMatch(Predicate<? super T> predicate)	해당 스트림의 일부 요소가 특정 조건을 만족할 경우에 true를 반환함.
boolean allMatch(Predicate<? super T> predicate)	해당 스트림의 모든 요소가 특정 조건을 만족할 경우에 true를 반환함.
boolean noneMatch(Predicate<? super T> predicate)	해당 스트림의 모든 요소가 특정 조건을 만족하지 않을 경우에 true를 반환함.
long count()	해당 스트림의 요소의 개수를 반환함.
Optional<T> max(Comparator<? super T> comparator)	해당 스트림의 요소 중에서 가장 큰 값을 가지는 요소를 참조하는 Optional 객체를 반환함.
Optional<T> min(Comparator<? super T> comparator)	해당 스트림의 요소 중에서 가장 작은 값을 가지는 요소를 참조하는 Optional 객체를 반환함.
T sum()	해당 스트림의 모든 요소에 대해 합을 구하여 반환함.
Optional<T> average()	해당 스트림의 모든 요소에 대해 평균값을 구하여 반환함.
<R,A> R collect(Collector<? super T,A,R> collector)	인수로 전달되는 Collectors 객체에 구현된 방법대로 스트림의 요소를 수집함.

병렬 Stream

Parallel Processing 병렬 처리를 지원해 대용량 데이터를 빠르게 처리 가능하다.

멀티코어 CPU를 활용해 성능이 향상되지만, 정렬이 필요한 경우, 병렬 처리는 결과가 예상 결과와 다를 수 있다. 

 

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

class Employee {

    private String name;
    private int salary;
    private String department;

    public Employee(String name, int salary, String department) {
    
        this.name = name;
        this.salary = salary;
        this.department = department;
    }

    public String getName() {
        return name;
    }

    public int getSalary() {
        return salary;
    }

    public String getDepartment() {
        return department;
    }

    @Override
    public String toString() {
        return name + " - " + salary + " (" + department + ")";
    }
}

public class StreamMixedExample {
    public static void main(String[] args) {

        // 직원 목록 생성
        List<Employee> employees = Arrays.asList(
                new Employee("Alice", 6000, "HR"),
                new Employee("Bob", 4000, "IT"),
                new Employee("Charlie", 5500, "Finance"),
                new Employee("David", 7000, "IT"),
                new Employee("Eve", 4800, "HR")
        );

        // Stream API 혼합 사용
        // filter : 필터링
        // map : 변환
        // sorted : 정렬
        // parallelStream : 병렬
        // toList : 리스트 변환 
        List<String> highSalaryNames = employees.parallelStream() // 병렬 처리
                .filter(emp -> emp.getSalary() >= 5000)  // 급여 5000 이상 필터링
                .map(emp -> emp.getName().toUpperCase()) // 이름을 대문자로 변환
                .sorted() // 이름순 정렬
                .collect(Collectors.toList()); // 리스트로 변환

        // 결과 출력
        System.out.println("급여 5000 이상 직원 이름 목록: " + highSalaryNames);

        // 급여 총합 계산 (reduce 사용)
        // map 
        // reduce
        int totalSalary = employees.stream()
                .map(Employee::getSalary)
                .reduce(0, Integer::sum);

        System.out.println("총 급여 합계: " + totalSalary);

        // 부서별 그룹화
        // groupingBy
        Map<String, List<Employee>> departmentGroup = employees.stream()
                .collect(Collectors.groupingBy(Employee::getDepartment));

        System.out.println("부서별 직원 목록:");
        // forEach
        departmentGroup.forEach((dept, empList) -> {
            System.out.println(dept + " -> " + empList);
        });
    }
}