Chapter16 스트림과 병렬 처리

스트림  - 컬렉션(배열 포함)의 저장 요소를 하나씩 참조해서 람다식(함수적-스타일(functional-style))으로 처리하는 반복자

            람다식으로 요소 처리 코드를 제공 : stream이 제공하는 대부분의 요소 처리 메소드는 함수적 인터페이스 매개 타입을 가지기 때문에 람다식 또는 메소드 참조를 이용해서 요소 처리 내용을 매개값으로 전달

            내부 반복자를 사용하므로 병렬 처리가 쉬움 : 외부 반복자는 개발자가 코드로 직접 컬렉션의 요소를 반복해서 가져오는 코드패턴(index를 이용하는 for문 Iterator를 이용하는 while문은 모두 외부 반복자를 이용), 내부 반복자는 컬렉션 내부에서 요소들을 반복시키고, 개발자는 요소당 처리해야 할 코드만  제공하는 코드 패턴, 내부 반복자의 장점은 어떻게 요소를 반복시킬 것인가는 컬렉션에 맡기고 개발자는 요소 처리 코드에만 집중, 내부 반복자는 요소들의 반복 순서를 변경하거나, 멀티 코어 CPU를 활용하기 위해 요소들을 분배시켜 병렬 작업을 할 수 있게 도와주기 떄문에 하나씩 처리하는 순차적 외부 반복자보다는 효율적으로 요소를 반복시킬 수 있음

                        병렬 처리 : 한 가지 작업을 서브 작업으로 나누고, 서브 작업들을 분리된 스레드에서 병렬적으로 처리하는 것, 병렬 처리 스트림을 이용하면 런타임 시 하나의 작업을 서브 작업으로 자동으로 나누고, 서브 작업의 결과를 자동으로 결합해서 최종 결과물을 생성

            스트림은 중간 처리와 최종 처리를 할 수 있음 : 중간 처리에서는 매핑, 필터링, 정렬을 수행하고 최종 처리에서는 반복, 카운팅, 평균, 총합 등의 집계 처리를 수행

스트림의 종류 - java.util.stream 패키지에는 BaseStream 인터페이스를 부모로 해서 하위로 Stream, IntStream, LongStream, DoubleStream이 있음

//리스트 컬렉션에서 스트림 얻기 Stream<Student> stream = studentList.stream(); stream.forEach(s -> System.out.println(s.getName())); //배열로부터 스트림 얻기 Stream<String> strStream = Arrays.stream(strArray); strStream.forEach(a -> System.out.print(a + ",")); IntStream intStream = Arrays.stream(intArray); intStream.forEach(a -> System.out.print(a + ",")); //정수 범위로부터 스트림 얻기 IntStream stream = IntStream.rangeClosed(1, 100); stream.forEach(a -> sum += a); //파일로부터 스트림 얻기 Path path = Paths.get("src/chapter16/linedata.txt"); Stream<String> stream; stream = Files.lines(path, Charset.defaultCharset()); //디렉토리로부터 스트림 얻기 Path path = Paths.get("C:/DDoongMause/workspace/ThisIsJava/src"); Stream<Path> stream = Files.list(path); stream.forEach( p -> System.out.println(p.getFileName()));

cs

스트림 파이프라인

중간 처리와 최종 처리 - 스트림은 데이터의 필터링, 매핑, 정렬, 그룹핑 등의 중간 처리와 합계, 평균, 카운팅, 최대값, 최소값 등의 최종 처리를 파이프라인(여러 개의 스트림이 연결되어 있는 구조)으로 해결, 파이프라인에서 최종 처리를 제외하고는 모두 중간 처리 스트림, 중간 스트림이 생설될 때 요소들이 바로 중간 처리되는 것이 아니라 최종 처리가 시작되기 전까지 중간처리는 지연(lazy)됨

리턴 타입이 스트림이라면 중간 처리 메소드이고, 기본 타입이거나 OptionalXXX라면 최종 처리 메소드임

필터링(distinct(), filter()) - 중간 처리 기능으로 요소를 걸러내는 역할

매핑(flatMapXXX(), mapXXX(), asXXXStream(), boxed()) - 매핑은 중간 처리 기능으로 스트림의 요소를 다른 요소로 대체하는 작업

            flatMapXXX() 메소드 - 요소를 대체하는 복수 개의 요소들로 구성된 새로운 스트림을 리턴

            mapXXX() 메소드 - 요소를 대체하는 요소로 구성된 새로운 스트림을 리턴

            asDoubleStream(), asLongStream(), boxed() 메소드 - asDoubleStream() 메소드는 int 요소 또는 long 요소를 double 타입 변환, asLongStream() 메소드는 int 요소를 long 요소로 타입 변환, boxed() 메소드는 int, long, double 요소를 Integer, Long, Double 요소로 박싱해서 Stream 생성

정렬(sorted()) - 요소를 정렬해서 최종 처리 순서를 변경, 객체 요소일 경우 Comparable을 구현하지 않으면 ClassCastException이 발생하므로 Comparable을 구현한 요소에서만 sorted() 사용해야 함

sorted();
sorted( (a,b) -> a.compareTo(b) );
sorted( Comparator.naturalOrder() );
//역방향
sorted( (a,b) -> b.compareTo(a) );
sorted( Comparator.reverseOrder() );

