[Java] 스트림

✔ 스트림의 이해

스트림이란?

• 데이터를 흐름을 뜻함
  • 또한 배열 또는 컬렉션 인스턴스에 저장된 데이터를 꺼내서 파이프에 흘려보내는 것을 뜻함
  • 데이터를 흘려보낼 파이프인 연산의 종류는 중간 연산과 최종 연산이 존재
  • 스트림을 생성하고 이를 대상으로 중간 연산과 최종 연산을 진행하면
    원하는 기준으로 데이터를 필터링하고 필터링된 데이터의 가공된 결과를 얻을 수 있게 됨

int[] ar = {1, 2, 3, 4, 5};
IntStream stm1 = Arrays.stream(ar); // Arrays 클래스의 stream 메소드를 통해 배열을 대상으로 스트림 생성
IntStream stm2 = stm1.filter(n -> n % 2 == 1); // 중간 연산
int sum = stm2.sum(); // 최종 연산

스트림의 특성

• 스트림을 위한 stream과 filter 메소드는 IntStream의 인스턴스 메소드이므로 반환형은 IntStream
  • 따라서 스트림의 생성과 그에 따른 연산의 과정을 하나의 문장으로 완성할 수 있음

public static IntStream stream(int[] array)
IntStream filter(IntPredicate predicate)

int[] ar = {1, 2, 3, 4, 5};
int sum = Arrays.stream(ar)
                 .filter(n -> n % 2 == 1)
                 .sum();

✔ 스트림의 생성

스트림 생성하기

• 배열에 저장된 데이터를 대상으로 스트림을 생성할 때는 Arrays 클래스의 메소드인 stream을 호출
  • 배열의 일부분을 대상으로도 스트림을 생성할 수 있음
  • 그 외로도 기본 자료형의 값을 담고 있는 대상은 모두 스트림으로 생성 가능
  • forEach 메소드는 내부적으로 스트림의 데이터를 하나씩 인자로 전달하면서 accept 메소드를 호출함

public static IntStream stream(int[] array)
public static IntStream stream(int[] array, int startInclusive, int endExclusive)

public static DoubleStream stream(double[] array)
public static DoubleStream stream(double[] array, int startInclusive, int endExclusive)

public static LongStream stream(long[] array)
public static LongStream stream(long[] array, int startInclusive, int endExclusive)

String[] names = {"YOON", "LEE", "PARK"};
Stream<String> stm = Arrays.stream(names, 1, 2);
stm.forEach(s -> System.out.println(s));

• 컬렉션 인스턴스를 대상으로 스트림을 생성할 때는 Collection<E> 인터페이스의 디폴트 메소드인 stream를 호출

default Stream<E> stream()

List<String> list = Arrays.asList("Toy", "Robot", "Box");
list.stream()
    .forEach(s -> System.out.println(s + "\t"));

필요한 데이터를 직접 전달하여 스트림 생성하기

• Stream<T> 인터페이스의 of 메소드를 통해 스트림 생성에 필요한 데이터를 인자로 직접 전달할 수 있음
  • 생성된 스트림에는 하나의 인스턴스만 존재하게 됨
  • 그 외로도 기본 자료형의 데이터로 이뤄진 스트림을 생성 가능하며 불필요한 오토 박싱과 오토 언박싱을 피할 수 있음
  • 또한 범위 내에 있는 값들로 스트림을 구성할 수도 있음

static <T> Stream<T> of(T t)
static <T> Stream<T> of(T...values)

static IntStream of(int...values)
static IntStream of(int t)

static IntStream range(int startInclusive, int endExclusive) // 마지막 제외
static IntStream rangeClosed(int startInclusive, int endInclusive) // 마지막 포함

static DoubleStream of(double...values)
static DoubleStream of(double t)

static LongStream of(long...values)
static LongStream of(long t)

static LongStream range(long startInclusive, long endExclusive)
static LongStream range(long startInclusive, long endInclusive)

Stream.of(11, 22, 33, 44)
      .forEach(n -> System.out.println(n + "\t"));

List<String> list = Arrays.asList("Toy", "Robot", "Box");
Stream.of(list)
      .forEach(s -> System.out.println(s + "\t"));
      
IntStream is = IntStream.of(7, 5, 3);

병렬 스트림으로 변경

