Stream API

임준철·2021년 3월 14일

Java Stream API 병렬스트림

JavaAdvanced

목록 보기

11/15

스트림 API(Stream API)

여기서 말하는 스트림과 나중에 자바 io에서 말하는 스트림은 다른 것이다.
- 여기서의 스트림은 자료의 연산을 위해서 사용하는 객체를 말한다.
- 자바 io에서의 스트림은 입출력을 위한 스트림이다.

스트림 API

Java 8에서 추가된 Java.util.stream 패키지
컬렉션의 요소를 람다식으로 처리할 수 있도록 하는 함수형 프로그래밍 도구
스트림 API는 데이터를 추상화하여 다루므로, 다양한 방식으로 저장된 데이터를 읽고 쓰기 위한 공통된 방법을 제공한다.
따라서 스트림 API를 이용하면 배열이나 컬렉션뿐만 아니라 파일에 저장된 데이터도 모두 같은 방법으로 다룰 수 있게 된다.
중간 연산과 최종 연산으로 구분 된다,최종 연산이 수행되어야 모든 연산이 적용되는 지연 연산이다.

스트림의 특징

1. 스트림은 한 번 밖에 사용 못한다.

내부적으로 몇번째 동작을 하고 있는지 체크하는 것이 들어가 있다. 그래서 사용할 때마다 다시 정의해주고 사용해야 한다.

2. 간결한 코드로 작성할 수 있다.

// 기존의 Java 7방식의 작성 방법
List<String> list = Arrays.asList("fast","campus","rocks");
List<String> newList = new ArrayList<>();

for (String s : list){
    newList.add(s.toUpperCase());
}
for (String s : newList){
    System.out.println(s);
}

// java 8 - Stream API -> 훨씬 더 간결한 코드로 작성할 수 있다.
List<String> list1 = Arrays.asList("fast","campus","rocks");
Stream<String> stream = list1.stream(); //스트림으로 변환해준다.

// 스트링에서 스트림으로 매핑해준다.
//uppercase를 string에 매핑, 각각의 요소에 적용해라
stream.map(String::toUpperCase).forEach(System.out::println); // foreach는 Consumer를 받는다,각각의 요소를 하나씩 출력한다.

3. 데이터 소스에 대한 공통된 접근 방식 제공.

Stream으로 변경해주고 나면, List, Set, Map 모두 동일한(표준화된) 방식으로 데이터를 접근 가능.
컬렉션 프레임워크는 각각을 다른 방식으로 데이터를 처리 해줘야 한다.
특히 Stream::sorted() 메소드는 공통된 정렬 방식을 제공한다.

4.중간 처리와 최종 처리를 지원한다.

Mapping, Filtering, Sorting 등 중간 처리 지원(여러 개 사용이 가능하다.)
Iteration, Count, Average, Summation, Reduce 등 최종 처리 지원(마지막에 하나 사용 가능하다.)

5.데이터 소스와 처리 결과를 분리

원본 데이터를 유지하여 처리하는 것에 의한 부작용을 방지한다.
처리 결과를 새로운 컬렉션으로 저장이 가능하다.

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

스트림이 제공하는 대부분의 요소처리 메소드는 함수적 인터페이스 매개타입을 가진다.
매개값으로 람다식 또는 메소드 참조를 대입할 수 있다.
(함수적 인터페이스 매개타입을 가지기 때문에 람다식,메소드 참조로 매개값을 제공할 수 있음)

7. 내부 반복자를 사용하므로 병렬 처리가 쉽다.

외부 반복자 - 개발자가 코드로 직접 컬렉션 요소를 반복해서 요청하고 가져오는 코드패턴
(Iterator를 생각하면 될 것 같다. Iterator을 사용해서 반복문을 구현해서 하나씩 값을 출력하게
코드를 구현해야 하는 것을 말하는 것 같다.)
내부 반복자 - 컬렉션 내부에서 요소들을 반복시키고 개발자는 요소당 처리해야할 코드만 제공하는 코드 패턴
(forEach메소드 처럼 내부가 어떻게 구현되어있는지 모르지만 이것을 사용하면 요소들 하나씩 가져오라는 코드가 없지만
알아서 반복해서 반환시켜 처리해준다. 이런 것을 의미하는 것 같다)
내부 반복자의 이점 - 개발자는 요소 처리 코드에만 집중, 멀티 코어 cpu를 활용하기 위해 요소들을 분배시켜 병렬 처리 작업을 수 있다.
- 컬렉션의 요소가 많을 경우에 요소를 나눠서 별도의 스트림으로 만들어서 각각의 코어에서 병렬적으로 처리할 수 있다.

