[Java] 스트림(Stream)

[Java] 스트림(Stream)

 

 

정의

스트림은 통상적으로는 시내, 물줄기 등을 의미합니다.

Java에서 스트림은 "데이터 처리 연산을 위해 데이터소스(컬렉션, 배열)에서 출력된 연속된 요소"라고 할 수 있습니다.

Java8에서 추가된 기능으로 순회하는 연산을 복잡한 반복문(for, while 등)대신 간결한 함수형으로 표현할 수 있습니다.

또한 쓰레드를 이용한 병렬처리(Parallel)를 쉽게 구현할 수 있고, 많은 요소를 빠르게 처리할 수 있습니다.

 

 

특징

1. Java의 I/O Stream과 이름은 스트림으로 동일하나 전혀 다른 개념이다.
2. 다양한 자료구조, 데이터소스(Map, Set, Array)를 동일한 방식으로 처리할 수 있다.
3. 데이터를 담는 저장소(컬렉션)가 아닌 데이터 처리 과정을 의미한다.
4. 원본 데이터를 변경하지 않고 별도의 요소를 Stream으로 생성한다.
5. lterator처럼 일회용이기 때문에 재사용이 불가능하다. 또 순회가 필요하면 다시 생성해야 한다.
6. 내부 반복으로 작업을 처리하기 때문에 어떻게 순회하는지 신경 쓸 필요 없고 코드가 간결하다.
7. 중간 연산과 최종 연산으로 구분된다. 최종 연산 호출전까지 중간연산은 수행되지 않는다. (지연 연산)

 

 

연산 흐름 (파이프 라인)

기본적인 연산의 흐름은 "스트림 생성 -> 중간 연산 -> 최종 연산"의 순서로 진행됩니다.

Stream Source는 배열,컬렉션 등의 데이터 소스를 의미하며 이 소스에서 Stream 인스턴스를 생성할 수 있습니다.

만들어진 Stream 인스턴스는 일반적으로 여러 개의 중간 연산과 하나의 최종 연산이 수행되어 결과물을 생성하게 됩니다.

 

아래 예시와 같이 연산들은 빌더 패턴처럼 메서드 체이닝 방식으로 구현되어 있어 연산의 흐름을 파악하기 쉽습니다.

Arrays.stream(strArr) // 스트림 생성
	.distinct() // 중간 연산
	.limit(5) // 중간 연산
	.sorted() // 중간 연산
	.forEach(System.out::print); // 최종 연산

 

중간 연산 (Intermediate Operation)

- 중간 연산의 결과물로 새로운 스트림이 반환된다.

- 중간 연산은 바로 실행되는게 아니라 최종 연산을 수행하는 시점에 연산을 실행한다. (Lazy)

- 맵핑(map), 필터링(filter, distinct), 정렬(sorted), 반복(peek), 제한(limit, skip) 등의 연산

 

최종 연산 (Terminal Operation)

 - 파이프라인의 마지막 연산으로 스트림이 아닌 다른 타입을 반환한다.

 - 지연 연산이 아닌 즉각 실행되는 연산이다.

 - 반복(forEach), 카운팅(count), 평균(average), 리듀스(reduce) 등의 연산

 

 

중간 연산은 왜 Lazy 할까?

스트림의 중간 연산은 즉시 연산을 실행하는 것이 아니라 최종 연산이 수행되는 시점에 실행됩니다.

왜 굳이 Lazy 하게 연산을 실행할까요? 심심해서, 아무 이유 없이, 그냥은 아닐 것 같습니다.

개발자가 원하는 연산을 바로 실행하는 것과 나중에 실행하는 것에 무슨 차이가 있는지 알 필요가 있습니다. 

 

결론부터 말하면 "모든 단계의 연산을 분석하여 최적화된 연산을 수행하기 위해서"입니다.

스트림 요소에 대한 불필요한 접근과 연산을 하지 않는 것이 대표적인 최적화 방법입니다.

 

최적화 전략 1. 루프 퓨전

루프 퓨전(loop fusion)은 여러 단계의 연산을 하나의 연산으로 합치는 전략을 말합니다.

아래 수도(pseudo) 코드로 예시를 들어 설명드리겠습니다.

배열[1,2,3].stream()
           .a()
           .b()
           .c()

 

어떤 배열에 1,2,3이라는 요소가 들어있고 이를 스트림으로 만들어 a, b, c의 연산을 순서대로 수행하는 코드입니다.

기본적으로는 다음과 같은 순서로 연산이 진행된다고 예측할 수 있습니다.

