1. 스트림
반복자로 컬렉션의 요소를 하나씩 참조해 람다식으로 처리할 수 있는 반복자
//iterator
List<String> list = Arrays.asList("a", "b", "c");
Iterator<String> iterator = list.iterator();
while(iterator.hasNext()){
String name = iterator.next();
System.out.println(name);
}
//stream
List<String> list = Arrays.asList("a", "b", "c");
Stream<String> stream = list.stream();
stream.forEach(name -> System.out.println(nama);스트림 특징
- 람다식으로 요소 처리 코드를 제공하며 대부분의 요소 처리 메소드는 함수적 인터페이스 매개타입을 가지며 매개값으로 람다식 또는 메소드 참조 대입 가능함
- 내부 반복자를 사용해 병렬 처리가 쉬움
- 내부 반복자
코드로 직접 컬렉션 요소 반복해 요청하고 가져오는 코드 패턴 - 외부 반복자
컬렉션 내부에서 자동으로 요소 반복시키고 개발자는 요소당 처리해야하는 코드만 제공하는 코드 패턴 - 내부 반복자의 이점
개발자는 요소 처리 코드에만 집중할 수 있고 멀티 코어 CPU 최대한 활용하기 위해 요소들을 분배하여 병렬 처리 작업 진행 가능 - 병렬 처리
한가지 작업을 서브 작업으로 나누고 서브 작업들을 분리된 스레드에서 병렬적으로 처리한 후, 서브 작업들의 결과들을 최종 결합하는 방법으로 자바에서는 ForkJoinPool 프레임워크 이용해 병렬 처리 가능
- 내부 반복자
- 중간 처리(요소들의 매핑, 필터링, 정렬) 및 최종 처리(반복, 카운트, 평균, 총합) 가능
병렬 처리 코드 예시
//순차처리
Stream<String> stream = list.stream();
stream.forEach(ParallelExample::print);
//병렬처리
Stream<String> parallelStream = list.parallelStream();
parallelStream.forEach(ParallelExample::print);2. 스트림 종류
java.util.stream
BaseStream : 모든 스트림에서 사용할 수 있는 공통 메소드들 정의됨
Stream-IntStream-LongStream-DoubleStream
컬렉션 to 스트림
List<Student> studentList = Arrays.asList(new Studnet("a"), new Student("b"), new Student("c"));
Stream<Student> stream = studentList.stream();
stream.forEash(s -> System.out.println(s.getName()));배열 to 스트림
String[] strArray = {"a", "b", "c"};
Stream<String> strStream = Arrays.stream(strArray);
strStream.forEach(a -> System.out.println(a + ","));숫자 범위 to 스트림
IntStream stream = IntStream.rangeClosed(1, 100);
stream.forEach(a -> sum += a);
System.out.println(sum);파일 to 스트림
Path path = Paths.get("src/sec02/stream_kind/linedata.txt");
Stream<String> stream;
stream = Files.lines(path, Charset.defaultCharset());
stream.forEach(System.out::println);
File file = path.toFile();
FileReader fileReader = new FileReader(file);
BufferedReader br = new BufferedReader(fileReader);
stream = br.lines();
stream.forEach(System.out::println);디렉토리 to 스트림
Path path = Paths.get("C:/JavaProgramming/source");
Stream<Path> stream = Files.list(path);
stream.forEach(p -> System.out.println(p.getFileName());3. 스트림 파이프라인
중간 처리와 최종 처리
리덕션
대량 데이터 가공해서 축소하는 작업으로 합계, 평균값, 카운팅, 최대값, 최소값 등을 집계하는 것을 의미하며 요소가 리덕션의 결과물로 바로 집계할 수 없을 경우 중간 처리(필터링, 매핑, 정렬, 그룹핑)가 필요함
파이프라인
스트림은 파이프라인을 통해 여러 중간 처리 메소드들을 연결하여 중간 처리와 최종처리를 연결, 최종 스트림의 집계 기능이 시작되기 전까지 중간 처리는 지연되어진다
중간 처리 메소드, 최종 처리 메소드
중간처리 메소드
리턴 타입이 스트림
- 필터링 : distinct(), filter(...)
- 매핑 : flatMap(...), flatMapToDouble(...), flatMapToInt(...), flatMapToLong(...), map(...), mapToDouble(...), mapToInt(...), mapToLong(...), mapToObj(...), asDoubleStream(), asLongStream(), boxed()
- 정렬 : sorted(...)
- 루핑 : peek(...)
최종 처리 메소드
리턴 타입이 기본 타입 or OptionalXXX
- 매칭 : allMatch(...), anyMatch(...), nonMatch(...)
- 집계 : count(), findFirst(), max(...), min(...), average(), reduce(...), sum()
- 루핑 : forEach(...)
- 수집 : collect(...)
4. 필터링
distinct() - 중복 제거
Stream에서는 equals()가 true인 경우 동일 객체로 판단해 중복 제거
IntStream, LongStream, DoubleStream에서는 동일값일 경우 중복 제거
filter() - 조건 필터핑
매개값으로 주어진 Predicate가 true 리턴하는 요소만 필터링
5. 매핑
스트림의 요소를 다른 요소로 대체하는 역할을 함
flatMapXXX()
한 개의 요소를 대체하는 복수 개의 요소들로 구성된 새로운 스트림 리턴
map()
요소를 대체하는 한 개의 요소로 구성된 새로운 스트림 리턴
asDoubleStream()
IntStream/LongStream의 int, long 요소를 double 요소로 타입 변환
asLongStream()
IntStream의 int 요소를 long 요소로 타입 변환해 LongStream 생성
boxed()
int, long, double 요소를 Integer, Long, Double 요소로 박싱해서 Stream 생성
6. 정렬
요소를 정렬
객체 요소일 경우 객체에 Comparable을 구현하지 않으면 sorted() 메소드 호출 시 ClassCastException 발생
Comparable 구현되지 않은 객체 요소나 다른 비교 방법으로 정렬하고 싶은 경우 sorted(Comparator<T>) 메소드 이용
7. 루핑
요소 전체를 반복하는 것으로 중간/최종 처리 둘 다 사용
중간 처리에서는 peek(), 최종 처리에서는 forEach()
8. 매칭
최종 처리 기능으로 요소들이 특정 조건 만족하는지 조사
allMatch
모든 요소들이 매개값으로 주어진 Predicate 조건 만족하는지 조사
anyMatch
최소 한 개 요소가 매개값으로 주어진 Predicate 조건 만족하는 지 조사
noneMatch
모든 요소들이 매개값으로 주어진 Predicate 조건 만족하지 않는지를 조사
9. 기본 집계
최종 처리 기능으로 카운팅, 합계, 평균값, 최대값, 최소값 등과 같이 하나의 값으로 산출하여 대량의 데이터를 가공해 축소하는 리덕션
Optional 클래스
값을 저장하는 값 기반 클래스로 집계값이 존재하지 않을 경우 디폴트값 설정가능하며 집계값 처리하는 Consumer 등록 가능
isPresent()
값이 저장되어 있는지 여부
orElse(T)/orElse(double)/orElse(int)/orElse(long)
값이 저장되지 않은 경우 디폴트값 설정
ifPresent(Consumer)/ifPresent(DoubleConsumer)/ifPresent(IntConsumer)/ifPresent(LongConsumer)
값이 저장되어 있을 경우 Comsumer로 집계 처리
10. 커스텀 집계
reduce() 로 다양한 집계 결과물 만들 수 있음
//reduce 예시
int sum = studentList.stream()
.map(Student::getScore)
.reduce((a,b) -> a+b)
.get();11. 수집
최종 처리 기능으로 요소들을 수집/그룹핑
필터링/매핑된 요소들로 구성된 새로운 컬렉션을 생성하고 요소들을 그룹핑하고 집계할 수 있음
필터링한 요소 수집
collector(Collector<T,A,R> collector)
T : 요소
A : 누적기
R : 요소가 저장될 새로운 컬렉션
Collectors 클래스의 정적 메소드 이용해 필터링 요소 수집 가능
Collectors.toCollection(Supplier<T>), Collectors.toConcurrentMap(...), Collectors.toList(), Collectors.toMap(...), Collectors.toSet()
List<Student> maleList = totalList.stream()
.filter(s -> s.getSex() == Student.Sex.MALE)
.collect(Collectors.toList());
Set<Student> femaleSet = totalList.stream()
.filter(s -> s.getSex() == Student.Sex.FEMALE)
.collect(Collectors.toCollections(HashSet::new));사용자 정의 컨테이너에 수집
collect(Supplier<R>, BiConsumer<R, ? super T>, BiConsumer<R, R>)
collect(Supplier<R>, ObjIntConsumer<R>, BiConsumer<R, R>)
collect(Supplier<R>, ObjLongConsumer<R, ? super T>, BiConsumer<R, R>)
collect(Supplier<R>, ObjDoubleConsumer<R, ? super T>, BiConsumer<R, R>)
Supplier : 요소들이 수집될 컨테이너 객체 생성하는 역할로 순차 처리 스트림에서는 단 한번, 병렬 처리 스트림에서는 스레드별로 컨테이너 생성
Consumer : 컨테이너 객체에 요소 수집하는 역할로 스트림에서 요소를 컨테이너에 누적할 때마다 실행
BiConsumer : 컨테이너 객체 결합하는 역할로 순차 처리 스트림에서는 실행되지 않지만 병렬 처리 스트림에서는 스레드 별로 생성된 컨테이너 결합해 최종 컨테이너 생성
요소 그룹핑 수집
collect() 통해 컬렉션 요소를 그룹핑해 Map 객체 생성 가능
Collectors.groupingBy() / Collectors.groupingByConcurrent()의 리턴 객체를 매개값으로 대입
Map<Student.Sex, List<Student>> mapBySex = totalList.stream()
.collect(Collectors.groupingBy(Student::getSex));
Map<Student.City, List<String>> mapByCity = totalList.stream()
.collect(
Collectors.groupingBy(
Student::getCity,
Collectors.mapping(Student::getName, Collectors.toList())
)
);그룹핑 후 매핑 및 집계
Collectors.groupingBy() 메소드는 그룹핑 후 매핑/집계 할 수 있도록 하기 위해 두번째 매개값으로 Collectors 메소드 대입 가능
mapping/averagingDouble/counting/joining/maxBy/minBy/reducing/summarizingXXX
Map<Student.Sex, Double> map = totalList.stream()
.collect(
Collectors.groupingBy(
Student::getSex,
Collectors.averagingDouble(Student::getScore)
)
);12. 병렬 처리
멀티 코어 CPU 환경에서 하나의 작업을 분할해 각각의 코어가 병렬적으로 처리하는 것으로 병렬처리의 목적은 작업 처리 시간을 줄이는 것
동시성
멀티 스레드 환경에서 스레드가 번갈아가며 실행하는 성질
병렬성
멀티 스레드 환경에서 코어들이 스레드를 병렬적으로 실행하는 성질
데이터 병렬성
한 작업 내에 있는 전체 데이터를 쪼개어 서브 데이터들로 만들고 이 서브 데이터들을 병렬 처리해서 작업을 빨리 끝내는 것
작업 병렬성
서로 다른 직업을 병렬 처리하는 것으로 웹서버와 같이 각 브라우저에서 요청한 내용을 개별 스레드에서 병렬로 처리하는 것
병렬 스트림
데이터 병렬성을 구현한 것으로 멀티 코어 수만큼 대용량 요소를 서브 요소들로 나누고 각각의 서브 요소들을 분리된 스레드에서 병렬 처리 시키며, ForkJoin 프레임워크 이용함
쿼드 코어의 경우 4개의 서브 요소들로 나눈 다음 4개의 스레드가 각각의 서브 요소들을 병렬 처리하게 됨
ForkJoin 프레임워크
Fork
데이터를 서브 데이터로 반복적으로 분리하여 서브 데이터를 멀티 코어에서 병렬로 처리
Join
서브 결과를 결합해 최종 결과 생성
ForkJoinPool
각각의 코어에서 서브 요소를 처리하는 것은 개별 스레드가 해야하므로 스레드 관리가 필요하며, ForkJoinPool을 통해 관리함
병렬 스트림 생성
parallelStream()
컬렉션으로부터 병렬 스트림 바로 리턴
parallel()
순환 처리 스트림을 병렬 스트림으로 변환해 리턴
//순차 처리 스트림
MaleStudent maleStudent = totalList.stream()
.filter(s -> s.getSex() == Student.Sex.MALE)
.collect(MaleStudent::new, MaleStudent::accumulate, MaleStudent::combine);
//병렬 스트림
MaleStudent maleStudent = totalList.parallelStream()
.filter(s -> s.getSex() == Student.Sex.MALE)
.collect(MaleStudent::new, MaleStudent::accumulate, MaleStudent::combine);<순차>
MaleStudent 객체는 단 하나 생성되어 accumulate 호출되면서 MaleStudent 객체 내부에 누적되어 combine은 호출되지 않음
<병렬>
코어 개수만큼 전체 요소는 서브 요소로 나뉘고, 해당 개수만큼 스레드 생성,
각 스레드는 서브 요소 수집하기 위해 MaleStudent 객체 생성하기 때문에 코어 수만큼 생성자가 실행되며 accumulate 또한 코어 수만큼 실행, 수집 완료된 MaleStudent는 최종 MaleStudent 로 만들어야하기 떄문에 combine은 단 한번만 싱행됨
병렬 처리 성능
스트림 병렬 처리가 항상 스트림 순차 처리보다 성능이 좋다고는 하지 못함
병렬 처리에 영향 미치는 요인
- 요소의 수와 요소당 처리 시간
- 스트림 소스의 종류
- 코어의 수