• 병렬 스트림은 하나의 작업을 둘 이상의 작업으로 나누어서 동시에 진행할 수 있음
• Stream<T> 인터페이스의 parallel 메소드를 통해 이미 스트림을 생성한 상태에서 이를 기반으로 병렬 스트림을 생성할 수 있음

Stream<T> parallel()

DoubleStream parallel()
IntStream parallel()
LongStream parallel()

List<String> list = Arrays.asList("Toy", "Robot", "Box");
Stream<String> ss = list.stream();

BinaryOperator<String> lc = (s1, s2) -> {
    if(s1.length() > s2.length()) 
        return s1;
    else
        return s2;
    };
    
String str = ss.parallel()
               .reduce("", lc);

스트림의 연결

• Stream<T> 인터페이스의 concat 메소드를 통해 두 개의 스트림을 연결하여 하나의 스트림을 생성할 수도 있음

static <T> Stream<T> concat(Stream<? extends T> a, Stream<? extends T> b)

static DoubleStream concat(DoubleStream a, DoubleStream b)
static IntStream concat(IntStream a, IntStream b)
static LongStream concat(LongStream a, LongStream b)

Stream<String> ss1 = Stream.of("Cake", "Milk");
Stream<String> ss2 = Stream.of("Lemon", "Jelly");

Stream.concat(ss1, ss2)
      .forEach(s -> System.out.println(s));

✔ 스트림의 중간 연산

필터링

• 스트림을 구성하는 데이터 중 일부를 조건에 따라 걸러내는데 사용하는 filter 메소드
  • 내부적으로 스트림의 데이터를 하나씩 인자로 전달하면서 test를 호출하고
    true가 반환되면 해당 데이터는 스트림에 남기고, false가 반환되면 해당 데이터는 버리게 됨
  • 스트림이 배열을 대상으로 생성되었건 컬렉션 인스턴스를 대상으로 생성되었건 모든 스트림에 적용 가능

Stream<T> filter(Predicate<? super T> predicate)
Predicate<T> boolean test(T t)

int[] ar = {1, 2, 3, 4, 5};
Arrays.stream(ar)
       .filter(n -> n % 2 == 1)
       .forEach(n -> System.out.println(n + "\t"));

맵핑

• 스트림을 기반으로 다른 스트림을 생성하는데 사용하는 map 메소드
  • 문자열의 길이를 맵핑의 기준으로 문자열 스트림을 숫자 스트림으로 맵핑하는 것 등에 사용
  • 맵핑을 진행하면 스트림의 데이터 형이 달라지는 특징을 가짐 
  • 기본 자료형의 값을 반환하는 경우를 고려하여 맵핑 관련 메소드를 제공하며 이 경우에는 오토 박싱이 진행되지 않음

<R> Stream<R> map(Function<? super T, ? extends R> mapper)
// <R> Stream<R> map(Function<T, R> mapper)
Function<T, R> R apply(T t)

IntStream mapToInt(ToIntFunction<? super T> mapper)
LongStream mapToLong(ToLongFunction<? super T> mapper)
DoubleStream mapToDouble(ToDoubleFunction<? super T> mapper)

List<String> list = Arrays.asList("Toy", "Robot", "Box");
list.stream()
    .map(s -> s.length())
    .forEach(n -> System.out.println(n + "\t"));
    
list.stream()
    .mapToInt(s -> s.length())
    .forEach(n -> System.out.println(n + "\t"));

• 또한 필터링 후 맵핑을 하는 등의 중간 연산을 두 번 진행할 수도 있음 

int sum = ls.stream()
            .filter(p -> p.getPrice() < 500)
            .mapToInt(t -> t.getPrice())
            .sum();

• 이 외에도 맵핑 관련 메소드의 종류는 다양함
  • flatMap 메소드에 전달할 람다식에서는 스트림을 생성하고 이를 반환해야 함
  • map 메소드에 전달할 람다식에서는 스트림을 구성할 데이터만 반환하면 됨
  • map은 일대 일 맵핑을 진행하지만 flatMap은 일대 다의 맵핑을 진행할 수 있음

<R> Stream<R> flatMap(Function<? super T, ? extends Stream<? stends R>> mapper)
// <R> Stream<R> flatMap(Function<T, Strema<R>> mapper)
Stream<R> apply(T t)

