JAVA - Stream (스트림)

1. 스트림 의미
1) 컬렉션의 저장 요소를 하나씩 참조해 람다식으로 처리할 수 있도록 도와주는 반복자. (자바8부터 추가)
2) 자바7 이전까지는 List< String > 컬렉션에서 요소를 순차적으로 처리하기 위해 Iterator 반복자를 사용했음.

/* Iterator */
List<String> list = Arrays.asList("홍길동", "김자바");
Iterator<String> iterator = list.iterator();
while(iterator.hasNext()){
	String name = iterator.next();
}

/* Stream */
List<String> list = Arrays.asList("홍길동", "김자바");
Stream<String> stream = list.stream();
stream.forEach( name -> System.out.println(name));

 

2. 스트림의 특징
1) 람다식으로 요소 처리 코드를 제공.
2) 스트림은 작업을 내부 반복으로 처리함. (for문을 메서드 안으로 넣어서 처리)

for문을 간단하게 forEach로 표현

forEach 메서드 안에 들어 있는 for문

3) 작업을 내부 반복으로 처리하기 때문에 병렬 처리가 쉬움.

병렬 스트림

4) 스트림은 데이터 소스로부터 데이터를 읽기만 할 뿐 원본을 변경하지 않음.

List<Integer> list = Arrays.asList(3,1,5,4,2);
List<Integer> sortedList = list.stream().sorted()     //list를 정렬해서
                      .collect(Collectors.toList());  //새로운 List에 저장
System.out.println(list);          // [3,1,5,4,2] 원본 변경 없이
System.out.println(sortedList);    // [1,2,3,4,5] 정렬된 복사본만 있음

5) 스트림은 Iterator 처럼 일회용임. (필요하면 다시 스트림을 생성해야 함)

strStream.forEach(System.out::println);   //모든 요소를 화면에 출력(최종 연산)
int numOfStr = strStream.count();         //에러. 최종 연산 후 스트림이 이미 닫힘.

6) 최종 연산 전까지 중간 연산이 수행되지 않음. (지연된 연산)

IntStream intStream = new Random().ints(1,46);    //1~45 범위의 무한 스트림
intStream.distinct().limit(6).sorted()            //중간 연산
         .forEach(i->System.out.print(i+","));    //최종 연산

7) 기본형 스트림 - IntStream, LongStream, DoubleStream....etc.
  - 오토박싱 & 언박싱의 비효율이 제거됨. (Stream<Integer> 대신 IntStream 사용하면 성능이 더 좋아짐)
  - 숫자와 관련된 유용한 메서드를 Stream<T>보다 더 많이 제공함.

3. 스트림 만드는 법
1) 스트림 만들기

스트림 연산 순서

2) 중간 연산 (0~3번까지)
  - 연산 결과가 스트림인 연산. 반복적으로 사용 가능.
  - 중간 처리에서는 매핑, 필터링, 정렬 수행.
3) 최종 연산 (0~1번까지)
  - 연산 결과가 스트림이 아닌 연산. 단 한 번만 적용 가능 (스트림의 요소를 소모)
  - 반복, 카운팅, 평균, 총합 등의 집계 처리 수행.

4. 스트림의 종류
1) 자바8부터 java.util.stream 패키지에서 스트림 API들을 제공함.

 

2) 컬렉션으로부터 스트림 얻기
  - Collection 인터페이스의 stream()으로 컬렉션을 스트림으로 변환

Stream<E> stream()   //Collection 인터페이스의 메서드

===예시 1===
List<Studunt> studentList = Arrays.asList(
	new Student("홍길동", 10), 
    	new Student("김자바", 20)
);

Stream<Student> stream = studentList.stream();


//


===예시 2===
List<Integer> list = Arrays.asList(1,2,3,4,5);
Stream<Integer> intStream = list.stream();   //list를 스트림으로 변환

