스트림 처리 메소드 1편

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

외부 반복자란 개발자가 코드로 직접 컬렉션의 요소를 반복해서 가져오는 코드 패턴을 말합니다. index를 이용하는 for문 그리고 Iterator를 이용하는 while문은 모두 외부 반복자를 이용하는 것입니다. 반면에 내부 반복자는 컬렉션 내부에서 요소들을 반복시키고, 개발자는 요소당 처리해야 할 코드만 제공하는 코드 패턴을 말합니다. 아래 그림을 보면 알수 있습니다.

외부 반복자와 내부반복자 그림

내부 반복자를 사용해서 얻는 이점은 컬렉션 내부에서 어떻게 요소를 반복시킬 것인가는 컬렉션에 맡겨두고, 개발자는 요소 처리 코드에만 집중할 수 있다는 것입니다. 내부 반복자는 요소들의 반복 순서를 변경하거나, 멀티 코어 CPU를 최대한 활용하기 위해 요소들을 분배시켜 병렬 작업을 할 수 있게 도와주기 때문에 하나씩 처리하는 순차적 외부 반복자보다는 효율적으로 요소를 반복시킬 수 있습니다.

병렬 처리란?

병렬 처리란 한 가지 작업을 서브 작업으로 나누고, 서브 작업들을 분리된 스레드에서 병렬적으로 처리하는 것을 말합니다. 병렬 처리 스트림을 이용하면 런타임 시 하나의 작업을 서브 작업으로 자동으로 나누고, 서브 작업의 결과를 자동으로 결합해서 최종 결과물을 생성합니다. 예를들어 컬렉션 요소 총합을 구할 때 순차 처리 스트림은 하나의 스레드가 요소들을 순차적으로 읽어 합을 구하지만, 병렬 처리 스트림을 이용하면 여러 개의 스레드가 요소들을 부분적으로 합하고 이 부분합을 최종 결합해서 전체 합을 생성합니다.

아래 코드는 순차 처리 스트림과 병렬 처리 스트림을 이용할 경우, 사용된 스레드의 이름이 무엇인지 콘솔에서 출력하는 예제입니다. 실행결과를 보면 병렬 처리 스트림은 main 스레드를 포함해서 ForkJoinPool(스레드 풀)의 작업 스레드들이 병렬적으로 요소를 처리하는 것을 볼 수 있습니다.

//병렬처리
import java.util.Arrays;
import java.util.List;
import java.util.stream.Stream;

public class ParallelExample {

    public static void main(String[] args) {
        List<String> list = Arrays.asList(
                "홍길동", "신용권", "김자바"
                ,"람다식","박병렬"
        );
        // 순차 처리
        Stream<String> stream = list.stream();
        stream.forEach(ParallelExample :: print);
        System.out.println();

        // 병렬 처리
        Stream<String> parallelStream = list.parallelStream();
        parallelStream.forEach(ParallelExample::print);

    }

    private static void print(String str) {
        System.out.println(str + " : " + Thread.currentThread().getName());
    }
}

실행결과

스트림은 중간 처리와 최종 처리를 할 수 있습니다.

스트림은 컬렉션 요소에 대해 중간 처리와 최종 처리를 수행할 수 있는데, 중간 처리에서는 매핑, 필터링, 정렬을 수행하고 최종 처리에서는 반복, 카운팅, 평균, 총합 등의 집계
처리를 수행합니다.

아래 예제는 List에 저장되어 있는 Student 객체를 중간 처리에서 score 필드 값으로 매핑하고, 최종 처리에서 score 평균값을 산출합니다.

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

public class MapAndReduceExample {
    public static void main(String[] args) {

        List<Student> students = Arrays.asList(
                new Student("임준영",100),
                new Student("배성탑",90),
                new Student("임광빈",80),
                new Student("양아름",75)
        );

        double avg = students.stream()
                .mapToInt(Student::getScore) // 중간처리(학생 객체를 점수로 매핑)
                .average()
                .getAsDouble();

        System.out.println("평균 점수: " + avg);
    }
}

스트림의 종류

자바 8부터 새로 추가된 java.util.stream 패키지에는 스트림 API들이 포진하고 있습니다. 패키지 내용을 보면 BaseStream 인터페이스를 부모로 해서 자식 인터페이스들이 아래 이미지처럼 상속 관계를 이루고 있습니다.

