모으기(aggregation) 연산

List 콜렉션으로 모으기(collectList)

collectList()의 리턴 타입은 Mono<List<T>>이므로 Mono를 구독해서 값을 사용하면 된다.

Mono<List<Integer>> mono = someFlux.collectList();

mono.subscribe(lst -> System.out.println(lst));

 

Map 콜렉션으로 모으기(collectMap)

keyExtractor : 데이터에서 맵의 키를 제공하는 함수

valueExtractor : 데이터에서 맵의 값을 제공하는 함수

mapSupplier : 사용할 Map 객체를 제공(mapSupplier가 없는 메서드는 기본으로 HashMap 사용)

Mono<List<Integer>> mono = someFlux.collectList();

mono.subscribe(lst -> System.out.println(lst));

collectMap은 중복된 키가 존재하면 마지막 데이터와 관련된 값이 사용된다.

Flux.just(1, 2, 3, 4)
        .collectMap(x -> x % 2)
        .subscribe(System.out::println); // {0=4, 1=3}

 

Map의 값을 콜렉션으로 모으기(collectMultiMap())

같은 키를 가진 데이터를 List로 갖는 Map을 생성할 수 있다.

Mono<Map<Integer, Collection<Integer>>> oddEvenList =
        Flux.just(1, 2, 3, 4).collectMultimap(x -> x % 2);
oddEvenList.subscribe(System.out::println); // {0=[2, 4], 1=[1, 3]}

 

개수 새기(count)

Mono<Long> countMono = Flux.just(1, 2, 3, 4).count();

countMono.subscribe(System.out::println);

 

누적(reduce)

각 값에 연산을 누적해서 결과를 생성

Flux의 데이터를 이용해서 단일 값을 생성하는 범용 기능이라고 보면 된다.

 

Mono<T> reduce(BiFunction<T, T, T> aggregator)

첫번째 인자 : 지금까지 누적된 값

두번째 인자 : 누적할 데이터

 

reduce를 사용한 곱셈 

Mono<Integer> mulMono = Flux.just(1, 2, 3, 4).reduce((acc, ele) -> acc * ele); // acc : 이전까지 누적된 값
mulMono.subscribe(sum -> System.out.println("sum : " + sum));

// 다음과 같은 계산을 거친다
acc1 = 1 // 첫 번째 값을 누적 값의 초기 값으로 사용
acc2 = aggregator(acc1, 2) // 1 * 2
acc3 = aggregator(acc2, 3) // 2 * 3
acc4 = aggregator(acc3, 4) // 6 * 4

 

reduce를 사용한 덧셈 

List<Integer> integers = Arrays.asList(1, 2, 3, 4, 5);

Flux.fromIterable(integers)
        .reduce(Integer::sum)
        .subscribe(System.out::println);

 

초기 누적값을 지정하고, 데이터와 다른 타입으로 누적

Flux.just(1, 2, 3, 4)
    .reduce("", (str, ele) -> str + "-" + ele.toString())
    .subscribe(System.out::println);