IntStream flatMapToInt(Function<? super T, ? extends IntStream> mapper)
LongStream flatMapToLong(Function<? super T, ? extends LongStream> mapper)
DoubleStream flatMapToDouble(Function<? super T, ? extends DoubleStream> mapper)

Stream<String> ss1 = Stream.of("MY_AGE", "YOUR_LIFE");
Stream<String> ss2 = ss1.flatMap(s -> Arrays.stream(s.split("_"))); // 두 개의 문자열이 네 개의 문자열로 나뉘게 됨
ss2.forEach(s -> System.out.println(s + "\t"));

IntStream si = sr.flatMapToInt(r -> IntStream.of(r.getKor(), r.getEng(), r.getMath()));

정렬

• 정렬 기능을 제공하는 sorted 메소드
  • sorted 메소드 정렬을 위해서는 스트림을 구성하는 인스턴스가 Comparable<T> 인터페이스를 구현하고 있어야 함

Stream<T> sorted(Comparator<? super T> comparator)
Stream<T> sorted()
IntStream sorted()
LongStream sorted()
DoubleStream sorted()

Stream.of("Box", "Apple", "Robot")
      .sorted()
      .forEach(s -> System.out.println(s + "\t"));
      
Stream.of("Box", "Apple", "Robot")
      .sorted((s1, s2) -> s1.length() - s2.length())
      .forEach(s -> System.out.println(s + "\t"));
      
IntStream.of(3, 9, 4, 2)
      .sorted()
      .forEach(d -> System.out.println(d + "\t"));

루핑

• 스트림을 이루는 모든 데이터 각각을 대상으로 특정 연산을 진행하는 행위인 루핑
  • 루핑 연산으로는 forEach가 존재하며 이는 최종 연산
  • 반면 중간 연산에도 루핑을 위한 peek 메소드가 존재

Stream<T> peek(Comparator<? super T> action)
IntStream peek(IntConsumer action)
LongStream peek(LongConsumer action)
DoubleStream peek(DoubleConsumer action)

IntStream.of(3, 9, 4, 2)
      .peek(d -> System.out.println(d + "\t"))
      .sum();

✔ 스트림의 최종 연산

리덕션과 reduce 메소드

• 데이터를 축소할 때 사용하는 reduce 메소드
  • sum의 경우 다수의 데이터를 더하여 합이라는 하나의 데이터를 남기므로 리덕션 연산
  • 다른 리덕션 연산은 연산의 내용이 이미 정해진 상태이지만 reduce 메소드는 전달하는 람다식에 의해
    apply를 호출하면서 스트림에 저장된 데이터를 줄여 나가게 되면서 하나의 데이터를 남기게 됨
  • 스트림이 비어 있지 않은 경우에는 첫 번째 인자로 전달된 값을 스트림의 첫 번째 데이터 간주하고 리덕션을 진행하며
    만약 스트림이 비어 있을 경우에는 첫 번째 인자로 전달된 값을 반환하게 됨

T reduce(T identity, BinaryOperator<T> accumulator)
BinaryOperator<T> T apply(T t1, T t2)

List<String> ls = Arrays.asList("Box", "Simple", "Complex", "Robot");

BinaryOperator<String> lc = (s1, s2) -> {
    if(s1.length() > s2.length())
        return s1;
    else
        return s2;
    };
    
String str = ls.stream()
               .reduce("", lc); // 스트림이 빈 경우 빈 문자열 반환

병렬 스트림

• 하나의 작업을 둘 이상의 작업으로 나누어서 동시에 진행할 수 있음
  • 이를 위해 parallelStream 메소드를 호출하면 이어지는 연산들이 CPU의 코어 수를 고려하여 적절하게 병렬로 처리하게 됨
  • 연산의 횟수는 동일하지만 둘 이상의 연산을 동시에 진행하기 때문에 연산의 단계가 줄어들어 속도 측면에서 이점

List<String> ls = Arrays.asList("Box", "Simple", "Complex", "Robot");

BinaryOperator<String> lc = (s1, s2) -> {
    if(s1.length() > s2.length())
        return s1;
    else
        return s2;
    };
    
String str = ls.parallelStream()
               .reduce("", lc);

그 외 다양한 최종 연산 메소드

• 합을 구하는 sum 메소드

int sum()
double sum()

int sum = IntStream.of(1, 3, 5, 7, 9)
                    .sum();

