[Java] 스트림과 병렬 처리 - 파이프라인

스트림과 병렬 처리는 여러 절로 구성되어 있습니다.

[Java] 스트림과 병렬 처리 - 스트림

[Java] 스트림과 병렬 처리 - 파이프라인

[Java] 스트림과 병렬처리 - 필터링, 매핑, 정렬

[Java] 스트림과 병렬 처리 - 루핑, 매칭, 집계

[Java] 스트림과 병렬 처리 - 수집, 병렬 처리

---

2. 스트림의 종류

 자바 8부터 추가된 java.util.stream 패키지에는 스트림(stream) API 들이 포진하고 있습니다.

 패키지 내용을 보면 BaseStream 인터페이스를 부모로 해서 자식 인터페이스들이 상속 관계를 이루고 있습니다.

 BaseStream 인터페이스에는 모든 스트림에서 사용할 수 있는 공통 메소드들이 정의되어 있을 뿐 코드에서 직접적으로 사용되지 않습니다. 

 하위 스트림인 Stream, IntStream, LongStream, DoubleStream이 직접적으로 이용되는 스트림인데, Stream은 객체 요소를 처리하는 스트림이고, IntStream, LongStream, DoubleStream 은 각각 기본 타입인 int, long, double 요소를 처리하는 스트림입니다.

 이 스트림 인터페이스의 구현 객체는 다양한 소스로부터 얻을 수 있습니다.

 주로 컬렉션과 배열에서 얻지만, 다음과 같이 소스로부터 스트림 구현 객체를 얻을 수도 있습니다.

리턴 타입	메소드(파라미터)	소스
Stream<T>	java.util.Collection.stream()   java.util.Collection.parallelStream()	컬렉션
Stream<T>  IntStream  LongStream  DoubleStream	Arrays.stream(T[]),       Stream.of(T[])  Arrays.stream(int[]),      IntStream.of(int[])  Arrays.stream(long[]),    LongStream.of(long[])  Arrays.stream(double[]), DoubleStream.of(double[])	배열
IntStream	IntStream.range(int, int)  IntStream.rangeClosed(int, int)	int 범위
LongStream	LongStream.range(long, long)  LongStream.rangeClosed(long, long)	long 범위
Stream<Path>	Files.find(Path, int, BiPredicate, FileVisitOption)  Files.list(Path)	디렉토리
Stream<String>	Files.lines(Path, Charset)  BufferedReader.lines()	파일
DoubleStream  IntStream  LongStream	Random.doubles(...)  Random.ints()  Random.longs()	랜덤 수

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

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

  * FromCollectionExam.java

package stream;
 
import java.util.Arrays;
import java.util.List;
import java.util.stream.Stream;
 
public class FromCollectionExam {
 
    public static void main(String[] args) {
        List<Student> list = Arrays.asList(new Student("Jackie", 40), new Student("Hwang", 71),
                new Student("Nick", 50));
 
        Stream<Student> stream = list.stream();
        stream.forEach(s -> System.out.println(s.getName()));
    }
 
}
 
 

 

 2.2 배열로부터 스트림 얻기

  다음 예제는 String[]과 int[] 배열로부터 스트림을 얻어내고 콘솔에 출력합니다.

  * FromArrayExam.java

 
package stream;
 
import java.util.Arrays;
import java.util.stream.IntStream;
import java.util.stream.Stream;
 
public class FromArrayExam {
 
    public static void main(String[] args) {
        String[] strArr = { "Kim", "Lee", "Park" };
        Stream<String> strStream = Arrays.stream(strArr);
        strStream.forEach(s -> System.out.print(s + ","));
 
        System.out.println();
 
        int[] intArr = { 10, 20, 30, 40, 50 };
        IntStream intStream = Arrays.stream(intArr);
        intStream.forEach(i -> System.out.print(i + ", "));
    }
 
}
 

 

 2.3 숫자 범위로부터 스트림 얻기

  다음은 숫자 1부터 100까지의 합을 구하기 위해 IntStream의 rangeClosed() 메소드를 이용했습니다.

  rangeClosed()는 첫 번째 파라미터부터 두 번째 파라미터까지 순차적으로 제공하는 IntStream을 리턴합니다. 

  IntStream의 또 다른 range() 메소드도 동일한 IntStream을 리턴하는데, 두 번째 파라미터는 포함하지 않습니다.

  *FromIntRangeExam.java

 
package stream;
 
import java.util.stream.IntStream;
 
public class FromIntRangeExam {
 
    public static int sum;
 
    public static void main(String[] args) {
        IntStream stream = IntStream.rangeClosed(1, 100);
        stream.forEach(a -> sum += a);
 
        System.out.println("1 ~ 100 총합: " + sum);
    }
 
}
 

 2.4 파일로부터 스트림 얻기

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

 * FromFileContentExam.java

 