//스트림의 모든 요소를 출력
intStream.forEach(System.out::print);   //12345
intStream.forEach(System.out::print);   //에러. 최종연산 후 일회용인 스트림이 이미 닫힘

intStream = list.stream();              //list로부터 stream을 다시 생성해서 출력 가능
intStream.forEach(System.out::print);

3) 객체 배열로부터 스트림 얻기

Stream<T> Stream.of(T... values)  //가변 인자
Stream<T> Stream.of(T[])
Stream<T> Arrays.stream(T[])
//from~to 배열의 범위 만들기 (마지막 to는 안들어감 -1)
Stream<T> Arrays.stream(T[] array, int startInclusive, int endExclusive)

===예시===
Stream<String> strStream = Stream.of("a", "b", "c");   //가변 인자
Stream<String> strStream = Stream.of(new String[]{"a", "b", "c"});
Stream<String> strStream = Arrays.stream((new String[]{"a", "b", "c"});
Stream<String> strStream = Arrays.stream((new String[]{"a", "b", "c"}, 0, 3);

 
- 기본형 배열로부터 스트림 얻기

IntStream IntStream.of(int... values)   //Stream 이 아닌 IntStream
IntStream IntStream.of(int[])
IntStream Arrays.stream(int[])
IntStream Arrays.stream(int[] array, int startInclusive, int endExclusive)

4) 특정 범위의 정수를 요소로 갖는 스트림 얻기 (IntStream, LongStream)

// rangeClosed() : 첫번째 인자값에서 두번째 인자값까지 순차적으로 IntStream 리턴 (끝 숫자 포함 안함)
IntStream IntStream.range(int begin, int end)

// rangeClosed() : 첫번째 인자값에서 두번째 인자값까지 순차적으로 IntStream 리턴 (끝 숫자 포함)
IntStream IntStream.rangeClosed(int begin, int end)

IntStream intStream = IntStream.range(1, 5);            // 1,2,3,4
IntStream intStream = IntStream.rangeClosed(1, 5);      // 1,2,3,4,5

5) 난수를 요소로 갖는 스트림 생성 (임의의 수 만들기)

IntStreamintStream = new Random().ints();          //무한 스트림 생성
intStream.limit(5).forEach(System.out::println);   //5개의 요소만 출력

IntStream intStream = new Random().ints(5);        //크기가 5인 난수 스트림 반환


//ints(), longs(), doubles() 메서드
Integer.MIN_VALUE <= ints() <= Integer.MAX_VALUE
Long.MIN_VALUE <= longs() <= Long.MAX_VALUE
0.0 <= doubles() < 1.0

6) 람다식으로 스트림 얻기 - iterate(), generate()
  - 람다식을 소스로 하는 무한 스트림 생성

static <T> Stream<T> iterate(T seed, UnaryOperator<T> f)  //이전 요소에 종속적
static <T> Stream<T> generate(Supplier<T> s)              //이전 요소에 독립적

iterate(). seed 라는 초기값을 사용함

generate(). 초기값을 사용하지 않음.

//iterate(T seed, UnaryOperator f) 단항 연산자
Stream<Integer> intStream = Stream.iterate(1, n -> n+2);  //홀수 출력
Stream<Integer> intStream2 = Stream.iterate(0, n -> n+2);  //짝수 출력
intStream.limit(10).forEach(System.out::println);
intStream2.limit(10).forEach(System.out::println);
		
//generate(Supplier s) : 입력만 없고, 출력만 있음
Stream<Integer> oneStream = Stream.generate(()->1);
oneStream.limit(10).forEach(System.out::println);

7) 파일로부터 스트림 얻기
  - 파일을 소스로 하는 스트림 생성하기

// 파일의 경로 정보를 가지고 있는 Path 객체 생성
Path path = Paths.get("src/sec02/stream_kind/lineddata.txt");

// Charset.defaultCharset()는 운영체제의 기본 문자셋
Stream<String> stream = Files.lines(path, Charset.defaultCharset());

