[Java8]Stream(스트림)

스트림과 컬렉션

스트림과 컬렉션은 모두 연속된 요소 형식의 값을 저장하는 자료구조의 인터페이스를 제공
(둘다 연속적인 자료르 처리하기 위해 만들어짐)

가장 큰 차이는 언제 계산하느냐!

컬렉션은 모든 값을 메모리에 저장하는 구조
어떤 계산이든 모든 요소들이 컬렉션에 추가되기 전에 미리 계산 (외부반복)

스트림은 요청할 때만 요소를 계산 (스트림에 요소를 추가하거나 제거하는 건 불가능)

Stream

• 자료의 대상과 관계없이 동일한 연산을 수행한다. (일관성)
• 배열, 컬렉션을 대상으로 연산을 수행
• 일관성 있는 연산으로 자료의 처리를 쉽고 간단하게 함
  "자료 처리를 추상화했다."라고 표현함

---

• 한번 생성하고 사용한 스트림은 재사용이 불가능하다.
• 자료에 대해 스트림을 생성하여 연산을 수행하면 스트림은 소모된다.
• 다른 연산을 수행하기 위해서는 스트림을 다시 생성해야한다.

---

• 스트림의 연산은 기존의 자료를 변경하지 않는다.
• 자료에 대한 스트림을 생성하면, 스트림이 사용하는 메모리 공간은 별도로 생성되므로 연산이 수행되도 기존 자료에 대한 변경은 발생하지 않는다.

---

• 스트림은 중간 연산과 최종 연산으로 구분된다.
• 스트림에 대해 중간 연산은 여러 개의 연산이 적용될 수 있지만, 최종 연산은 마지막에 한번만 적용된다.
• 최종 연산이 호출되어야 중간 연산에 대한 수행이 이루어지고 그 결과가 만들어진다.
  때문에 중간 연산에 대한 결과를 연산 중에 알 수 없는데 이를 '지연 연산(lazy)'이라고 한다.

---

Stream 사용 예제

import java.util.Arrays;
import java.util.stream.IntStream;

public class IntArrayStreamTest {
    public static void main(String[] args) {
        int[] arr = {1, 2, 3, 4, 5};
        for(int num : arr){
            System.out.println(num);
        }
        System.out.println();

        //스트림을 생성하는 경우
        Arrays.stream(arr).forEach(n -> System.out.println(n));

        IntStream is = Arrays.stream(arr);
        is.forEach(n-> System.out.println(n));
    }
}

• 배열을 만들면 Arrays(java.util.Arrays) 클래스를 활용할 수 있음
  Arrays 내부의 stream메소드를 사용
  Arrays.stream(arr)
  : Integer배열에 있는 자료를 갖고 stream operation을 할 수 있는 Intstream 객체를 return

• forEach(intConsumer arg0)
  : 최종 연산 스트림 / 요소들을 하나씩 꺼내는 메소드
  intConsumer > parmeter를 소모하면서 해야할 액션을 구현 = 람다식 구현

---

중간 연산과 최종 연산

import java.util.ArrayList;
import java.util.List;
import java.util.stream.Stream;

public class ArrayListStreamTest {
    public static void main(String[] args) {
        List<String> sList = new ArrayList<String>();
        sList.add("Dalgu");
        sList.add("Bingo");
        sList.add("Bukang");

        System.out.println(sList);

        // Collection의 메소드 stream()
        Stream<String> stream = sList.stream();
        stream.forEach(s-> System.out.println(s));

        //중간 스트림 (.sorted())
        sList.stream().sorted().forEach(s-> System.out.println(s));

        //객체 자체에 대한 operation을 사용할 때는 map()을 이용
        // (ex. 객체 자체가 가진 get,set메소드 등을 이용할 때)
        sList.stream().map(s->s.length()).forEach(n-> System.out.println(n));

        //filter()
        sList.stream().filter(s->s.length() >= 6).forEach(n -> System.out.println(n));
    }
}

• 중간 연산의 예 filter(),map(),sorted() 등
  조건에 맞는 요소를 추출(filter)하거나 요소를 변환 함(map)
• 최종 연산이 호출 될 때, 중간 연산이 수행되고 결과가 생성 됨

---

Stream의 특징

람다식으로 요소 처리 코드를 제공

import java.util.Arrays;
import java.util.List;
import java.util.stream.Stream;