스트림 API를 이용한 자료 처리

스트림 API의 종류

종류	처리 대상
`Stream<T>`	일반적인 객체를 처리
`IntStream`	기본 자료형 `int`를 처리
`LongStream`	기본 자료형 `long`을 처리
`DoubleStream`	기본 자료형 `double`을 처리

스트림 객체 생성 메소드

데이터 소스(인터페이스, 클래스)로부터 스트림 생성

데이터 소스	메소드
Collection	default Stream stream()
Collection	default Stream parallelStream()
Arrays	public static Stream stream(T[] array)
Arrays	public static Stream of(T ... values)
Arrays	public static IntStream stream(int[] array)
Arrays	public static IntStream of(int ... values)
Arrays	public static LongStream stream(long[] array)
Arrays	public static LongStream of(long ... values)
Arrays	public static DoubleStream stream(double[] array)
Arrays	public static DoubleStream of(double ... values)

정수 범위와 java.util.Random으로부터 생성

구분	메소드
int형 범위	public static IntStream range(int startInclusive, int endExclusive)
int형 범위	public static IntStream rangeClosed(int startInclusive, int endInclusive)
long형 범위	public static LongStream range(long startInclusive, long endExclusive)
long형 범위	public static LongStream rangeClosed(long startInclusive, long endInclusive)
Random p형 값	public PStream ps()
Random p형 값	public PStream ps(long streamSize)
Random p형 값	public PStream ps(long streamSize, p origin, p bound)

스트림 생성 방식 예제

컬렉션의 인스턴스 메소드 stream()

Stream<String> stream1 = list1.stream();

Arrays 클래스의 클래스 메소드 stream() 을 이용

 // 함수형 인터페이스랑 비슷함 P Type
 // LongStream, DoubleStream 도 있다.
 int[] ints = {4, 6, 2, 19, 2, 58, 4, 6, 5};
 IntStream intStream = Arrays.stream(ints);
 intStream.forEach(System.out::println);

위의 두 개의 다른점
Stream<Integer>로 할 경우, Primitive Type이 들어올 때 오토박싱으로 변경되고 다시 출력할 때 언방식이 되는데 이럴 때 오버헤드가 발생한다.
IntStream을 사용 할 경우, 오버헤드 필요 없이, WrapperClass없이 사용 가능하다. 그래서 효율적이다.
Stream 클래스의 클래스 메소드 of()를 이용해서 Collection을 걸치지 않고도 스트림을 생성 가능

DoubleStream doubleStream = DoubleStream.of(0.4, 0.6, 0.2, 1.2, 0.94);
doubleStream.forEach(System.out::println);

range()를 이용한 스트림 -> for i문 (for(int i=0...))을 대체하는 스트림
rangeClosed() : 0~10까지 10을 포함한다.
순서대로 값을 출력한다.

IntStream intStream1 = IntStream.range(0,10); // 0~9까지 10은 포함되지 않는다.
intStream1.forEach(System.out::println);

IntStream intStream2 = IntStream.rangeClosed(0,10); // 0~10까지 10포함된다.
intStream2.forEach(System.out::println);
// LongStream도 range, rangeClosed 가 있다.

Random 객체를 이용한 스트림이 있다. Java.util 패키지 안에 있다.

Random random = new Random();
// LongStream longStream =  random.longs();
// longStream.forEach(System.out::println); // 개수 제한 없이 무한히 출력

// 개수 제한 가능
LongStream longStream1 =  random.longs(100);
longStream1.forEach(System.out::println); // 개수를 정해질 수있다 100개

// 개수제한 + 범위 제한 가능 젤 많이 사용함
LongStream longStream2 =  random.longs(100,0,1000);
longStream2.forEach(System.out::println); // 개수를 정해질 수있다 0~1000까지 100개를 출력

중간 처리 메소드

https://velog.io/@im_joonchul/Java-Stream%EC%A4%91%EA%B0%84%EC%B2%98%EB%A6%AC%EB%A9%94%EC%86%8C%EB%93%9C

