Stream

Stream이란

컬섹션에 저장되어있는 엘리먼트들을 하나씩 순회하면서 데이터를 처리할 수 있는 코드패턴이다.

람다식과 함께 사용되어 컬렉션에 들어있는 데이터에 대한 처리를 매우 간결한 표현으로 작성할 수 있다.

또 한, 내부 반복자를 사용하기 때문에 병렬처리가 쉽다.

 

Stream을 사용하기 전과 후의 코드를 확인 해보자

List<Integer> list = List.of(1, 2, 3, 4, 5, 6);

// 스트림 사용 이전
for (int i : list) {
    System.out.println("i = " + i);
}

for (int i : list) {
    if( i % 2 ==0 ){
        System.out.println("even = " + i);
    }
}
System.out.println();

// 스트림 사용 이후

list.forEach(x -> System.out.println("x = " + x));

list.stream()
        .filter(f -> f%2==0)
        .forEach(x -> System.out.println("even = " + x));

 

결과

1 2 3 4 5 6 
2 
4 
6 

1 2 3 4 5 6 
2 
4 
6

위 코드처럼 Stream을 쓰면 간격하고 가독성 좋게 소스를 구성할수 있다.

 

 

List<Integer> list1 = Arrays.asList(1, 2, 3, 4, 5);
Integer[] arr = new Integer[]{11,22,33,44,55};

Stream<Integer> list1Stream = list1.stream();
Stream<Integer> stream = Arrays.stream(arr);

list1Stream.sorted().forEach(x -> System.out.println("x = " + x));
stream.sorted().forEach(x -> System.out.println("x = " + x));

위 소스와 같이 데이터 소스는 서로 다르지만 정렬하고 화면에 출력하는 방법은 동일하다.

 

Stream은 데이터소스를 변경하지 않는다

스트림은 데이터 소스로 부터 데이터를 읽기만할 뿐, 데이터 소스를 변경하지 않는다.

예를들어 스트림을 사용해서 정렬을 한후 리스트로 반환하여도 기존 데이터에는 영향이 없다.

List<Integer> list = Arrays.asList(1,6,4,2,7,8,4,5,3);

List<Integer> collect = list.stream().sorted().collect(Collectors.toList());

System.out.println("Stream Sort");
collect.forEach(x -> System.out.print(x + " "));
System.out.println();

System.out.println("기존 데이터");
list.forEach(x -> System.out.print(x + " "));

 

결과

Stream Sort
1 2 3 4 4 5 6 7 8 
기존 데이터
1 6 4 2 7 8 4 5 3

위와같이 기존 데이터에는 변화가 없다.

 

 

스트림은 일회용이다

스트림은 Iterator처럼 일회용이다. 컬렉션의 요소를 모두 읽고 나면 닫히기 때문에 다시 사용할 수 없다. 필요하다면 스트림을 다시 생성해야한다.

Stream<Integer> stream = List.of(1, 2, 3, 4, 5, 6, 7, 8).stream();

stream.sorted().forEach(x -> System.out.println(x + " "));

stream.filter(f -> f % 2 == 0).forEach(x -> System.out.println(x + " ")); // 에러발생

결과

 

1 
2 
3 
4 
5 
6 
7 
8 
Exception in thread "main" java.lang.IllegalStateException: stream has already been operated upon or closed
	at java.base/java.util.stream.AbstractPipeline.<init>(AbstractPipeline.java:203)
	at java.base/java.util.stream.ReferencePipeline.<init>(ReferencePipeline.java:94)
	at java.base/java.util.stream.ReferencePipeline$StatelessOp.<init>(ReferencePipeline.java:696)
	at java.base/java.util.stream.ReferencePipeline$2.<init>(ReferencePipeline.java:165)
	at java.base/java.util.stream.ReferencePipeline.filter(ReferencePipeline.java:164)
	at test.stream_test2.StreamTest4.main(StreamTest4.java:13)

위와같이 스트림이 이미 닫혀서 사용할 수 없다고 Excpetion을 발생 시킨다.

 

 