BaseStream 인터페이스에는 모든 스트림에서 사용할 수 있는 공통 메소드들이 정의되어 있을 뿐 코드에서 직접적으로 사용되지는 않습니다. 하위 스트림인 Stream, IntStream,
LongStream, DoubleStream이 직접적으로 이용되는 스트림인데, Stream은 객체 요소를 처리하는 스트림이고, IntStream,LongStream, DoubleStream은 각각 기본 타입인 int, long, double 요소를 처리하는 스트림입니다. 이 스트림 인터페이스의 구현 객체는 다양한 소스로부터 얻을 수 있습니다. 주로 컬렉션과 배열에서 얻지만, 다음과 같은 소스로부터 스트림 구현 객체를 얻을 수도 있습니다.

컬렉션으로부터 스트림 얻기

다음 예제는 List 컬렉션에서 Stream를 얻어내고 요소를 콘솔에 출력합니다.

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

public class FromCollectionExample {
    public static void main(String[] args) {

        List<Student> students = Arrays.asList(
                new Student("임준영",100),
                new Student("배성탑",90),
                new Student("임광빈",80)
        );

        students.forEach(s -> System.out.println(s.getName()));

    }
}


package stream;

public class 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;
    }

}

숫자 범위부터 스트림 얻기

아래 예제 코드는 1부터 100까지의 합을 구하기 위해 IntStream의 rangeClosed() 메소드를 이용하였습니다. rangeClosed()는 첫 번째 매개값에서부터 두 번째 매개값까지 순차적으로 제공하는 IntStream을 리턴한다. IntStream의 또 다른 range() 메소드도 동일한 IntStream을 리턴하는데, 두 번째 매개값은 포함하지 않습니다.

import java.util.stream.IntStream;

public class FromIntRangeExample {

    public static int sum;

    public static void main(String[] args) {

        IntStream intStream = IntStream.rangeClosed(1,100);
        intStream.forEach(a -> sum += a);
        System.out.println("총합 :" +  sum);

    }
}

파일로부터 스트림 얻기

아래 예제 코드는 Files의 정적 메소드인 lines()와 BufferedReader의 lines() 메소드를 이용하여 문자 파일의 내용을 스트림을 통해 행 단위로 읽고 콘솔에 출력합니다.

import java.io.BufferedReader;
import java.io.File;
import java.io.FileReader;
import java.io.IOException;
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 FromFileContentExample {

    public static void main(String[] args) throws IOException {

        // 파일의 경로 정보를 가지고 있는 Path 객체 생성
        Path path = Paths.get("/Users/limjun-young/workspace/privacy/linedata.txt");
        Stream<String> stream;

        //Files.Line() 메소드 이용
        stream = Files.lines(path, Charset.defaultCharset());
        stream.forEach(System.out :: println);
        System.out.println();

        //BufferedReader의 lines() 메소드 이용
        File file = path.toFile();
        FileReader fileReader = new FileReader(file);
        BufferedReader bufferedReader = new BufferedReader(fileReader);
        stream = bufferedReader.lines();
        stream.forEach(System.out :: println);

    }

}

실행 결과

디렉토리부터 스트림 얻기

import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.stream.Stream;

public class FromDirectoryExample {
    public static void main(String[] args) throws IOException {
        Path path = Paths.get("/Users/limjun-young/workspace/privacy");
        Stream<Path> stream = Files.list(path);
        stream.forEach(p -> System.out.println(p.getFileName()));

    }
}

실행 결과

스트림 파이프라인

대량의 데이터를 가공해서 축소하는 것을 일반적으로 리덕션이라고 합니다. 데이의 합계, 평균값, 카운팅, 최대값, 최소값 등이 대표적인 리덕션의 결과물이라고 볼 수 있습니다. 그러나 컬렉션의 요소를 리덕션의 결과물로 바로 집계할 수 없을 경우에는 집계하기 좋도록 필터링, 매핑, 정렬, 그룹핑 등의 중간 처리가 필요합니다.

중간 스트림이 생성될 때 요소들이 바로 중간처리(필터링, 매핑, 정렬)되는 것이 아니라 최종 처리가 시작되기 전까지 중간 처리는 지연 됩니다. 최종 처리가 시작되면 비로소 컬렉션 요소가 하나씩 중간 스트림에서 처리되고 최종 처리까지 오게 됩니다.

Stream 인터페이스에는 필터링, 매핑, 정렬 등의 많은 중간 처리 메소드가 있습니다. 이 메소드들은 중간 처리된 스트림을 리턴합니다. 그리고 이 스트림에서 다시 중간 처리 메소드를 호출해서 파이프라인을 형성하게 됩니다. 예를 들어 회원 컬렉션에서 남자만 필터링하는 중간 스트림을 연결하고, 다시 남자의 나이로 매핑하는 스트림을 연결한 후, 최종 남자 평균 나이를 집계한 다면 다음 그림처럼 파이프라인이 형성됩니다.