package stream;
 
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 FromFileContentExam {
 
    public static void main(String[] args) throws IOException {
        Path path = Paths.get("tmp_data.txt");
        Stream<String> stream;
 
        // Files.lines() 메소드 이용
        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 br = new BufferedReader(fileReader);
        stream = br.lines();
        stream.forEach(System.out::println);
 
        stream.close();
        br.close();
 
    }
 
}
 

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

  다음 예제는 Files의 정적 메소드인 list() 를 이용해서 디렉토리의 내용 (서브 디렉토리 또는 파일 목록)을 스트림을 통해 읽고 콘솔에 출력합니다.

  * FromDirectoryExam.java

 
package stream;
 
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 FromDirectoryExam {
 
    public static void main(String[] args) throws IOException {
        Path path = Paths.get("C:/Users/zhfld/workspace/ExampleJava/src/stream");
        Stream<Path> stream = Files.list(path);
        stream.forEach(p -> System.out.println(p.getFileName()));
    }
 
}
 

3. 스트림 파이프라인

 대량의 데이터를 가공해서 축소하는 것을 일반적으로 리덕션(Reduction) 이라고 하는데, 데이터의 합계, 평균값, 카운팅, 최대값, 최소값 등이 대표적인 리덕션의 결과물이라고 볼 수 있습니다.

 그러나 컬렉션의 요소를 리덕션의 결과물로 바로 집계할 수 없을 경우에는 집계하기 좋도록 필터링, 매핑, 정렬, 그룹핑 등의 중간 처리가 필요합니다.

 3.1 중간 처리와 최종 처리

  스트림은 데이터의 필터링, 매핑, 정렬, 그룹핑 등의 중간 처리와 합계, 평균, 카운팅, 최대값, 최소값 등의 최종 처리를 파이프라인(pipelines)으로 해결합니다. 파이프 라인은 여러 개의 스트림이 연결되어 있는 구조를 말합니다. 파이프라인에서 최종 처리를 제외하고는 모두 중간 처리 스트림입니다.

 

  중간 스트림이 생성될 때 요소들이 바로 중간 처리(필터링, 매핑, 정렬)되는 것이 아니라 최종 처리가 시작되기 전까지 중간 처리는 지연(lazy)됩니다.

  최종 처리가 시작되면 비로소 컬렉션의 요소가 하나씩 중간 스트림에서 처리되고 최종 처리까지 오게 됩니다.

  Stream 인터페이스에는 필터링, 매핑, 정렬 등의 많은 중간 처리 메소드가 있는데, 이 메소드들은 중간 처리된 스트림을 리턴합니다.

  그리고 이 스트림에서 다시 중간 처리 메소드를 호출해서 파이프라인을 형성하게 됩니다.

  예를 들어 회원 컬렉션에서 남자만 필터링하는 중간 스트림을 연결하고, 다시 남자의 나이로 매핑하는 스트림을 연결한 후, 최종 남자 평균 나이를 집계한다면 다음 그림처럼 파이프라인이 형성됩니다.

  파이프라인을 자바 코드로 표현하면 다음과 같습니다.

Stream<Member> maleFemaleStream = list.stream();
Stream<Member> maleStream = maleFemaleStream.filter(m -> m.getSex() == Member.MALE);
IntStream ageStream = maleStream.mapToInt(Member::getAge);
OptionalDouble optionalDouble = ageStream.average();
double ageAvg = optionalDouble.getAsDouble();
 

  로컬 변수를 생략하고 연결하면 다음과 같은 형태의 파이프라인 코드만 남습니다.

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

  filter( m -> m.getSex() == Member.MALE)는 남자 Member 객체를 요소로 하는 새로운 스트림을 생성합니다. 

  mapToInt(Member :: getAge())는 Member 객체를 age 값으로 매핑해서 age를 요소로 하는 새로운 스트림을 생성합니다.

  average() 메소드는 age 요소들의 평균을 OptionalDouble에 저장합니다. OptionalDouble에서 저장된 평균값을 읽으려면 getAsDouble() 메소드를 호출하면 됩니다.

  * StreamPipelinesExam.java

 
package stream;
 
import java.util.Arrays;
import java.util.List;
 
public class StreamPipelinesExam {
 
    public static void main(String[] args) {
        List<Member> list = Arrays.asList(
                new Member("Kush", Member.MALE, 40),
                new Member("Pierre", Member.MALE, 22),
                new Member("Jolie", Member.FEMALE, 18),
                new Member("Sozi", Member.FEMALE, 22)
        );
        
        double ageAvg = list.stream()
                .filter(m -> m.getSex() == Member.MALE)
                .mapToInt(Member :: getAge)
                .average()
                .getAsDouble();
        
        System.out.println("남자 평균 나이: " + ageAvg);
    }
 
}
 

 * Member.java

 
package stream;
 
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) {
        super();
        this.name = name;
        this.sex = sex;
        this.age = age;
    }
 
    public String getName() {
        return name;
    }
 
    public void setName(String name) {
        this.name = name;
    }
 
    public int getSex() {
        return sex;
    }
 
    public void setSex(int sex) {
        this.sex = sex;
    }
 
    public int getAge() {
        return age;
    }
 
    public void setAge(int age) {
        this.age = age;
    }
 
}
 

 * 이 포스트은 서적 '이것이 자바다' 를 참고하여 작성한 포스트입니다.