최종 처리 메소드

https://velog.io/@im_joonchul/Java-Stream%EC%B5%9C%EC%A2%85%EC%B2%98%EB%A6%AC%EB%A9%94%EC%86%8C%EB%93%9C

병렬 스트림

병렬 처리

한가지 작업을 서브 작업으로 나누고, 서브 작업들을 분리된 스레드에서 병렬 적으로 처리한 후,
서브 작업들의 결과들을 최종 결합하는 방법, 자바는 ForkJoinPool 프레임워크를 이용해서 병렬 처리를 한다.

병렬 스트림의 생성

stream() 대신 parallelStream()으로 변경
stream 생성 후 parallel()으로 병렬화

combiner를 이용해 병렬 스트림으로 생성된 컬렉션을 결합

BiConsumer<T, K> combiner : T 객체에 K 객체를 결합

인터페이스	리턴타입	메소드(매개변수)
java.util.Collection	Stream	parallelStream()
java.util.Stream	Stream	parallel()
java.util.Intstream	IntStream
java.util.Longstream	Longstream
java.util.Doublestream	Doublestream

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

병렬성 구분

데이터 병렬성
- 데이터 병렬성은 한 작업 내에 있는 전체 데이터를 쪼개어 서브 데이터들로 만들고
  이 서브 데이터들을 병럴 처리해서 작업을 빨리 끝내는 것을 말한다.
작업 병렬성
- 작업 병렬성은 서로 다른 작업을 병렬 처리하는 것을 말한다.
- 작업 병렬성은 대표적인 예는 웹서버이다.
  웹 서버는 각각의 브라우저에서 요청한 내용을 개별 스레드에서 병렬로 처리한다.
병렬 스트림은 데이터 병렬성을 구현한 것이다.
- 멀티 코어의 수만큼 대용량 요소를 서브 요소들로 나누고,
  각각의 서브 요소들을 분리된 스레드에서 병렬 처리시킨다.
- 예를 들어 쿼드 코어cpu일 경우 4개의 서브 요소들로 나누고,
  4개의 스레드가 각가의 서브 요소들을 병렬처리한다.
- 병렬 스트림은 포크조인 프레임워크를 이용한다.

Stream<String> parStream = Arrays.stream(arr).parallel(); //  스트림을 parallel로 바꿔줌
System.out.println(parStream.map(String::length).count());
// parallelStream을 사용하면 연산 순서가 달라질 수 있다.

List<String> list4 = List.of("atwe","bff","cqqqw","dtwer");
Stream<String> stream6 = list4.parallelStream(); // 어떤 요소가 먼저 수행되는지 알 수 없음
// Stream<String> stream6 = list4.stream();

stream6.map(String::length).peek(s->System.out.println("A: "+ s))
              .filter(value->value >=3)
              .peek(s-> System.out.println("B :"+ s))
              .forEach(System.out::println);

포크조인 프레임워크(ForkJoin Framework)

포크조인 프레임워크 동작 방식
- 포크 단계
  - 데이터를 서브 데이터로 반복적으로 분리한다.
  - 서브 데이터를 멀티 코어에서 병렬로 처리한다.
- 조인 단계
  - 서브 결과를 결합해서 최종 결과를 만들어 낸다.
- 실제로 병렬 처리 스트림은 포크 단게에서 차례대로 요소를 4등분 하지 않는다.내부적으로 서브 요소를 나누는 알고리즘이 있기 때문에 개발자는 신경 쓸 필요가 없다.
포크조인풀(포크조인프레임워크는 내부적으로 스레드를 포크조인풀에서 관리한다.)
- 각각의 코어에서 서브요소를 처리하는 것은 개별 스레드가 해야하므로 스레드 관리가 필요
  - 코어에서 작업을 할려면 코어별로 각각의 스레드가 존재해야한다.
- 포크조인 프레임워크는 ExcutorService의 구현 객체인 ForkJoinPool을 사용
- 4개의 코어를 가진 cpu일 경우

병렬 처리 성능

