이것이 자바다 15일차 - Chapter17 스트림 요소 처리

스트림이란?

지금까지 컬렉션 및 배열에 저장된 요소를 반복 처리하기 위해 for문이나 iterator를 썼다.
java8 부터는 또 다른 방법으로 컬렉션 및 배열의 요소를 반복 처리하기 위해 Stream을 사용할 수 있다. 스트림은 요소들이 하나씩 흘러가면서 처리된다는 의미를 가지고 있다.

Stream<String> stream = list.stream();
stream.forEach( item -> //처리 
				); 

package ch17.sec01.exam01;

import java.util.HashSet;
import java.util.Set;
import java.util.stream.Stream;

public class StreamExample {
    public static void main(String[] args) {
        Set<String> set = new HashSet<>();
        set.add("홍길동");
        set.add("신용권");
        set.add("김자바");

        Stream<String> stream = set.stream();
        stream.forEach(e->System.out.println(e));
    }
}

Stream은 Iterator와 비슷한 반복자이지만 차이점이 있다.

• 내부 반복자이므로 처리 속도가 빠르고 병렬 처리에 효율적이다.
• 람다식으로 다양한 요소 처리를 정의할 수 있다.
• 중간 처리와 최종 처리를 수행하도록 파이프 라인을 형성할 수 있다.

내부 반복자

for문과 Iterator는 컬렉션의 요소를 컬렉션 바깥쪽으로 반복해서 가져와 처리하는데, 이것을 외부 반복자라고 한다. 반면 스트림은 요소 처리 방법을 컬렉션 내부로 주입시켜 요소를 반복 처리하는데, 이것을 내부 반복자라고 한다.

외부 반복자일 경우 컬렉션의 요소를 외부로 가져오는 코드와 처리하는 코드를 모두 개발자 코드가 가지고 있어야 한다. 반면 내부 반복자일 경우는 개발자 코드에서 제공한 데이터 처리 코드(람다식)을 가지고 컬렉션 내부에서 요소를 반복 처리한다.

이는 멀티 코어 CPU를 최대한 활용하기 위해 요소들을 분배시켜 병렬 작업을 할 수 있다.
하나씩 처리하는 순차적 외부 반복자보다 효율적으로 요소를 반복시킬 수 있다는 장점이 있다.

package ch17.sec02;

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

public class ParallelStreamExample {
    public static void main(String[] args) {
        List<String> list = new ArrayList<>();
        list.add("홍길동");
        list.add("신용권");
        list.add("김자바");
        list.add("람다식");
        list.add("박병렬");

        Stream<String> parallelStream = list.parallelStream();
        parallelStream.forEach(e->{
            System.out.println(e+": "+Thread.currentThread().getName());
        });
    }
}

이렇게 중괄호를 넣어서 내부 반복자를 구현할 수 있다.

중간 처리와 최종 처리

스트림은 하나 이상 연결될 수 있다. 스트림에 또 다른 스트림을 이어지게 하는 것을 파이프라인 이라고 한다.

주로 필터링, 매핑 같은 데이터 정재 작업을 하고 마지막에 집계를 내는 식으로 쓴다.

package ch17.sec03;

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

public class StreamPipeLineExample {
    public static void main(String[] args) {
        List<Student> list = Arrays.asList(
                new Student("홍길동",10),
                new Student("신용권",20),
                new Student("유미선",30)
        );

        double avg = list.stream()
                .mapToInt(student -> student.getScore())
                .average()
                .getAsDouble();
        System.out.println(avg);
    }
}

리소스로부터 스트림 얻기

스트림은 기본적으로 BaseStream 인터페이스를 부모로 한 자식인터페이스들로 이루어진다.

Stream은 객체 요소를 처리하는 스트림, IntStream, DoubleStream, LongStream등 기본타입에 대한 스트림도 존재한다.

Array <-> stream

        String[] strArray = {"홍길동","신용권","김미나"};
        Stream<String> strStream = Arrays.stream(strArray);

String일 때는 String이 객체이므로 그냥 Stream을 써주면 되고

        int[] intArray = {1,2,3,4};
        IntStream intStream = Arrays.stream(intArray);

int 배열은 IntStream을 통해 변환 할 수 있다. 다른 자료형도 마찬가지.

package ch17.sec04.sec03;

import java.util.stream.IntStream;

public class StreamExample {
    public static int sum;

    public static void main(String[] args) {
        IntStream stream = IntStream.rangeClosed(1,100);
        stream.forEach(e->sum +=e);
        System.out.println(sum);
    }
}

이렇게 숫자 범위의 스트림을 얻을 수도 있다.

파일로부터 스트림 읽기