// 스트림 파이프 라인
import java.util.Arrays;
import java.util.List;

public class StreamPipelinesExample {

    public static void main(String[] args) {
        List<Member> members = Arrays.asList(
                new Member("홍길동", Member.MALE, 30),
                new Member("김나리", Member.FEMALE, 20),
                new Member("신용권", Member.MALE, 45),
                new Member("박수미", Member.FEMALE, 27)
        );

        double ageAvg = members.stream()  
                .filter(m -> m.getSex() == Member.MALE) 
                .mapToInt(Member::getAge)
                .average()
                .getAsDouble();

        System.out.println("남자 평균 나이: " + ageAvg);
    }
}


// 회원 클래스
public class Member {

    public static int MALE = 0;
    public static int FEMALE = 1;

    private String name;
    private int sex;
    private int age;

    public Member(String name, int sex, int age) {
        this.name = name;
        this.sex = sex;
        this.age = age;
    }

    public int getSex() {
        return sex;
    }

    public int getAge() {
        return age;
    }
}

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

filter(m -> m.getSex() == Member.MALE)는 남자 Member 객체를 요소로 하는 새로운 스트림을 생성합니다. mapToInt(Member :: getAge())는 Member 객체를 age 값으로 매핑해서 age를 요소로 하는 새로운 스트림을 생성합니다. average() 메소드는 age 요소들의 평균을 OptionalDuble에 저장합니다. OptionalDouble에서 저장된 평균값을 읽으려면 getAsDouble() 메소드를 호출하면 됩니다.

중간 처리 메소드와 최종 처리 메소드를 쉽게 구분하는 방법은 리턴 타입을 보면 됩니다. 리턴 타입이 스트림이면 중간 처리 메소드이고, 기본 타입이거나 OptionalXXX라면 최종 처리 메소드 입니다. 소속된 인터페이스에서 공통의 의미는 Stream, IntStream, LongStream, DoubleStream에서 모두 제공된다는 뜻입니다.

필터링(distinct(), filter())

필터링은 중간 처리 기능으로 요소를 걸러내는 역할을 합니다. 필터링 메소드인 distinct()와 filter()메소드는 모든 스트림이 가지고 있는 공통 메소드 입니다.

distinct() 메소드는 중복을 제거하는데, Stream의 경우 Object.equals(Object)가 true이면 동일한 객체로 판단하고 중복을 제거합니다. IntStream, LongStream, DoubleStream은 동일값일 경우 중복을 제거합니다.

아래 예제는 이름 List에서 중복된 이름을 제거하고 출력합니다. 그리고 성이 "신"인 이름만 필터링해 서 출력합니다.

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

public class FilteringExample {

    public static void main(String[] args) {
        List<String> names = Arrays.asList(
                "홍길동", "신용권", "김자바", "신용권", "신민철"
        );

        names.stream()
                .distinct()
                .filter(s -> s.startsWith("신"))
                .forEach(System.out::println);
    }

}

매핑(flatMapXXX(), mapXXX(), asXXXStream(), boxed())

매핑은 중간 처리 기능으로 스트림의 요소를 다른 요소로 대체하는 작업을 말합니다.
스트림에서 제공하는 매핑 메소드는 flatXXX()와 mapXXX(),그리고 asDoubleStream(),asLongStream(), boxed()가 있습니다.

flatMapXXX() 메소드

flatMapXXX() 메소드는 요소를 대체하는 복수 개의 요소들로 구성된 새로운 스트림을 리턴합니다.

아래 예제는 입력된 데이터들이 List에 저장되어 있다고 가정하고, 요소별로 단어를 뽑아 단어 스트림으로 재생성합니다. 만약 입력된 데이터들이 숫자라면 숫자를 뽑아 숫자 스트림으로 재생성합니다.

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

public class FlatMapExample {
    public static void main(String[] args) {
        List<String> inputList1 = Arrays.asList("java8 lamda", "stream mapping");


        inputList1.stream()
                  .flatMap(data -> Arrays.stream(data.split(" ")))
                  .forEach(System.out :: println);


        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; i < strArr.length; i++) {
                          intArr[i] = Integer.parseInt(strArr[i].trim());
                      }
                      return Arrays.stream(intArr);
                  }).forEach(number -> System.out.println(number));
    }

}

스트림 메소드 2편

참조: 이것이 자바다