[모던 자바 인 액션] chapter 6. 스트림으로 데이터 수집 (2)

 

[목차]
chapter 1. 자바 8, 9, 10, 11 : 무슨 일이 일어나고 있는가?
chapter 2. 동작 파라미터화 코드 전달하기
chapter 3. 람다 표현식(1)
chapter 3. 람다 표현식(2)
chapter 4. 스트림 소개
chapter 5. 스트림 활용
chapter 6. 스트림으로 데이터 수집(1)
chapter 6. 스트림으로 데이터 수집(2)

 

분할

분할 함수(partitioning function)라 불리는 프레디케이트를 분류 함수로 사용하는 특수한 그룹화 기능

• 분할 함수는 불리언을 반환하므로 맵의 키 형식은 Boolean
• 그룹화 맵은 최대 (참 or 거짓을 갖기 때문에) 2개의 그룹으로 분류됨.

// partitioningBy를 이용해 채식 요리와 채식이 아닌 요리 분리
Map<**Boolean**, List<Dish>> partitionedMenu = 
                menu.stream().collect(**partitioningBy**(Dish::isVegetarian));

List<Dish> vegetarianDishes = partitionedMenu.get(true);

// filter를 이용해 동일한 결과를 얻을 수 있음
List<Dish> vegetarianDishes = menu.stream().filter(Dish::isVegetarian).collect(toList());

분할의 장점

• 분할의 장점은 분할 함수가 반환하는 참, 거짓 두 가지 요소의 스트림 리스트를 모두 유지한다는 것!
• 컬렉터를 두 번째 인수로 전달할 수 있는 오버로드된 버전의 partitioningBy도 존재한다. → 다수준의 특수한 그룹화 결과를 반환할 수 있음

Map<Boolean, **Map<Dish.Type, List<Dish>>**> vegetarianDishesByType =
                menu.stream().collect(
                        **partitioningBy(Dish::isVegetarian, groupingBy(Dish::getType))**);

• partitioningBy는 내부적으로 특수한 맵과 두 개의 필드로 구현되었다. → Partition이라는 내부 클래스가 존재함

private static final class Partition<T>
            extends AbstractMap<Boolean, T>
            implements Map<Boolean, T> {
        final T forTrue;
        final T forFalse;

        Partition(T forTrue, T forFalse) {
            this.forTrue = forTrue;
            this.forFalse = forFalse;
        }

        @Override
        public Set<Map.Entry<Boolean, T>> entrySet() {
            return new AbstractSet<Map.Entry<Boolean, T>>() {
                @Override
                public Iterator<Map.Entry<Boolean, T>> iterator() {
                    Map.Entry<Boolean, T> falseEntry = new SimpleImmutableEntry<>(false, forFalse);
                    Map.Entry<Boolean, T> trueEntry = new SimpleImmutableEntry<>(true, forTrue);
                    return Arrays.asList(falseEntry, trueEntry).iterator();
                }

                @Override
                public int size() {
                    return 2;
                }
            };
        }
    }

• 다수준으로 분할하는 기법도 존재한다.

// 채식 요리와 아닌 것으로 분할하고, 그 안에서도 500칼로리가 넘는 것과 넘지 않는 것으로 분할
menu.stream().collect(partitioningBy(Dish::isVegetarian, 
			                partitioningBy(d -> d.getCalories() > 500)));

