"스트림 API의 공백을 채우는 새로운 방법" 스트림 수집기 알아보기

자바 22에 데이터 스트림 조작을 위한 새로운 메커니즘인 스트림 수집기(stream gatherer)가 도입된다. 스트림 수집기는 JEP 461에 제공되는 기능으로, 개발자가 복잡한 작업을 간소화하는 맞춤형 중간 연산자를 만들 수 있게 해준다. 스트림 수집기에 대한 첫인상은 약간 복잡하고 모호하다. 또한 왜 필요한지 의문이 들 수도 있다. 그러나 특정 종류의 스트림 조작이 필요한 상황에 직면하게 되면 스트림 API에 추가된 수집기가 반갑게 느껴질 것이다.

ⓒ Getty Images Bank

스트림 API와 스트림 수집기

자바 스트림은 요소의 동적 컬렉션을 모델링한다. 사양에 나와 있듯 '스트림은 느리게 계산되고 잠재적으로 무한한 값의 시퀀스'다.

즉, 무한하게 데이터 스트림을 소비하고 이를 기반으로 작업을 할 수 있다는 의미다. 강가에 앉아 강물이 흘러가는 모습을 지켜본다고 생각해 보자. 이때 강이 끝날 것이라는 생각은 하지 않을 것이다. 스트림에서는 강과 강에 포함된 모든 요소로 그냥 작업을 하고 끝나면 떠나면 된다.

스트림 API에는 값 시퀀스의 요소를 다루기 위한 여러 메소드가 내장돼 있다. filter, map과 같은 함수적 연산자가 이러한 메소드다.

스트림 API에서 스트림은 이벤트의 소스로 시작되며 filter, map과 같은 연산을 "중간" 연산이라고 한다. 각 중간 연산은 스트림을 반환하므로 이를 묶어 구성할 수 있다. 스트림 API에서 자바는 스트림이 "터미널" 연산에 도달하기 전까지는 이러한 연산을 적용하지 않는다. 이 방식 덕분에 많은 연산자가 체인으로 연결된 경우에도 효율적인 처리가 가능하다.

스트림의 내장된 중간 연산자는 강력하긴 하지만 그렇다고 모든 요구사항을 처리할 수는 없다. 기본 틀에서 벗어난 상황에서는 맞춤형 연산을 정의할 방법이 필요한데, 수집기가 그 방법을 제공한다.

스트림 수집기로 할 수 있는 것

예를 들어 강가에서 숫자가 적힌 나뭇잎이 강물을 따라 흘러가는 모습을 보고 있다고 생각해 보자. 여기서 간단한 일, 예를 들어 눈에 띄는 모든 짝수의 배열을 만들려면 내장된 filter 메소드를 사용하면 된다.

List<Integer> numbers = Arrays.asList(1, 2, 3, 4, 5, 6);
numbers.stream().filter(number -> number % 2 == 0).toArray()
// result: { 2, 4, 6 }

위 예제에서 정수 배열(소스)로 시작한 다음 스트림으로 변환하고 2로 나누었을 때 나머지가 없는 수만 반환하는 필터를 적용한다. toArray() 호출은 터미널 호출이다. 이는 각 나뭇잎의 짝수성을 검사해서 통과하면 따로 빼 두는 것과 같다.

스트림 수집기의 내장 메소드

java.util.stream.Gatherers 인터페이스에는 맞춤형 중간 연산을 구성할 수 있게 해주는 몇 가지 내장 함수가 제공된다. 각 내장 함수의 기능을 살펴보자.

windowFixed 메소드
떠내려가는 모든 나뭇잎을 주워서 두 개의 바구니에 담으려는 경우를 생각해 보자. 내장 함수 연산자를 사용할 경우 생각 외로 힘든 작업이다. 한 자리 숫자의 배열을 배열의 배열로 변환해야 하기 때문이다.
The windowFixed method is a simpler way to gather your leaves into buckets:
windowFixed 메소드를 사용하면 더 간단한 방법으로 나뭇잎을 바구니에 담을 수 있다.

Stream.iterate(0, i -> i + 1)
.gather(Gatherers.windowFixed(2))
.limit(5)
.collect(Collectors.toList());

이 코드는 정수를 1 단위로 반복하는 스트림을 만들고 매 두 개의 요소를 하나의 새 배열로 변환하고 이를 5번 반복하고 마지막으로 스트림을 List로 변환하라는 뜻이다. 결과는 다음과 같다.

[[0, 1], [2, 3], [4, 5], [6, 7], [8, 9]]

윈도잉은 스트림 위에서 프레임을 이동하는 것과 같으며 스냅샷을 찍을 수 있게 해준다.

windowSliding 메소드
또 다른 윈도잉 함수는 windowSliding이다. windowFixed()와의 차이는 각 윈도우의 시작점이 마지막 윈도우의 끝이 아니라 소스 배열의 다음 요소라는 점이다. 예를 들면 다음과 같다.

Stream.iterate(0, i -> i + 1)
.gather(Gatherers.windowSliding(2))
.limit(5)
.collect(Collectors.toList());

출력은 다음과 같다.

[[0, 1], [1, 2], [2, 3], [3, 4], [4, 5]]

windowSliding 출력을 windowFixed 출력과 비교하면 차이점을 볼 수 있다. windowFixed와 달리 windowSliding의 각 하위 배열에는 이전 하위 배열의 마지막 요소가 포함된다.