• 개수를 구하는 count 메소드

long count()

long sum = IntStream.of(1, 3, 5, 7, 9)
                    .count();

• 평균을 구하는 average 메소드

OptionalDouble average()

IntStream.of(1, 3, 5, 7, 9)
         .average()
         .ifPresent(av -> System.out.println("avg = " + av);

• 최소를 구하는 min 메소드

OptionalInt min()
OptionalDouble min()

IntStream.of(1, 3, 5, 7, 9)
         .min()
         .ifPresent(mn -> System.out.println("min = " + mn);

• 최대를 구하는 max 메소드

OptionalInt max()
OptionalDouble max()

IntStream.of(1, 3, 5, 7, 9)
         .max()
         .ifPresent(mx -> System.out.println("max = " + mx);

• 스트림을 이루는 모든 데이터 각각을 대상으로 특정 연산을 진행하는 forEach 메소드

void forEach(Comparator<? super T> action)
void forEach(IntConsumer action)
void forEach(LongConsumer action)
void forEach(DoubleConsumer action)

• 스트림의 데이터가 조건을 모두 만족하는 경우 true를 반환하는 allMatch 메소드

boolean allMath(Predicate<? super T> predicate)
boolean allMath(IntPredicate predicate)
boolean allMath(DoublePredicate predicate)

boolean b = IntStream.of(1, 2, 3, 4, 5)
                     .allMath(n -> n % 2 == 0);

• 스트림의 데이터가 조건을 하나라도 만족하는 경우 true를 반환하는 anyMatch 메소드

boolean anyMath(Predicate<? super T> predicate)
boolean anyMath(IntPredicate predicate)
boolean anyMath(DoublePredicate predicate)

boolean b = IntStream.of(1, 2, 3, 4, 5)
                     .anyMath(n -> n % 2 == 0);

• 스트림의 데이터가 조건을 하나도 만족하지 않는 경우 true를 반환하는 noneMath 메소드

boolean noneMath(Predicate<? super T> predicate)
boolean noneMath(IntPredicate predicate)
boolean noneMath(DoublePredicate predicate)

boolean b = IntStream.of(1, 2, 3, 4, 5)
                     .noneMath(n -> n % 2 == 0);

• 파이프라인을 통해서 가공되고 걸러진 데이터를 최종 연산 과정에서 별도로 저장하는 collect 메소드
  • 병렬 스트림을 대상으로 호출할 경우
    첫 번째 인자로 전달된 람다식을 기반으로 다수의 저장소가 생성되고
    두 번째 인자로 전달된 람다식을 기반으로 다수의 저장소에 데이터가 나뉘어 저장됨
    저장이 끝난 다음에는 세 번째 인자로 전달된 람다식을 기반으로 저장소에 담긴 데이터를 하나로 묶게 됨
  • 이러한 병렬 처리는 어히여 속도가 느려지는 경우도 많으므로 테스트를 통해 병렬 처리의 적합성을 판단

<R> R collect<Supplier<R> supplier, BiConsumer<R, ? super T> accumulator, BiConsumer<R, R> conbiner)
<R> R collect<Supplier<R> supplier, ObjIntConsumer<R> accumulator, BiConsumer<R, R> conbiner)
<R> R collect<Supplier<R> supplier, ObjLongConsumer<R> accumulator, BiConsumer<R, R> conbiner)
<R> R collect<Supplier<R> supplier, ObjDoubleConsumer<R> accumulator, BiConsumer<R, R> conbiner)

String[] words = {"Hello", "Box", "Robot", "Toy"};
Stream<String> ss = Arrays.stream(words);
List<String> ls = ss.filter(s -> s.length() < 5)
                    .collect(() -> new ArrayList<>(), // 저장소 생성
                             (c, s) -> c.add(s), // 컬렉션 인스턴스에 스트림의 데이터 저장
                             (lst1, lst2) -> lst1.addAll(lst2)); // 순차 스트림에서는 의미 없음
                             
List<String> ls = ss.parallel()
                    .filter(s -> s.length() < 5)
                    .collect(() -> new ArrayList<>(), // 다수의 저장소 생성
                             (c, s) -> c.add(s), // 다수의 저장소에 데이터가 나뉘어 저장
                             (lst1, lst2) -> lst1.addAll(lst2)); // 저장소에 담긴 데이터들을 하나로 묶음