// partitioningBy는 프레디케이트(불리언 반환)를 요구하므로 아래 코드는 컴파일되지 않는다.
// error -> reason: Incompatible types: Type is not convertible to boolean
menu.stream().collect(partitioningBy(Dish::isVegetarian,
			                partitioningBy(Dish::getType));

 

숫자를 소수와 비소수로 분할하기

정수 n을 인수로 받아서 2에서 n까지의 자연수를 소수와 비소수로 나누는 프로그램을 구현하자.

public boolean isPrime(int candidate) {
    int candidateRoot = (int) Math.sqrt((double) candidate);
    return IntStream.rangeClosed(2, candidateRoot)
                    .noneMatch(i -> candidate % i == 0);
}

public Map<Boolean, List<Integer>> partitionPrimes(int n) {
    return IntStream.rangeClosed(2, n).boxed()
                    .collect(partitioningBy(this::isPrime));
}

 

Collector 인터페이스

Collector 인터페이스는 리듀싱 연산(컬렉터)을 어떻게 구현할지 제공하는 메서드 집합으로 구성된다.

자주 사용되면서도 구현하기 쉬운 컬렉터인 toList()가 어떻게 구현되었는지 살펴보면서 Collector 인터페이스를 이해해보자.

// Collector 인터페이스의 시그니처
public interface Collector<**T, A, R**> {
		...
}

• T는 수집될 스트림 항목의 제네릭 형식이다.
• A는 누적자, 즉 수집 과정에서 중간 결과를 누적하는 객체의 형식이다.
• R은 수집 연산 결과 객체의 형식(대개 컬렉션 형식)이다.

Collector 인터페이스의 메서드 살펴보기

Collector 인터페이스에는 다섯 개의 메서드가 정의되어 있다.

• 먼저 살펴볼 4개의 메서드는 collect 메서드에서 실행하는 함수를 반환한다.
• 마지막 메서드는 힌트 특성 집합을 제공 → collect 메서드가 어떤 최적화를 이용해서 리듀싱 연산을 수행할 것인지 결정하도록 돕는 역할

supplier() : 새로운 결과 컨테이너 만들기

수집 과정에서 빈 누적자 인스턴스를 만드는 함수로, 파라미터가 없는 함수다.

/**
* A function that creates and returns a new mutable result container.
* @return a function which returns a new, mutable result container
*/
Supplier<A> supplier();

accumulator() : 결과 컨테이너에 요소 추가하기

리듀싱 연산을 수행하는 함수를 반환한다. 스트림에서 n번째 요소를 탐색할 때 두 인수, 즉 누적자(n-1까지의 항목을 수집한 상태)와 n번째 요소를 함수에 적용한다.

/**
* A function that folds a value into a mutable result container.
* @return a function which folds a value into a mutable result container
*/
BiConsumer<A, T> accumulator();

finisher() : 최종 변환값을 결과 컨테이너로 적용하기

스트림 탐색을 끝내고 누적자 객체를 최종 결과로 변환하면서 누적 과정을 끝낼 때 호출할 함수를 반환한다.

/**
* Perform the final transformation from the intermediate accumulation type
* {@code A} to the final result type {@code R}.
*
* <p>If the characteristic {@code IDENTITY_TRANSFORM} is
* set, this function may be presumed to be an identity transform with an
* unchecked cast from {@code A} to {@code R}.
*
* @return a function which transforms the intermediate result to the final
* result
*/
Function<A, R> finisher();

combiner() : 두 결과 컨테이너 병합

스트림의 서로 다른 서브파트를 병렬로 처리할 때 누적자가 이 결과를 어떻게 처리할지 정의한다.

/**
* A function that accepts two partial results and merges them.  The
* combiner function may fold state from one argument into the other and
* return that, or may return a new result container.
* @return a function which combines two partial results into a combined
* result
*/
BinaryOperator<A> combiner();

스트림의 병렬 리듀싱 수행 과정

• 스트림으로 분할해야 하는지 정의하는 조건이 거짓으로 바뀌기 전까지 원래 스트림을 재귀적으로 분할한다. (분산된 작업의 크기가 너무 작아지면 병렬 수행 속도는 순차 수행보다 느려지기 때문에 주의. 프로세싱 코어의 개수를 초과하는 병렬 작업은 비효율적임)
• 모든 서브스트림의 각 요소에 리듀싱 연산을 순차적으로 적용해서 서브스트림을 병렬로 처리할 수 있다.
• 마지막에는 컬렉터의 combiner 메서드가 반환하는 함수로 모든 부분 결과를 쌍으로 합친다.

characteristics()

스트림을 병렬로 리듀스할 것인지 그리고 병렬로 리듀스한다면 어떤 최적화를 선택해야 할지 힌트를 제공

/**
* Returns a {@code Set} of {@code Collector.Characteristics} indicating
* the characteristics of this Collector.  This set should be immutable.
* @return an immutable set of collector characteristics
*/
Set<Characteristics> characteristics();

Characteristics는 다음 세 항목을 포함하는 열거형(Enum)이다.

• UNORDERED : 리듀싱 결과는 스트림 요소의 방문 순서나 누적 순서에 영향을 받지 않는다.
• CONCURRENT : 다중 스레드에서 accumulator 함수를 동시에 호출할 수 있으며 이 컬렉터는 스트림의 병렬 리듀싱을 수행할 수 있다. 컬렉터의 플래그에 UNORDERED를 함께 설정하지 않았다면 데이터 소스가 정렬되어 있지 않은(즉, 집합처럼 요소의 순서가 무의미한) 상황에서만 병렬 리듀싱을 수행할 수 있다.
• IDENTITY_FINISH : fisnisher 메서드가 반환한는 함수는 단순히 identity를 적용할 뿐이므로 이를 생략할 수 있다. 따라서 리듀싱 과정의 최종 결과로 누적자 객체를 바로 사용할 수 있다. 또한 누적자 A를 결과 R로 안전하게 형변환할 수 있다.

→ characteristics()는 Characteristic이라는 enum을 Set 자료구조로 갖는 객체를 반환한다.

toList 구현 살펴보기

// toList() 구현 코드
public static <T>
    Collector<T, ?, List<T>> toList() {
        return new CollectorImpl<>((Supplier<List<T>>) ArrayList::new, // Supplier
																		List::add, // Accumulator
                                   (left, right) -> { left.addAll(right); return left; }, // combiner
	                                  CH_ID // Characteristics
																	);
    }

// CollectorImpl 생성자
CollectorImpl(Supplier<A> supplier,
                      BiConsumer<A, T> accumulator,
                      BinaryOperator<A> combiner,
                      Set<Characteristics> characteristics) {
            this(supplier, accumulator, combiner, castingIdentity(), characteristics);
        }

• toList의 Supplier에서는 빈 ArrayList 인스턴스를 생성한다.
• toList의 Accumulator에서는 누적자(list)와 그 다음 요소(T)에 대해 수행할 리듀싱 연산으로 리스트 타입의 누적자에 요소를 add한다.
• toList의 Combiner에서는 병렬 처리 시 생긴 서브파트의 결과들에 대한 처리로, addAll 메서드를 통해 두 리스트의 요소를 합친다.
• toList의 Finisher는 castingIdentity()의 반환값을 사용한다.
• private static <I, R> Function<I, R> castingIdentity() { return i -> (R) i; }
• toList의 Characteristics는 CH_ID라는 상수(static final)값을 사용한다.→ 복잡해보이지만 결국 IDENTITY_FINISH가 들어있는 Set이다.
• static final Set<Collector.Characteristics> CH_ID = Collections.unmodifiableSet(EnumSet.of(Collector.Characteristics.IDENTITY_FINISH));

 

응용하기

class ToListCollector<T> implements Collector<T, List<T>, List<T>> {

    @Override
    public Supplier<List<T>> supplier() {
        return ArrayList::new;
    }

    @Override
    public BiConsumer<List<T>, T> accumulator() {
        return List::add;
    }

    @Override
    public BinaryOperator<List<T>> combiner() {
        return (left, right) -> {
            left.addAll(right);
            return left;
        };
    }

    @Override
    public Function<List<T>, List<T>> finisher() {
        return Function.identity();
    }

    @Override
    public Set<Characteristics> characteristics() {
        return Collections.unmodifiableSet(EnumSet.of(
	               Collector.Characteristics.CONCURRENT,
                 Collector.Characteristics.IDENTITY_FINISH));
        }
  }

// 다음과 같이 사용할 수 있다.
List<Dish> dishes = menu.stream()
                        .collect(new ToListCollector<>());

컬렉터 구현을 만들지 않고도 커스텀 수집 수행하기

IDENTITY_FINISH 수집 연산에서는 Collector 인터페이스를 완전히 새로 구현하지 않고도 같은 결과를 얻을 수 있다.

Stream은 세 함수(발행, 누적, 합침)를 인수로 받는 collect 메서드를 오버로드하며 각각의 메서드는 Collector 인터페이스의 메서드가 반환하는 함수와 같은 기능을 수행한다.

List<Dish> dishes = menu.stream().collect(
                ArrayList::new, // 발행 (supplier)
                List::add,  // 누적 (accumulator)
                List::addAll);  // 합침 (combiner)

→ 이전 코드에 비해 좀 더 간결하고 축약되어 있지만 가독성은 떨어진다. 또한 Characteristics를 전달할 수 없다.(IDENTITY_FINISH, CONCURRENT지만 UNORDERED는 아닌 컬렉터로만 동작)

 

커스텀 컬렉터를 구현해서 성능 개선하기

앞서 작성한 소수, 비소수 분할 코드를 커스텀 컬렉터를 이용해서 성능을 개선해보자.

public Map<Boolean, List<Integer>> partitionPrimes(int n) {
        return IntStream.rangeClosed(2, n).boxed()
                        .collect(partitioningBy(candidate -> isPrime(candidate)));
}
    
public boolean isPrime(int candidate) {
        int candidateRoot = (int) Math.sqrt((double) candidate);
        return IntStream.rangeClosed(2, candidateRoot)
                        .noneMatch(i -> candidate % i == 0);
}

소수로 나누어 떨어지는지 확인해서 대상의 범위를 좁힌다. 제수(나누는 수)가 소수가 아니면 소용없기 때문에 제수를 현재 숫자 이하에서 발견한 소수로 제한할 수 있다.

주어진 숫자가 소수인지 판단하기 위해 지금까지 발견한 소수 리스트에 접근해야 한다.

→ 기존의 컬렉터로는 수집 과정에서 부분 결과에 접근할 수 없기 때문에 커스텀 클래스로 문제를 해결한다.

중간 결과 리스트가 있다면 isPrime 메서드로 중간 결과 리스트를 전달하도록 구현한다.

public boolean isPrime(List<Integer> primes, int candidate) {
        return primes.stream().noneMatch(i -> candidate % i == 0);
}

대상 숫자의 제곱근보다 작은 소수만 사용하도록 코드를 최적화해야한다. 다음 소수가 대상의 루트보다 크면 소수로 나누는 검사를 중단하는 방식으로 성능을 개선한다.

→ 스트림 API에는 이런 기능을 제공하는 메소드가 없다.

리스트와 프레디케이트를 인수로 받아 프레디케이트를 만족하는 긴 요소로 이루어진 리스트를 반환하는 takeWhile이라는 메소드를 구현한다. (takeWhile은 java9버전에서 지원하므로 직접 구현해야한다.)

// java 8
public static <A> List<A> takeWhile(List<A> list, Predicate<A> p) {
        int i = 0;
        for (A item : list) {
            if (!p.test(item)) {
                return list.subList(0, i);
            }
            i++;
        }
        return list;
    }

public boolean isPrime(List<Integer> primes, int candidate) {
        int candidateRoot = (int) Math.sqrt((double) candidate);
        return takeWhile(primes, i -> i <= candidateRoot)
                .stream()
                .noneMatch(p -> candidate % p == 0);
    }

// java 9
public boolean isPrime(List<Integer> primes, int candidate) {
        int candidateRoot = (int) Math.sqrt((double) candidate);
        return primes.stream()
                     .takeWhile(i -> i <= candidateRoot)
                     .noneMatch(i -> candidate % i == 0);
}

1단계: Collector 클래스 시그니처 정의

Collector 인터페이스 정의를 참고해서 클래스 시그니처를 만들자.

public interface Collector<T, A, R>

• T : 스트림 요소의 형식
• A : 중간 결과를 누적하는 객체의 형식
• R : collect 연산의 최종 결과 형식

우리가 구현해야할 컬렉터는 Map<Boolean, List<Integer>>로, 참과 거짓을 키로 갖소 소수와 소수가 아닌 수를 값으로 가져야 한다.

public class PrimeNumbersCollector 
					implements Collector<Integer, // 스트림 요소의 형식
															 Map<Boolean, List<Integer>>, // 누적자 형식
															 Map<Boolean, List<Integer>>> // 수집 연산의 결과 형식

2단계: 리듀싱 연산 구현

Collector 인터페이스에 선언된 다섯 메서드를 구현해야 한다.

supplier()

		@Override
    public Supplier<Map<Boolean, List<Integer>>> supplier() {
        return () -> new HashMap<Boolean, List<Integer>>() {{
            put(true, new ArrayList<Integer>());
            put(false, new ArrayList<Integer>());
        }};
    }

→ 누적자로 사용할 맵을 만들면서 true, false 키와 빈 리스트로 초기화

accumulator()

		@Override
    public BiConsumer<Map<Boolean, List<Integer>>, Integer> accumulator() {
        return (Map<Boolean, List<Integer>> acc, Integer candidate) -> {
						acc.get(isPrime(acc.get(true), candidate)) // isPrime 결과에 따라 소수 리스트와 비소수 리스트를 만든다.
               .add(candidate); // candidate를 알맞은 리스트에 추가한다.
        };
    }

→ 스트림의 요소를 어떻게 수집할지 결정. 지금까지 발견한 소수 리스트와 candidate를 인수로 isPrime을 호출. 결과에 따라 알맞는 리스트로 candidate를 추가.

3단계: 병렬 실행할 수 있는 컬렉터 만들기(가능하다면)

병렬 수집 과정에서 두 부분 누적자를 합칠 수 있는 메서드를 만든다. (combiner)

		@Override
    public BinaryOperator<Map<Boolean, List<Integer>>> combiner() {
        return (Map<Boolean, List<Integer>> map1, Map<Boolean, List<Integer>> map2) -> {
            map1.get(true).addAll(map2.get(true));
            map1.get(false).addAll(map2.get(false));
            return map1;
        };
    }

→ 사실 해당 기능은 알고리즘 자체가 순차적이어서 컬렉터를 실제 병렬로 사용할 수 없다. 사용될 일이 없기 때문에 학습 목적으로 구현한 것임. (빈 구현으로 남겨두거나, UnsupportedOperationException을 던지도록 구현)

4단계: finisher 메서드와 컬렉터의 characteristics 메서드

accumulator의 형식은 컬렉터 결과 형식과 같으므로 변환 과정이 필요 없다. 따라서 항등 함수 identity를 반환하도록 finisher 메서드를 구현한다.

		@Override
    public Function<Map<Boolean, List<Integer>>, Map<Boolean, List<Integer>>> finisher() {
        return Function.identity();
    }

또한, 이 커스텀 컬렉터는 CONCURRENT도 아니고 UNORDERED도 아니지만 IDENTITY_FINISH이므로 다음처럼 characteristics 메서드를 구현할 수 있다.

		@Override
    public Set<Characteristics> characteristics() {
        return Collections.unmodifiableSet(EnumSet.of(Characteristics.IDENTITY_FINISH));
    }