// BufferedReader의 lines() 이용
File file = path.toFile();
FileReader fileReader = new FileReader(file);
BufferedReader br = new BufferedReader(fileReader);
stream = br.lines();

8) 비어있는 스트림 만들기

9) 디렉토리로부터 스트림 얻기

 

Path path = Paths.get("C:/JavaProgramming/source");
Stream<Path> stream = Files.list(path);
stream.forEach( p -> System.out.println(p.getFileName()));

5. 스트림 파이프라인
1) 리덕션
  - 대량의 데이터를 가공해서 축소하는 것.
  - 예) 데이터의 합계, 평균, 최대값, 카운팅 등
  - 컬렉션의 요소를 리덕션의 결과물로 바로 집계할 수 없는 경우 리덕션이 집계하기 좋도록 필터링, 매핑, 정렬, 그룹핑 등 중간 처리 필요
2) 중간처리와 최종 처리
  - 스트림은 데이터의 중간 처리와 최종 처리를 파이프라인(여러 개의 스트림이 연결된 구조)로 해결. 최종 처리 시작 전까지 중간 처리는 지연됨.

예) 회원 컬렉션에서 남자 평균 나이 집계하기

double ageAvg = list.stream()  			// 오리지날 스트림
    .filter(m -> m.getSex() == Member.MALE)  	// 중간 처리 스트림
    .mapToInt(Member :: getAge)  		// 중간 처리 스트림
    .average()  				// 최종 처리
    .getAsDouble

 

6. 스트림의 연산 - 중간 연산 메서드

1) 필터링
  - 요소들을 걸러내는 역할을 하며 distinct(), filter()는 모든 스트림이 가지고 있는 메서드.

2) 매핑
  - 스트림의 요소를 다른 요소로 대체하는 작업.
  -  flatMap() : 요소를 대체하는 복수 개의 요소들로 구성된 새로운 스트림 리턴. 스트림의 스트림을 스트림으로 변환.

List<String> inputList1 = Arrays.asList("java8 lamda", "stream mapping");
inputList1.stream().flatMap(data -> Arrays.stream(data.split(" ")));

List<String> inputList2 = Arrays.asList("10, 20, 30", "40, 50, 60");
inputList2.stream()
    .flatMapToInt(data -> {
        String strArr = data.split(",");
        int intArr = new int[strArr,length];
        for(int i = 0; strArr > i; i++){
        	intArr[i] = Integer.parseInt(strArr[i].trim());
        }
        return Arrays.stream(intArr);
    })

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

 

List<Studunt> studentList = Arrays.asList(
	new Student("홍길동", 10), 
    	new Student("김자바", 20)
);

studentList.stream().mapToInt(Student :: getScore);

  - asDoubleStream(),asLongStream() : IntStream의 int 요소 또는 LongStream의 Long 요소를 double 요소로 타입 변환해서 DoubleStream을 생성한다.
  - boxed() : int, double 요소를 Integer, Double 요소로 박싱해서 Stream을 생성한다.

3) 정렬
  - 요소가 최종 처리되기 전에 중간 단계에서 요소를 정렬.

  - 객체 요소일 경우 클래스가 Comparable을 구현하지 않으면 sorted()를 호출했을 때 예외가 발생.

  - 기본 비교 방법으로 정렬하고 싶을 때

sorted();
sorted( (a,b) -> a.compareTo(b) );
sorted(Comparator.naturalOrder());

  - 기본 비교 방법과 정반대 방법으로 정렬하고 싶을 때

sorted( (a,b) -> b.compareTo(a) );
sorted( Comparator.reversOrder() );

 

7. 스트림의 연산 - 최종 연산

1) 매칭
  - allMatch() : 모든 요소들이 매개값으로 주어진 Predicate의 조건을 만족하는지 조사
  - anyMatch() : 최소한 1개의 요소가 매개값으로 주어진 Predicate의 조건을 만족하는지 조사
  - noneMatch() : 모든 요소들이 매개값으로 주어진 Predicate의 조건을 만족하지 않는지 조사