병렬 처리는 항상 빠르다?
- 스트림 병렬 처리가 스트림 순차 처리보다 항상 실행 성능이 좋다고 판단해서는 안된다.
병렬 처리에 영향을 미치는 3가지 요인
- 요소의 수와 요소당 처리 시간 - 요소우의 수가 적고 요소당 처리시간이 짧으면 순차 처리가 오히려 병렬처리보다
  빠를 수 있다 병렬처리는 스레드풀 생성, 스레드 생성이라는 추가적인 비용이 발생하기 때문이다.
- 스트림 소스의 종류
  - ArrayList, 배열은 램덤 액세스를 지원(인덱스로 접근)하기 때문에 포크 단계에서 쉽게 요소를
    분리할 수 있어 병렬 처리 시간이 절약된다, 반면 HashSet, TreeSet은 요소를 분리하기가 쉽지않고,
    LinkedList는 램덤 액세스를 지원하지 않아서 요소를 분리하기가 쉽지 않다.
    또한, BufferedReader.lines()은 전체 요소의 수를 알기 어렵기 때문에 포크 단계에서 부분 요소로
    나누기 어렵다. 따라서 이들 소스들은 ArrayList, 배열 보다는 상대적으로 병럴처리가 늦다.
- 코어의 수
  *싱글코어 cpu일 경우 순차처리가 빠르다. 코어의 수가 많으면 병렬처리 속도가 빨라진다.

패럴레 스트림을 사용한다는 것이 어떤 의미인가

원래 스트림(페럴레아닌 상태)은 [0,0,0,0,0,0] 이런게 있다로 시작, 스트림을 정하고 맵이 있으면 맵을 한 결과가 나와야함[0,0,0,0,0,0]
그런다음 맵을 한 번 더 실행하고 [0,0,0,0,0,0] foreach가 실행이 된다. 하나하나가 함수로 소비가 되는 상태가 된다
하나씩 처리가 된다 각 요소 가 모든 연산이 쭉되고 다음 요소가 모든 연산이 실행이 됨
그 중간에 peek(),peek(),foreach()의 결과가 출력이 된다 이것이 우리가 생각하는 기본적인 스트림

병렬 스트림 동작

자바에서 스레드가 기본적인건 모르겠으나 여기서 7개의 스레드가 각각있다고 가정하고 동시에 동작을한다
스레드는 프로그램을 동작할 때 한줄씩 실행되는데 이 한 줄씩 실행되는 것이 여러개 있는 것이다.각각 독립적으로 한줄을 실행한다.
한줄 한줄 계산하는게 동시에 계산해도 된다 각각을 순수함수로 되어있어야 하니깐
- 여기서 한줄이란 세로로 한줄이 시작되고 그옆에 세로가 시작되는건데 병렬처리면 스레드의 개수에 따라
  이 세로줄의 실행이 스레드 개수에 따라서 한번에 실행된다는 의미이다 4개스레드이면 4개의 세로줄이 동시에 실행됨.
- 한줄 계산되는 동안 다른 한줄이 같이 계산되고 몇개가 같이 연산이 진행되는진 모르지만 스레드의 개수에 따라 한줄이 같이 진행됨
[0,0,0,0,0,0]
[0,0,0,0,0,0]
[0,0,0,0,0,0]
[0,0,0,0,0,0]
명령형/ 순차처리 <-> 함수형/병렬처리 큰 간극을 보인다

임준철

지금, 새로운 문을 열자! 문 저편에 무엇이 있을지 두렵더라도!!

Stream API

JavaAdvanced

스트림 API(Stream API)

스트림 API

스트림의 특징

1. 스트림은 한 번 밖에 사용 못한다.

2. 간결한 코드로 작성할 수 있다.

3. 데이터 소스에 대한 공통된 접근 방식 제공.

4.중간 처리와 최종 처리를 지원한다.

5.데이터 소스와 처리 결과를 분리

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

7. 내부 반복자를 사용하므로 병렬 처리가 쉽다.

스트림 API를 이용한 자료 처리

스트림 API의 종류

스트림 객체 생성 메소드

스트림 생성 방식 예제

중간 처리 메소드

최종 처리 메소드

병렬 스트림

병렬 처리

병렬 스트림의 생성

병렬성 구분

포크조인 프레임워크(ForkJoin Framework)

병렬 처리 성능

패럴레 스트림을 사용한다는 것이 어떤 의미인가

병렬 스트림 동작

Java Stream_최종처리메소드

This

0개의 댓글