스트림의 연산

스트림이 제공하는 다양한 연산을 이용해서 복잡한 작업들을 간단히 처리할 수 있다.

스트림이 제공하는 연산은 중간 연산과 최종 연산으로 분류할 수 있는데, 중간 연산은 연산결과를 스트림으로 반환하기 때문에 중간 연산을 연속해서 연결할 수 있다. 반면에 최종 연산은 스트림의 요소를 소모하면서 연산을 수행하므로 단 한번만 연산이 가능한다.

 

중간 연산 : 연산 결과가 스트림인 연산. 스트림에 연속해서 중간 연산을할 수 있음
최종 연산 : 연산 결과가 스트림이 아닌 연산. 스트림의 요소를 소모하므로 단 한번만 가능

 

stream.disticnt().limit(5).sorted().forEach(System.out::println);

                 ↑          ↑          ↑                        ↑

                  중  간   연   산                   최종 연산

 

모든 중간 연산의 결과는 스트림이지만, 연산 전의 스트림과 같은 것은 아니다. 위의 문장과 달리 모든 스트림 연산을 나누어 쓰면 아래와 같다.

 

List<Integer> list = List.of(1, 1, 2, 2, 3, 3, 4, 4, 5, 5, 6, 6, 7, 7, 8, 8, 9, 9, 10, 10);

Stream<Integer> stream = list.stream(); //  Integer형 스트림 생성
Stream<Integer> filterStream = stream.filter(f -> f % 2 == 0);  // list에서 짝수만 걸러낸다(중간연산)
Stream<Integer> distinctStream = filterStream.distinct();       // 중복을 걸러낸다(중간연산)
Stream<Integer> sortedStream = distinctStream.sorted();         // 정렬한다(중간연산)
Stream<Integer> limitStream = sortedStream.limit(5);//  스트림의 개수를 자른다(중간연산)
long count = limitStream.count();//  요소 개수 세기(최종 연산)
System.out.println("count = " + count);

위 소스와 같이 중간연산의 경우 Stream 형태이지만 최종 연산의 결과는 Stream의 형태가 아니다.

 

 

Stream에 정의된 연산

중간연산	설명
Stream distinct()	중복을제거
Stream filter (Predicate predicate)	조건에 안 맞는요소제외
Stream limit (long maxSi ze)	스트림의 일부를 낸다.
Stre n skip(long n)	스트림의 일부를 건너띈다.
Stre n skip(long n)	스트림의 소에 작업수행
Stream sorted()  Stream sorted(Comparator comparator)	스트림의 요소를 정렬한다
Stream map (Function mapper) DoubleStream mapToDouble (ToDoubleFunction mapper) IntStream mapTolnt (TolntFunction mapper) LongStream mapToLong (ToLongFunction mapper) Stream flatMap (Function> mapper) DoubleStream flatMapToDouble (Function m) IntStream flatMapTolnt (Function m) LongStream flatMapToLong (Function m)	스트림의 요소를 변환한다.

최종 연산	설명
void forEach (Consume r action ) void forEachOrdered(Consumer action)	각 소에 정된 작업 수행
long count ( )	스트림의 의 개수 반환
Optional max (Comparator comparator) Optional min (Comparator compa rator)	스트림의 최대 /최소값을 반환
Optional findAny ( ) Optional findFirst ( )	스트림의 요소 하나를 반환
boolean allMatch (Predicate p) boolean anyMatch (Predicate p) boolean noneMatch (Predi cate<> p)	주어진 조건을 모든 요소가 만족시키는지, 만족시키지 않는지 확인
Object[] toArray() A[] toArray (IntFunction generator)	스트림의 모든 요소를 배열로 반환
Optional reduce (BinaryOperator accumulator) T reduce (T identity, BinaryOperator accumulator)  U reduce (U identity, BiFunction accumulator, BinaryOperator combiner)	스트림의 요소를 하나씩 줄여가면서(리듀싱)계산 한다.
R collect (Collector collector) R collect (Supplier supplier, BiConsumer              accumulator, BiConsumer combiner)	스트림의 요소를 수집한다. 주로 요소를 그룹화하거나 분할한 결과를 컬렉션에 담아 반환하는데 사용한다.

 