1 요소의 a연산
2 요소의 a연산
3 요소의 a연산
1 요소의 b연산
2 요소의 b연산
3 요소의 b연산
1 요소의 c연산
2 요소의 c연산
3 요소의 c연산

 

그러나 실제로는 스트림의 최적화 전략을 통하여 아래와 같은 순서로 연산이 진행됩니다.

1 요소의 a연산
1 요소의 b연산
1 요소의 c연산
2 요소의 a연산
2 요소의 b연산
2 요소의 c연산
3 요소의 a연산
3 요소의 b연산
3 요소의 c연산

단순하게 생각하면 똑같이 총 9번의 연산이 수행되기 때문에 차이가 없어 보이지만, 하나의 요소를 꺼내서 모든 연산을 연속적으로 수행하기 때문에 스트림 요소에 대한 접근이 줄어들게 됩니다.

 

앞에 예측에서는 총 9번의 스트림 접근이 필요할 것 같았지만 실제 최적화 이후로는 3번의 접근만으로도 모든 연산을 수행할 수 있습니다.

원시(primitive) 타입의 접근보다 래퍼(wrapper) 타입에 대한 접근은 상대적으로 오버헤드가 크기 때문에 스트림에 대한 접근을 최소화할수록 성능 측면에서 더 효과적입니다.

 

참고로 모든 stream 연산에 대해서 루프 퓨전 방식이 적용되는 것은 아닙니다.

예를 들어 sorted 연산이 포함된 경우는 모든 요소를 확인 후 정렬 작업을 수행하기 때문에 하나의 요소만 연속적으로 연산할 수 없습니다.

이런 특수한 연산이 포함된 파이프라인은 루프 퓨전 같은 최적화 전략이 적용되지 않습니다.

 

최적화 전략 2. 쇼트서킷

쇼트서킷(short circuit)은 필요 없는 연산 과정은 수행하지 않고 넘어가는 전략을 말합니다.

스트림에서는 limit, findFirst 같은 연산이 포함된 경우를 예로 들 수 있습니다.

참고로 limit(n) 연산이 의미하는 것은 n 개의 데이터만 소스 스트림에서 가져와서 새로운 스트림을 생성하고 리턴하는 연산입니다.

배열[1,2,3].stream()
           .a()
           .b()
           .limit(1)

배열의 1,2,3 요소를 스트림으로 만들어 a, b 두 연산을 수행한 뒤 1개의 요소만 스트림으로 만들겠다는 파이프라인입니다.

이 경우 아래와 같은 연산 과정을 수행하게 됩니다.

1 요소의 a연산
1 요소의 b연산

최적화 전략이 없었다면 배열의 1,2,3 요소 모두 a, b연산을 수행한 후 1개의 요소만 스트림으로 리턴됐을 것입니다.

그러나 쇼트서킷 전략으로 인해 배열의 2,3요소는 최종 결과로 리턴될 수 없기 때문에 불필요한 a, b 연산을 수행하지 않습니다.

 

결론적으로 스트림의 중간 연산은 위에 설명한 최적화 전략을 수행할 수 있도록 모든 연산을 파악한 뒤 Lazy 하게 동작하는 것입니다.

 

 

병렬 처리

스트림을 사용하면 순차적인 처리뿐만 아니라 병렬 처리도 쉽게 구현할 수 있습니다.

스트림에서 병렬처리는 하나의 작업을 여러 개의 서브작업으로 나누고, 각각의 서브 작업들을 각 스레드에서 병렬적으로 처리 후 결과를 취합하는 것을 말합니다.

Java에서 스트림을 병렬처리 할 때 내부적으로 포크-조인(Fork/Join) 프레임워크를 사용합니다.

 

아래 예시 코드처럼 데이터 소스에서 parallelStream() 메서드를 사용하면 병렬 스트림을 생성할 수 있습니다.

Arrays.parallelStream()
      .forEach()

 

또한 기존 stream에서 parallel() 메서드를 사용해서 병렬 스트림을 생성할 수도 있습니다.

Arrays.stream()
      .parallel()

주의할 점은 parallel() 메서드를 호출한다고 해서 스트림 자체에 변화가 바로 생기는 것이 아니라, parallel() 이후의 연산이 병렬로 수행돼야 한다는 것을 설정하는 것뿐입니다.

 

위에서 설명한 것처럼 병렬처리를 위해 내부적으로 ForkJoinPool을 사용해서 스레드를 생성하고 관리합니다.

디폴트로 생성되는 스레드 숫자는 (CPU 논리 프로세서 - 1) 개입니다