루핑(peek(), forEach()) - 전체를 반복하는 메소드는 peek(), forEach()가 있고, peek()는 중간 처리 메소드이므로 반드시 최종 처리 메소드가 호출 되어야 하고, forEach()는 최종 처리 메소드임

매칭(allMatch(), anyMatch(), noenMatch()) - 최종 처리 단계에서 요소들이 특정 조건에 만족하는지 조사할 수 있도록 세가지 매칭 메소드(allMatch(), anyMatch(), noenMatch())가 있음

            allMatch() - 모든 요소들이 매개값으로 주어진 Predicate의 조건을 만족하는지 조사

            anyMatch() - 최소한 한 개의 요소가 매개값으로 주어진 Predicate의 조건을 만족하는지 조사

            noneMatch() - 모든 요소들이 매개값으로 주어진 Predicate의 조건을 만족하지 않는지 조사

기본 집계(sum(), count(), average(), max(), min()) - 집계는 최종 처리 기능으로 요소들을 처리해서 카운팅, 합계, 평균값, 최대값, 최소값등과 같이 하나의 값으로 산출하는 것이고, 대량의 데이터를 가공해서 축소하는 리덕션(Reduction)

            스트림이 제공하는 기본 집계 

            Optional 클래스 - 

커스텀 집계(reduce()) - 프로그램화해서 다양한 집계 결과물을 만들 수 있도록 reduce() 제공

수집(collect()) - 필터링 또는 매핑된 요소들을 새로운 컬렉션에 수집하고, 이 컬렉션을 리턴(T는 요소, A는 누적기, R은 요소가 저장될 컬렉션), T 요소를 A 누적기가 R에 저장함

?는 T를 저장하는 방법을 알고 있어서 A가 필요 없기 때문, ConcurrentMap은 스레드에 안전함

첫번째 Supplier는 요소들이 수집될 컨테이너 객체(R)를 생성, 병렬 처리 스트림에서는 여러 번 Supplier가 실행되고 스레드별로 여러개의 컨테이너 객체를 생성 하지만 최종적으로 하나의 컨테이너 객체로 결합됨

두번째 XXXConsumer는 컨테이너 객체(R)에 요소(T)를 수집하는 역할

세번째 BiConsumer는 컨테이너객체를 결합하는 역할, 순차 처리 스트림에서는 호출되지 않고, 병렬 처리 스트림에서만 호출되어 스레드별로 생성된 컨테이너 객체를 결합해서 최종 컨테이너 객체를 완성

요소들을 그룹핑해서 Map객체를 생성하는 기능도 제공

Collectors.groupingBy() 메소드는 그룹핑 후, 매핑이나 집계를 할 수 있도록 두 번째 매개값으로 Collector를 가질 수 있음

병렬 처리 -  병렬 스트림을 제공하기 떄문에 컬렉션(배열)의 전체 요소 처리 시간을 줄임

            동시성과 병렬성 : 동시성은 멀티 작업을 위해 멀티 스레드가 번갈아가며 실행하는 성질을 말하고, 병렬성은 멀티 작업을 위해 멀티 코어를 이용해서 동시에 실행하는 성질을 말함

                        데이터 병렬성 : 전체 데이터를 쪼개어 서브 데이터들로 만들고 이 서브 데이터들을 병렬 처리해서 작업을 빨리 끝내는 것

                        작업 병렬성 : 서로 다른 작업을 병렬 처리 하는 것

            포크조인 프레임 워크 : 포크 단계에서는 전체 데이터를 서브 데이터로 분리하고 서브 데이터를 멀티 코어에서 병렬로 처리한 후, 조인 단계에서는 서브 결과를 결합해서 최종 결과를 생성, ExecutorService의 구현 객체인 ForkJoinPool을 사용해서 작업 스레드를 관리

            병렬 스트림 생성 : parallelStream()는 컬렉션으로부터 병렬 스트림을 바로 리턴, parallel()는 순차 처리 스트림을 병렬 처리 스트림으로 변환해서 리턴

            병렬 처리 성능 : 병렬 처리가 항상 실행 성능이 좋다고 판단해서는 안됨

                        요소의 수와 요소당 처리 시간 : 컬렉션에 요소의 수가 적고 요소당 처리 시간이 짧으면 순차 처리가 오히려 병렬 처리보다 빠를 수 있음, 스레드풀 생성, 스레드 생성이라는 추가적인 비용이 발생하기 때문

                        스트림 소스의 종류 : ArrayList, 배열은 인덱스로 요소를 관리하므로 포크 단계에서 요소를 쉽게 분리할 수 있어 병렬 처리 시간이 절약되지만, HashSet, TreeSet은 요소 분리가 쉽지 않고, LinkedList 역시 링크를 따라가야 하므로 요소 분리가 쉽지 않아서 ArrayList, 배열보다는 병렬 처리가 늦음

                        코어의 수 : 싱글 코어 CPU일 경우는 순차 처리가 빠름, 병렬 스트림을 사용할 경우 스레드의 수만 증가하고 동시성 작업으로 처리되기 때문에 좋지 못한 결과