Stream<Integer>와 IntStream

요소의 타입이 T인 스트림은 기본적으로 Stream<T>이지만, 오토박싱&언박싱으로 인한 비효율을 줄이기 위해 데이터 소스의 요소를 기본형으로 다루는 스트림, IntStream, LongStream, DoubleStream이 제공된다.

일반적으로 Stream<Interger< 대신 IntStream을 사용하는 것이 더 효율적이고, IntStream에는 int타입의 값으로 작업하는데 유용한 메서드들이 포함되어 있다.

 

병렬 스트림

스트림으로 데이터를 다룰 때의 장점 중 하나가 바로 병렬 처리가 쉽다는 것이다. 스트림에 parallel()이라는 메서드를 호출하면 병렬로 연산을 수행하도록할 수 있다.반대로 병렬로 처리되지 않게 하려면 sequential()을 호출하면 된다. 하지만 모든 스트림은 기본적으로 병렬 스트림이 아니므로 sequential()을 호출할 필요가 없다. 이 메서드는 parallel()을 호출한 것을 취소할 때만 사용한다.

 

int sum = strStream.parallel()	//	strStream을 병렬 스트림으로 전환
			.mapToInt(s -> s.length)
            .sum();

참고
parallel()과 sequential()은 새로운 스트림을 생성하는 것이 아니라, 그저 스트림의 속성을 변경할 뿐이다.

참고
병렬처리가 항상 더 빠른 결과를 얻게 해주는 것은 아니라는 것을 명심하자

 

Stream<Integer>, IntStream, 병렬스트림 비교

TimeChecker timeChecker = new TimeChecker();
List<Integer> collect = new Random().ints(0, 10000000).limit(100000000).boxed().collect(Collectors.toList());

// IntStream 스트림 병렬 처리
timeChecker.start();
int asInt = collect.stream().mapToInt(x -> x).parallel().reduce(Integer::sum).getAsInt();
timeChecker.end();
System.out.println("IntStream 병렬 처리 = " + asInt);

// IntStream 스트림 순차 처리
timeChecker.start();
int asInt1 = collect.stream().mapToInt(x -> x).sequential().reduce(Integer::sum).getAsInt();
timeChecker.end();
System.out.println("IntStream 순차 처리 = " + asInt1);

// Stream<T> 병렬 처리
timeChecker.start();
Integer integer = collect.stream().parallel().reduce(Integer::sum).get();
timeChecker.end();
System.out.println("Stream<T> 병렬 처리 = " + integer);

// Stream<T>  순차 처리
timeChecker.start();
Integer integer2 = collect.stream().reduce(Integer::sum).get();
timeChecker.end();
System.out.println("Stream<T> 순차 처리 = " + integer2);

결과

 

IntStream 병렬 처리 = 241ms

IntStream 순차 처리 = 599ms

Stream<T> 병렬 처리 = 1907ms

Stream<T> 순차 처리 = 2740ms

위 결과와 같이 IntStream 병렬 처리가 가장 빠른것을 볼 수 있다.

 

속도 차이

IntStream 병렬 처리 > IntStream 순차 처리 > Stream<T> 병렬 처리 > Stream 순차 처리

하지만 병렬처리가 항상 더 빠른 결과를 얻게 해주는 것은 아니라는 것을 명심하자

 

 

range, reangeClosed

IntStream과 LongStream은 다음과 같이 지정된 범위의 연속된 정수를 스트림으로 생성해서 반환하는 range(), rangeCloed()를 가지고 있다.

IntStream	IntStream.range(int begin, int end)
IntStream	IntStream.rangeClosed(int begin, int end)

range()의 경우 경계의 끝인 end가 범위에 포함되지 않는다.(begin ~ (end -1))

ex. ) IntStream.range(1,5) => 1,2,3,4

 