2) Optional<T>
  - T 타입 객체의 래퍼 클래스.
  - 모든 종류의 객체를 저장할 수 있음. (null 도 가능)
  - 집계 값을 저장하며 집계 값이 존재하지 않을 경우 디폴트 값을 설정할 수 있음.
  - 집계 값을 처리하는 Consumer도 등록할 수 있음.
  - OptionalXXX 타입에서 값을 얻으려면 get(), getAsXXX()를 호출.

 

// isPresent()
OptionalDoble optional = list.stream().mapToInt(Integer :: intValue).average();

if(optional.isPresent()){
	~
}else{
	~
}

// orElse()
double avg = list.stream().mapToInt(Integer :: intValue).average().orElse(0.0);

// ifPresent() : 값이 있을 경우 람다식 실행
list.stream().mapToInt(Integer :: intValue).average().ifPresent(a -> ~);

3) 커스텀 집계
  - reduce() : 다양한 집계 결과를 만드는 메소드

// 요소가 없을 경우 NoSuchElementException 발생
int sum = studentList.stream()
	.map(Student :: getScore)
	.reduce((a,b) -> a + b)
	.get();
    
// 요소가 없어도 default 값 0 리턴    
int sum = studentList.stream()
	.map(Student :: getScore)
	.reduce(0, (a,b) -> a + b)
	.get();

 

4) 수집
  - collect() : 요소를 필터링, 매핑한 후 컬렉션에 수집하는 최종 처리 메소드

// 매개값인 Collector는 어떤 요소를 수집할 것인지 결정
// 타입 파라미터의 T는 요소, A는 누적기, R은 요소가 저장될 컬렉션
R(리턴타입) | collect(Collector<T,A,R> collector) | Stream(인터페이스)

 
  -  Collector의 구현 객체를 얻을 수 있는 Collectors의 정적 메소드

 

  - 사용자 정의 컨테이너에 수집

1) 첫 번째 Supplier는 요소들이 수집된 컨테이너 객체를 생성.
  - 순차 처리(싱글 스레드) 스트림 : 한번 실행, 하나의 컨테이너 객체 생성
  - 병렬 처리(멀티 스레드) 스트림 : 여러번 실행, 스레드 별로 여러 개 생성 -> 최종적으로 하나의 컨테이너 객체로 결

2) 두번 째 XXXConsumer는 컨테이너 객체(R)에 요소(T)를 수집.
  - 스트림에서 요소를 컨테이너에 수집할 때마다 실행.
  - 세번 째 XXXConsumer는 컨테이너 객체(R)를 결합. (병렬 처리만 호출)

// 기본
Stream<Student> totalStream = totalList.stream();
Stream<Student> maleStream = totalList.filter(s -> s.getSex() == Student.Sex.MALE);
Supplier<MaleStudent> supplier = ()->new MaleStudent();
BiConsumer<MaleStudent, Student> accumulator = (ms, s) -> ms.accumulate(s);
BiConsumer<MaleStudent, MaleStudent> combiner = (ms1, ms2) -> ms1.combine(ms2);
MaleStudent maleStudent = maleStream.collect(supplier, accumulator, combiner);

// 변수 생략
MaleStudent maleStudent = totalList.stream()
	.filter(s->s.getSex() == Student.Sex.MALE)
    .collect(
    	() -> new MaleStudent(),
        (r, t) -> r.accumulate(t),
        (r1, r2) -> r1.combine(r2)
    );
    
// 람다식 메소드 참조
MaleStudent maleStudent = totalList.stream()
	.filter(s -> s.getSex() == Student.Sex.MALE)
    .collect(MaleStudent :: new
    	   , MaleStudent :: accumulate
           , MaleStudent :: combine);

  - 요소를 그룹핑해서 수집
     - collect()는 컬렉션의 요소들을 그룹핑해서 Map 객체를 생성하는 기능도 제공