java.nio.file.Files.의 lines() 메소드를 이용하면 텍스트 파일의 행 단위 스트림을 읽을 수 있다.
data.txt

{pno: 1, name: 상품1, company: 멋진 회사, price: 2209}
{pno: 2, name: 상품2, company: 멋진 회사, price: 4272}
{pno: 3, name: 상품3, company: 멋진 회사, price: 787}
{pno: 4, name: 상품4, company: 멋진 회사, price: 5593}
{pno: 5, name: 상품5, company: 멋진 회사, price: 6645}

이런 식의 텍스트 파일이 존재한다고 치자.

package ch17.sec04.sec04;

import java.io.IOException;
import java.net.URISyntaxException;
import java.nio.charset.Charset;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.stream.Stream;

public class StreamExample {
    public static void main(String[] args) throws URISyntaxException, IOException {
        Path path = Paths.get(StreamExample.class.getResource("data.txt").toURI());
        Stream<String> stream = Files.lines(path, Charset.defaultCharset());
        stream.forEach(e->System.out.println(e));
        stream.close();
    }
}

이렇게 경로를 지정 후 파일 스트림을 읽을 수 있다.

요소 걸러내기

스트림에서 쓰는 필터링 메소드는 크게 distinct()와 filter가 있다.
disctinct()는 중복 제거, filter()는 특정 조건을 만족하는 요소만 스트림으로 만든다.

객체스트림일 경우 disctinct()의 동일비교는 equals()만으로 진행한다.

package ch17.sec05;

import java.util.ArrayList;
import java.util.List;

public class FilteringExample {
    public static void main(String[] args) {
        List<String> list = new ArrayList<>();
        list.add("홍길동");    list.add("신용권");
        list.add("신도라애몽"); list.add("신민철");
        list.add("김기동");    list.add("김기동");
        
        list.stream().distinct().forEach(e->System.out.println(e)); // 중복 제거
        
        list.stream().filter(e-> e.startsWith("신"))
        .forEach(e->System.out.println(e)); // 성씨가 신인 것만 
        
        list.stream()
        .distinct()
        .filter(e->e.startsWith("신"))
        .forEach(e->System.out.println(e)); // 파이프라인 
    }
}

이런 식으로 응용 할 수 있다.

요소 변환(매핑)

매핑 스트림의 요소를 다른 요소로 변환하는 기능이 있다.

package ch17.sec06.exam02;

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

public class MapExample {
    public static void main(String[] args) {
        int[] intArray = {1,2,3,4,5};

        IntStream intStream = Arrays.stream(intArray);
        intStream
                .asDoubleStream()
                .forEach(e-> System.out.println(e));

        System.out.println();

        intStream = Arrays.stream(intArray);
        intStream.boxed().forEach(obj -> System.out.println(obj.intValue()));
    }
}

요소 정렬

package ch17.sec07.exam01;


public class Student implements Comparable<Student> {
    private String name;
    private int score;

    public Student(String name, int score) {
        this.name = name;
        this.score = score;
    }
    public String getName() {return name;}
    public int getScore(){return score;}

    @Override
    public int compareTo(Student o){
        return Integer.compare(score, o.score);
    }
}

package ch17.sec07.exam01;

import java.util.ArrayList;
import java.util.Comparator;
import java.util.List;

public class SortingExample {
    public static void main(String[] args) {
        List<Student> studentList = new ArrayList<>();
        studentList.add(new Student("홍길동",30));
        studentList.add(new Student("신용권",10));
        studentList.add(new Student("유미선",20));

        studentList.stream()
                .sorted()
                .forEach(e->System.out.println(e.getName()+" : "+e.getScore()));
        System.out.println("");

        studentList.stream()
                .sorted(Comparator.reverseOrder())
                .forEach(e->System.out.println(e.getName()+" : "+e.getScore()));


    }
}

요소를 하나씩 처리 (루핑)

루핑은 스트림에서 요소를 하나씩 반복해서 가져와 처리하는 것을 말한다.
루핑 메소드에는 peek()과 forEach()가 있다.

package ch17.sec08;

import java.util.Arrays;

public class LoopingExample {
    public static void main(String[] args) {
        int[] intArr = {1,2,3,4,5};
        Arrays.stream(intArr)
                .filter(e->e%2==0)
                .peek(n->System.out.println(n)); // 최종 처리가 없으므로 작동하지 않음
        int total = Arrays.stream(intArr)
                .filter(e->e%2==0)
                .peek(n->System.out.println(n))
                .sum();
        System.out.println(total);

        Arrays.stream(intArr)
                .filter(e->e%2==0)
                .forEach(n->System.out.println(n));
    }
}