Gatherers.fold 메소드
Gatherers.fold는 Stream.reduce 메소드의 더 다듬어진 버전이라고 할 수 있다. fold()가 reduce()보다 더 쓸모 있는 경우는 다소 모호하다. 이 글에서 이와 관련된 좋은 내용을 볼 수 있다. 저자인 빅터 클랭은 fold와 reduce의 차이점에 대해 다음과 같이 말한다.

접기(folding)는 축소(reduction)의 일반화다. 축소에서 결과 형식은 요소 형식과 동일하고 컴바이너는 결합형이며 초기 값은 컴바이너의 ID다. 접기의 경우 병렬화 가능성을 포기하기는 하지만 이러한 조건이 불필요하다.

reduce는 fold의 일종이다. 축소는 스트림을 가져와 하나의 값으로 변환한다. 접기도 같은 작업을 수행하지만 요구사항은 더 느슨해서, 1) 반환 형식이 스트림 요소와 같은 형식이어야 하고 2) 컴바이너가 결합형이어야 하며 3) fold의 이니셜라이저가 정적 값이 아닌 실제 생성기 함수여야 한다.

두 번째 요구사항은 잠시 후에 더 자세히 살펴볼 병렬화와 관계가 있다. 스트림에서 Stream.parallel을 호출하는 것은 엔진이 이 작업을 여러 쓰레드로 분할할 수 있음을 의미한다. 이는 연산자가 결합형인 경우에만 작동한다. 즉, 연산의 순서가 결과에 영향을 미치지 않는 경우 작동한다.

fold의 간단한 사용 예는 다음과 같다.

Stream.of("hello","world","how","are","you?")
.gather(
Gatherers.fold(() -> "",
(acc, element) -> acc.isEmpty() ? element : acc + "," + element
)
)
.findFirst()
.get();

이 예제는 문자열 컬렉션을 받아서 이를 쉼표와 결합한다. 같은 작업을 reduce로 하면 다음과 같다.

String result = Stream.of("hello", "world", "how", "are", "you?")
.reduce("", (acc, element) -> acc.isEmpty() ? element : acc + "," + element);

fold에서는 초기 값(“”) 대신 함수(() -> “”)를 정의하는 것을 볼 수 있다. 즉, 개시자에 대해 더 복잡한 처리가 필요한 경우 closure 함수를 사용할 수 있다.

이제 형식의 다양성과 관련된 fold의 장점에 대해 생각해 보자. 예를 들어 혼합 객체 형식의 스트림이 있고 발생 횟수를 세려고 한다.

var result = Stream.of(1,"hello", true).gather(Gatherers.fold(() -> 0, (acc, el) -> acc + 1));
// result.findFirst().get() = 3

result var는 3이다. 스트림에는 번호와 문자열, 부울 값이 있다. 비슷한 작업을 reduce로 하기는 어렵다. 누적자 인수(acc)가 강한 형식이기 때문이다.

// bad, throws exception:
var result = Stream.of(1, "hello", true).reduce(0, (acc, el) -> acc + 1);
// Error: bad operand types for binary operator '+'

collector를 사용해서 이 작업을 수행할 수 있다.

var result2 = Stream.of("apple", "banana", "apple", "orange")
.collect(Collectors.toMap(word -> word, word -> 1, Integer::sum, HashMap::new));

그러나 이 경우 더 복잡한 로직이 필요할 때 이니셜라이저와 접기 함수 본문에 액세스하지 못하게 된다.

Gatherers.scan 메소드
scan은 windowFixed와 비슷하지만 요소를 배열이 아닌 하나의 요소로 누적한다. 마찬가지로 예제를 통해 더 명확히 볼 수 있다(예제의 출처는 자바독스).

Stream.of(1,2,3,4,5,6,7,8,9)
.gather(
Gatherers.scan(() -> "", (string, number) -> string + number)
)
.toList();

출력은 다음과 같다.

["1", "12", "123", "1234", "12345", "123456", "1234567", "12345678", "123456789"]

scan을 사용하면 스트림 요소를 따라 이동하면서 누적 방식으로 결합할 수 있다.

mapConcurrent 메소드
mapConcurrent를 사용하면 제공된 map 함수를 실행할 때 동시에 사용할 쓰레드의 최대 수를 지정할 수 있다. 가상 쓰레드가 사용된다. 다음은 숫자를 제곱하면서 동시성을 쓰레드 4개로 제한하는 간단한 예제다. (참고로 이와 같은 간단한 데이터 집합에 mapConcurrent를 사용하는 것은 과하다.)

Stream.of(1,2,3,4,5).gather(Gatherers.mapConcurrent(4, x -> x * x)).collect(Collectors.toList());
// Result: [1, 4, 9, 16, 25]

쓰레드 최대값 외에는 표준 map 함수와 똑같이 작동한다.

결론

스트림 수집기가 기능으로 승격될 때까지는 Gatherer 인터페이스와 그 기능에 액세스하려면 --enable-preview 플래그를 사용해야 한다. 실험하려면 JShell을 사용해서 $ jshell --enable-preview와 같이 쉽게 할 수 있다.

스트림 수집기는 매일 필요한 기능은 아니지만 스트림 API의 오랜 공백을 채워주며 개발자가 함수형 자바 프로그램을 더 쉽게 확장하고 맞춤 구성할 수 있게 해준다.
editor@itworld.co.kr

Matthew Tyson editor@itworld.co.kr
저작권자 한국IDG & ITWorld, 무단 전재 및 재배포 금지