rangeClosed()의 경우 end까지 포함된다. (begin ~ end)

ex. ) IntStream.rangeClosed(1,5) => 1,2,3,4,5

 

임의의수 생성

난수를 생성하는데 사용하는 Random클래스에는 아래와 같은 인스턴스 메서드들이 포함되어 있다. 이 메서드들은 해당 타입의 난수들로 이루어진 스트림을 반환한다.

IntStream	ints()
LongStream	longs()
DoubleStream	doubles()

이 메서드들이 반환하는 스트림은 크기가 정해지지 않은 무한 스트림이므로 limit()도 같이 사용해서 스트림의 크기를 제한해 주어야 한다. limit()은 스트림의 개수를 지정하는데 사용되며, 무한 스트림을 유한 스트림으로 만들어 준다

 

위 메서드들에 의해 생성된 스트림의 난수는 아래의 범위를 같는다.

	Integer.MIN_VALUE	<=	ints()		<=	Integer.MAX_VALUE
	Long.MIN_VALUE		<=	longs()		<=	Long.MAX_VALUE
		0.0		<=	doubles()	<=	1.0

또한 스트림의 크기를 지정하거나, 지정된 범위의 난수를 발생시키는 스트림을 생성할수 있다.

아래는 스트림의 크기를 지정하는 메서드이다.

IntStream	ints(long streamSize)
LongStream	longs(long streamSize)
DoubleStream	doubles(long streamSize)

IntStream ints1 = new Random().ints(5);
ints1.forEach(x -> System.out.println("x = " + x));

-- 결과 --
x = -467402834
x = -2036584167
x = 391277077
x = -937549216
x = 1383982062

 

 

아래는 지정된 범위(begin~end)의 난수를 발생시키는 메서드이다.

IntStream	ints(int begin, int end)	// limit() 사용 필수
LongStream	longs(long begin, long end) // limit() 사용 필수
DoubleStream	doubles(double begin, double end) // limit() 사용 필수
IntStream	ints(long streamSize, int begin, int end)
LongStream	longs(long streamSize, long begin, long end)
DoubleStream	doubles(long streamSize, double begin, double end)

 

iterate(), generate()

iterate()와 generate()는 람다식을 매개변수로 받아서, 이 람다식에 의해 계산되는 값들을 요소로 하는 무한 스트림을 생성한다.

 

static <T> Stream<T> iterate(T seed, UnaryOperator<T> f)
static <T> Stream<T> generate(Supplier<T> s)

iterate()는 seed값으로 지정된 값부터 시작해서, 람다식 f에 의해 계산된 결과를 다시 seed값으로 해서 계산을 반복한다.

Stream.iterate(0, a -> a + 2).limit(5).forEach(x -> System.out.println("x = " + x));

--- 결과 ---
x = 0
x = 2
x = 4
x = 6
x = 8

 

generate()도 iterate()처럼, 람다식에 의해 계산되는 값을 요소로 하는 무한 스트림을 생성해서 반환하지만, iterate()와 달리, 이전결과를 이용해서 다음 요소를 계산하지 않는다.

Stream.generate(() -> 1).limit(10).forEach(x -> System.out.println("x = " + x));

--- 결과 ---
x = 1
x = 1
x = 1
x = 1
x = 1
x = 1
x = 1
x = 1
x = 1
x = 1

 

 

두 스트림의 연결

Stream의 static메서드인 concat()을 사용하면, 두 스트림을 하나로 연결할 수 있다.

연결하려는 두 스트림의 데이터 요소는 같은 타입이어야 한다.

Stream<String> stream1 = List.of("aa", "bb", "cc", "dd").stream();
Stream<String> stream2 = List.of("AA", "BB", "CC", "DD").stream();

Stream.concat(stream1, stream2).forEach(x -> System.out.println("x = " + x));

--- 결과 ---
x = aa
x = bb
x = cc
x = dd
x = AA
x = BB
x = CC
x = DD

 

 

References

Java의 정석, 남궁 성 지음