요소 조건 만족 여부(매칭)

package ch17.sec09;

import java.util.Arrays;

public class MatchingExample {
    public static void main(String[] args) {
        int[] intArr = {2,4,6};

        boolean result = Arrays.stream(intArr).allMatch(a ->a%2==0);// 짝수인 것만 존재하면 true
    }
}

allMatch() : 모든 요소가 조건을 만족해야함

        result = Arrays.stream(intArr).anyMatch(a -> a%3 ==0); // 하나라도 3의 배수가 존재하면 true
        System.out.println(result);

anyMatch() : 하나라도 조건을 만족하면 true

        result = Arrays.stream(intArr).noneMatch(a -> a%3==0); // 3의 배수가 하나도 존재하지 않으면 true
        System.out.println(result);

nonMatch() 단 하나도 만족하는 조건이 없으면 true

요소 기본 집계

스트림은 카운팅, 최대, 최소, 평균, 합계 등을 구할 수 있다.

package ch17.sec10;

import java.util.Arrays;

public class AggregateExample {
    public static void main(String[] args) {
        int[] arr = {1,2,3,4,5};

        long count = Arrays.stream(arr).filter(e -> e%2==0).count();
        long sum = Arrays.stream(arr).filter(e->e%2==0).sum();
        double avg = Arrays.stream(arr).filter(e->e%2==0).average().getAsDouble();
        int max = Arrays.stream(arr).filter(e -> e%2==0).max().getAsInt();
        int min = Arrays.stream(arr).filter(e -> e%2==0).min().getAsInt();
    }
}

요소 커스텀 집계

합, 평균, 최소, 최대 등 이것 말고도 다양한 집계 결과물을 볼 수 있도록 reduce(); 메소드를 지원한다.

reduce()는 스트림에 요소가 없을 경우 예외가 발생하지만 indentity 매개값이 주어지면 이 값을 디폴트 값으로 리턴한다. 다음 중 왼쪽 코드는 스트림에 요소가 없을 경우 NoSuchElementException을 발생시키지만, 기본값을 넣어주면 해당 디폴트값을 리턴한다.

int sum = stream.reduce(0, (a,b) -> a+b).getAsInt();

package ch17.sec11;

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

public class ReductionExample {
    public static void main(String[] args) {
        List<Student> studentList = Arrays.asList(
                new Student("홍길동",92),
                new Student("신용권",95),
                new Student("김자바",88)
        );
        int sum1 = studentList.stream()
                .mapToInt(Student::getScore).sum();

        int sum2 = studentList.stream()
                .map(Student::getScore)
                .reduce(0, (a,b)->a+b);

        System.out.println(sum1);
        System.out.println(sum2);
    }
}

요소 수집

package ch17.sec12.exam01;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

public class CollectExample {
    public static void main(String[] args) {
        List<Student> totalList = new ArrayList<>();
        totalList.add(new Student("홍길동","남",92));
        totalList.add(new Student("김수영","여",87));
        totalList.add(new Student("김자바","남",95));

        List<Student> maleList = totalList.stream().filter(e->e.getSex().equals("남")).toList();
        System.out.println();
        Map<String, Integer> map = totalList.stream()
                .collect(
                        Collectors.toMap(s ->s.getName(),s->s.getScore())
                );
        System.out.println(map);
    }
}

요소 그룹핑

package ch17.sec12.exam02;

import ch17.sec12.exam01.Student;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

public class CollectExample {
    public static void main(String[] args) {
        List<Student> totalList = new ArrayList<>();
        totalList.add(new Student("홍길동","남",92));
        totalList.add(new Student("김수영","여",87));
        totalList.add(new Student("김자바","남",95));

        Map<String, List<Student>> map = totalList.stream().collect(
                Collectors.groupingBy(s -> s.getSex())
        );

        List<Student> maleList = map.get("남");
        maleList.stream().forEach(s->System.out.println(s.getName()));
        System.out.println();

        List<Student> famaleList = map.get("여");
        famaleList.forEach(s->System.out.println(s.getName()));


    }
}

package ch17.sec12.exam03;

import ch17.sec12.exam01.Student;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

public class CollectExample {
    public static void main(String[] args) {
        List<ch17.sec12.exam01.Student> totalList = new ArrayList<>();
        totalList.add(new ch17.sec12.exam01.Student("홍길동","남",92));
        totalList.add(new ch17.sec12.exam01.Student("김수영","여",87));
        totalList.add(new Student("김자바","남",95));

        Map<String,Double> map = totalList.stream()
                .collect(
                        Collectors.groupingBy(
                                s->s.getSex(),
                                Collectors.averagingDouble(s->s.getScore())
                        )
                );
        System.out.println(map);
    }
}