public class LambdaExpressionExample {
    public static void main(String[] args) {
        List<Student> list = Arrays.asList(
                new Student("달구", 90),
                new Student("빙고", 92)
        );

        //스트림생성
        Stream<Student> stream = list.stream();
        
        //List 컬렉션에서 Student를 매개값으로 가져와 람다식의 매개값으로 제공
        stream.forEach(s -> {
            String name = s.getName();
            int score = s.getScore();
            System.out.println(name + " : " + score);
        });
    }
}

내부 반복자를 사용, 병렬 처리가 용이

외부반복자(external iterator)

개발자가 코드로 직접 컬렉션의 요소를 반복해서 가져오는 코드 패턴.
index를 이용하는 for문, Iterator를 이용하는 while문 등은모두 외부 반복자를 이용한다.

내부반복자(internal iterator)

컬렉션 내부에서 요소들을 반복시키고, 개발자는 요소당 처리해야할 코드만 제공하는 코드 패턴

👉내부 반복자는 요소들의 반복 순서를 변경하거나, 멀티 코어 CPU를 최대한 활용하기 위해 요소들을 분배시켜 병렬 작업을 할 수 있게 도와주기 때문에 하나씩 처리하는 순차적 외부 반복자보다는 효율적으로 요소를 반복시킬 수 있음!

👉병렬 처리 역시 컬렉션 내부에서 처리되므로 good
cf.병렬처리 : 한 가지 작업을 서브 작업으로 나누고, 서브 작업들을 분리된 스레드에서 병렬적으로 처리하는 것. 런타임 시에 하나의 작업을 서브 작업으로 자동으로 나누고, 서브 작업의 결과를 자동으로 결합해서 최종 결과물을 생성.

>🚨다만, Parallel()은 공유된 thread pool을 사용하기 때문에 심각한 성능 장애를 일으킬 수 있다.

Java8 Paralle Stream, 성능장애
stream을 사용할 때 흔히 하는 실수 10가지

import java.util.Arrays;
import java.util.List;
import java.util.stream.Stream;

public class ParallelExample {
    public static void main(String[] args) {
        List<String> list = Arrays.asList("조성주", "이병렬", "김도욱", "김유진", "조성호");

        //순차처리
        Stream<String> stream = list.stream();
        //stream.forEach(s->ParallelExample.print(s))와 동일
        stream.forEach(ParallelExample::print);
        System.out.println();

        //병렬처리
        Stream<String> parallelStream = list.parallelStream();
        parallelStream.forEach(ParallelExample::print);
    }

    public static void print(String str){
        System.out.println(str + " : " + Thread.currentThread().getName());
    }
}

• 결과 (병렬처리는 main,ForkJoinPool(스레드풀) 2가지 스레드에서 작업을 한 걸 확인 가능)
  

중간 처리와 최종 처리가 가능

import java.util.Arrays;
import java.util.List;

public class MapAndReduceExample {
    public static void main(String[] args) {
        List<Student> studentList = Arrays.asList(
                new Student("달구", 10),
                new Student("빙고", 15),
                new Student("부강", 20)
        );

        double avg = studentList.stream()
                .mapToInt(Student::getScore)        //중간처리(학생 객체를 점수로 매핑)
                .average()                          //최종처리(평균점수)
                .getAsDouble();

        System.out.println(avg);
    }
}

• 대량의 데이터를 가공해서 축소하는 것을 Reduction(리덕션)이라고 함.
• 합계, 평균값, 카운팅, 최대,최소값 등이 모두 대표적 리덕션의 결과물
  결과물을 바로 집계할 수 없을 경우, 집계하기 좋도록 stream의 중간처리가 필요한 것
  ex) 남자 평균 나이

double avg = list.stream()
	.filter(m -> m.getSex()==Member.MALE)		// 남자만 필터링
    .mapToInt(Member::getAge)					// 객체를 나이로 매핑 
    .average()									// 평균 나이 집계
    .getAsDouble();

---

(+)Java8 Parallele Stream, 성능장애

• 일반 병렬처리 예제

ExecutorService executor = Executors.newFixedThreadPool(5);
for (int i = 0; i < dealmaxList.size(); i++) {
	final int index = i;
    executor.submit(() -> {
		Thread.sleep(5000);
		System.out.println(Thread.currentThread().getName() 
			+ ", index=" + index + ", ended at " + new Date()); 	 
    });
}       
executor.shutdown();

• Java Parallel Stream 병렬처리 예제

dealmaxList.parallelStream().forEach(index -> {
	System.out.println("Starting " + Thread.currentThread().getName() 
		+ ", index=" + index + ", " + new Date());
	try {
		Thread.sleep(5000);
	} catch (InterruptedException e) { }
});