위의 사진에서 물리적 CPU 코어는 4개지만 하이퍼스레딩 기술로 실제 논리 프로세서는 8개입니다

https://www.intel.co.kr/content/www/kr/ko/gaming/resources/hyper-threading.html

하이퍼 스레딩이란 무엇입니까? - 인텔

 

실제 parallelStream 메서드를 사용해 디버깅해보면 스레드풀의 스레드를 확인할 수 있습니다.

main 스레드 1개 + 추가 생성된 스레드 7개를 확인할 수 있습니다.

만약 생성할 스레드 개수를 변경하고 싶으면 다음과 같이 커스터마이징 할 수 있습니다.

ForkJoinPool customThreadPool = new ForkJoinPool(5);
customThreadPool.submit(() -> Arrays.parallelStream().forEach());

ForkJoinPool customThreadPool2 = new ForkJoinPool(5);
customThreadPool2.submit(() -> Arrays.parallelStream().forEach());

위의 수도 코드에서 ForkJoinPool을 생성할 때 파라미터로 전달하는 숫자 5는 Pool에 생성할 스레드 수를 의미합니다.

위에 예시에서는 스레드풀을 2번 생성했기 때문에 아래와 같이 실제로는 총 10개의 스레드가 생성됩니다.

 

또 다른 방법으로는 시스템 프로퍼티의 설정을 통해 스레드 숫자를 변경할 수 있습니다.

System.setProperty("java.util.concurrent.ForkJoinPool.common.parallelism","10");

 

병렬 처리는 항상 효율적이고 빠른가?

일반적으로 생각하면 스트림을 순차적으로 처리하는 것보다 병렬로 처리하는 것이 빠르다고 생각할 수 있습니다.

일부 상황에서는 맞는 말이지만 그렇다고 병렬 처리가 항상 더 빠른 것은 아닙니다.

스트림의 병렬 처리는 멀티 스레드로 동작하는데, 이는 하나의 작업을 여러 개로 나누고 합치는 작업뿐 아니라 스레드를 생성하고 문맥 정보를 관리하는 등의 추가적인 작업 때문에 오히려 느려질 수도 있습니다.

 

병렬 처리를 하기 위해 기본적으로 고려해야 할 사항은 다음과 같습니다.

- 순차 스트림보다 병렬 스트림에서 성능이 떨어지는 연산을 사용하고 있는지 확인한다

- 스트림 요소의 개수가 많고, 하나의 요소를 처리하는데 충분히 오래 걸린다면 병렬 처리 하는 것을 고려한다.

- 분해성이 좋지 않은 데이터 소스를 사용하는지 확인한다. (ArrayList는 분해성이 좋지만 LinkedList는 좋지 않음)

- CPU 코어 숫자를 확인한다.(싱글 코어인 경우 순차처리가 효율적임)

 

이외에도 여러 가지 주의사항이 있기 때문에 아무 상황에서나 병렬로 처리하는 것은 좋지 않습니다.

언제 사용해야 하는지 성능을 이론적으로 계산하기 힘들다면 병렬 처리를 시간, 리소스에 대해 측정해 보는 것이 가장 쉬운 결정 방법인 것 같습니다.

 

 

스트림 단점

스트림을 사용하면 여러 가지 장점도 있지만 다음과 같은 단점도 있을 수 있습니다.

 

1. 러닝 커브가 있다.

개발을 처음 배울 때 일반적으로 람다 표현식으로 배우는 게 아니기 때문에 기본적인 러닝커브가 있는 편입니다.

람다에 익숙하지 않은 개발자에게는 코드 가독성도 떨어질 수 있습니다.

 

2. 디버깅이 힘들다

스트림은 기본적으로 람다를 사용하기 때문에 익명 함수의 특성을 일부 따르게 됩니다. 익명 함수는 일반 함수에 비해 함수 콜 스택 추적이 어렵고 IDE에서 디버깅하기 힘들 때가 있습니다. (플러그인을 추가로 설치하거나 peek를 사용해 해결)

 

3. 성능이 떨어질 수도 있다.

스트림 내부적으로 최적화를 하긴 하지만 for-loop도 충분한 최적화를 하고 있으며, 스트림의 오버헤드가 더 크기 때문에 성능이 떨어질수도 있습니다.

 

4. 예전 요소로 다시 돌아갈 수 없다.

일부 반복문과 다르게 스트림 요소를 순회하면서 지나간 앞의 요소를 다시 접근할 수 없습니다. 

 

5. 재사용이 불가능하다.

스트림 연산이 한번 끝나면 재사용하는 것이 아니라 새로운 스트림을 생성하게 됩니다.