요소 병렬 처리

요소 병렬 처리란 멀티 코어 CPU 환경에서 전체 요소를 분할해서 각각의 코어가 병렬적으로 처리하는 것을 말한다. 요소 병렬 처리의 목적은 작업 처리 시간을 줄이기 위한 것에 있따. 자바를 요소 병렬 처리를 위해 병렬 스트림을 제공한다.

동시성과 병렬성

멀티 스레드는 동시성 또는 병렬성으로 실행된다. 동시성을 멀티 작업을 위해 멀티 스레드가 하나의 코어에서 번갈아 가며 실행하는 것을 말하고, 병렬성은 멀티 작업을 위해 멀티 코어를 각각 이용해서 병렬로 실행하는 것을 말한다.

동시성은 한 시점에 하나의 작업만 실행한다. 번갈아 작업을 실행하는 것이 워낙 빠르다보니 동시에 처리되는 것 처럼 보일 뿐이다. 병렬성은 한 시점에 여러 개의 작업을 병렬로 실행하기 때문에 동시성보다 좋은 성능을 낸다.

데이터 병렬성
데이터 병렬성은 전체 데이터를 분할해서 서브 데이터셋으로 만들고 이 서브 데이터셋들을 병렬 처리해서 작업을 빨리 끝내는 것을 말한다. 자바 병렬 스트림은 데이터 병렬성을 구현한 것이다.

작업 병렬성
작업 병렬성은 서로 다른 작업을 병렬 처리하는 것을 말한다. 작업 병렬성의 대표적인 예는 서버 프로그램이다. 서버는 각각의 클라이언트에서 요청한 내용을 개별 스레드에서 병렬로 처리한다.

포크조인 프레임워크

자바 병렬 스트림은 요소들을 병렬 처리하기 위해 포크조인 프레임워크를 사용한다. 포크조인 프레임워크는 포크 단계에서 전체 요소들을 서브 요소셋으로 분할하고, 각각의 서브 요소셋을 멀티 코어에서 병렬로 처리한다. 조인 단계에서는 서브 결과를 결합해서 최종 결과를 만들어낸다.
예를 들어 쿼드 코어 CPU에서 병렬 스트림으로 요소들을 처리할 경우 먼저 포크 단계에서 스트림의 전체 요소들을 4개의 서버 요소셋으로 분할한다. 그리고 각각의 서브 요소셋을 개별 코어에서 처리하고, 조인 단게에서는 3번의 결합 과정을 거쳐 최종 결과를 산출한다.

병렬 스트림 사용

자바에서는 스트림을 쓰면 쉽게 병렬처리를 할 수 있다.

parallelStream()과 parallel()을 이용해 자동으로 포크조인 프레임워크를 쓰는 기능을 사용할 수 있다.

package ch17.sec13;

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

public class ParallelExample {
    public static void main(String[] args) {
        Random random = new Random();
        List<Integer> scores = new ArrayList<>();
        for(int i = 0; i<100000000; i++){
            scores.add(random.nextInt(101));
        }
        double avg = 0.0;
        long startTime = 0;
        long endTime = 0;
        Stream<Integer> stream = scores.stream();
        startTime = System.nanoTime();
        avg = stream.mapToInt(i->i.intValue()).average().getAsDouble();
        endTime = System.nanoTime();
        System.out.println("일반 스트림 처리 시간 : "+(endTime-startTime)+"ns");

        Stream<Integer> parallelStream = scores.parallelStream();
        startTime = System.nanoTime();
        avg = parallelStream.mapToInt(i->i.intValue()).average().getAsDouble();
        endTime = System.nanoTime();
        System.out.println("병렬 스트림 처리 시간 : "+(endTime-startTime)+"ns");
    }
}

일반 스트림 처리 시간 : 106322100ns
병렬 스트림 처리 시간 : 48725000ns

Process finished with exit code 0

속도 차이는 엄청나다.

연습문제

4번 요소를 모두 처리하고 나면 스트림이 끝납니다.

2번, 범위를 배열로 바꿔줘야 합니다.

4번 중간처리는 최종 처리가 있을 때만 가능합니다.

3번 항상 빠른건 아닙니다.

.filter(e->e.toLowerCase().contains("java").forEach(e->System.out.println(e));

.mapToInt(Member::getAge).average().getAsDouble();

.filter(m->m.getJob().equals("개발자")).collect(Collectors.toList());

.collect(Collectors.groupingBy(m -> m.getJob()));

groupingMap.get("개발자").stream()
 .forEach(m -> System.out.println(m));

groupingMap.get("디자이너").stream()
 .forEach(m -> System.out.println(m));