Map<Student.Sex, List<Student>> mapBySex = totalList.stream()
	.collect(Collectors.groupingBy(student :: getSex));

 
- 그룹핑 후 매핑 및 집계
     - 그룹핑 후 매핑, 집계를 할 수 있도록 두 번째 매개값으로 Collector를 가질 수 있음.

Map<Student.Sex, Double> mapBySex = totalList.stream()
	.collect(
    	Collectors.groupingBy{
        	Student :: getSex,
            Collectors.averagingDouble(Student :: getScore)
        }
    );

 

8. 병렬 처리
1) 병렬 처리 : 멀티 코어 CPU 환경에서 하나의 작업을 분할해서 각각의 코어가 병렬적으로 처리하는 것
2) 목적 : 작업 처리 시간 감소
3) 동시성(Concurrency)과 병렬성(Parallelism)
  - 동시성 : 멀티 작업을 위해 멀티 스레드가 번갈아가며 실행하는 성질
  - 병렬성 : 멀티 작업을 위해 멀티 코어를 이용해 동시에 실행하는 성질

4) 병렬성은 데이터 병렬성과 작업 병렬성로 구분할 수 있다.
  - 데이터 병렬성 : 전체 데이터를 쪼개어 서브 데이터로 만들고, 서브 데이터들을 병렬 처리해서 작업을 빠르게 끝내는 것
  → 자바8이 지원하는 병렬 스트림은 데이터 병렬성을 구현한 것
  → 멀티 코어의 수만큼 대용량 요소를 서브 요소로 나누고, 각각의 서브 요소들을 분리된 스레드에서 병렬처리 시킴
  - 작업 병렬성 : 서로 다른 작업을 병렬 처리하는 것
    → 예) 웹 서버, 각 브라우저에서 요청한 것을 개별 스레드에서 병렬로 처리

9. 포크조인(ForkJoin) 프레임워크
1) 병렬 스트림을 이용하면 런타임 시에 포크조인 프레임워크 동작
2) 포크 단계 : 전체 데이터를 서브 데이터로 분리한 후 서브 데이터를 멀티 코어에서 병렬로 처리
3) 조인 단계 : 서브 결과를 결합해서 최종 결과를 낸다.

  - 포크조인 프레임워크는 스레드풀인 ForkJoinPool을 제공.
  - 각 코어에서 서브 요소를 처리하는 것은 개별 스레드가 해야 하므로 스레드 관리 필요

10. 병렬 스트림 생성

1) parallelStream()은 컬렉션으로부터 병렬 스트림을 바로 리턴.
2) parallel()는 순차 처리 스트림을 병렬 처리 스트림으로 변환해서 리턴.
3 병렬 처리 성능
  - 요소의 수와 요소당 처리 시간
    → 컬렉션에 요소의 수가 적고 각 요소의 처리 시간이 짧으면 순차 처리가 더 빠를 수 있음.
    → 병렬 처리는 스레드풀 생성, 스레드 생성이라는 추가적인 비용 발생.
4) 스트림 소스의 종류
  - ArrayList, 배열은 인덱스로 요소를 관리하기 때문에 쉽게 분리할 수 있어 병렬 처리 시간이 절약된다.
  - HashSet, TreeSet, LinkedList는 요소 분리가 쉽지 않아 상대적으로 느리다.
  - 코어의 수 : 싱글 코어일 경우 순차 처리가 빠르다. 병렬 스트림 사용 시 스레드의 수만 증가하고 동시성 작업으로 처리되어 좋지 않다.

 

 

 

 

참고 : [한빛미디어] 이것이 자바다 (신용권의 Java 프로그래밍 정복) Chapter 16.스트림
참고 : [도우출판] JAVA의 정석(3ND EDITION)-자바의 정석 최신 Java 8.0 포함 Chapter 14.람다와 스트림