람다와 스트림

반환 타입으로는 스트림보다 컬렉션이 낫다.

스트림은 반복을 지원하지 않는다. 스트림과 반복을 알맞게 조합해야 좋은 코드가 나온다.

 

Stream인터페이스는 Iterable 인터페이스가 정의한 추상 메서드를 전부 포함할 뿐 아니라 Iterable 인터페이스가 정의한 방식대로 동작한다. 아래코드는 자바에서 타입추론이 안되게 때문에 컴파일 에러가 난다.

for(ProcessHandle ph : ProcessHandle.allProcesses()::iterator) {
    // 프로세스르르 처리한다.
}

 

아래코드는 문법적으로는 맞지만 직관성이 떨어진다.

for(ProcessHandle ph : (Iterable<ProcessHandle>) ProcessHandle.allProcesses()::iterator) {
    // 프로세스르르 처리한다.
}

 

Steam은 Iterable인터페이스를 포함하기 때문에 스트림을 받아 Iterable로 리턴해 줄 수 있다. Adapter로 직관성이 높아지지만 파이프라인을 처리하지 못하게 된다.

public static <E> Iterable<E> iterableOf(Stream<E> stream) {
    return stream::iterator;
}

for(ProcessHandle ph : iterableOf(ProcessHandle.allProcesses())) {
    // 프로세스르르 처리한다.
}

 

Stream을 리턴받게 한다면 파이프라인 처리를 할 수 있다.

public static <E> Stream<E> streamOf(Iterable<E> iterable) {
    return StreamSupport.stream(iterable.spliterator(), false);
}

 

원소 시퀀스를 반환하는 공개API의 반환 타입에는 Collection이나 그 하위 타입을 쓰는게 일반적으로 최선이다. 하지만, 덩치 큰 시퀀스를 메모리에 올려서는 안된다. 

아래코드는 BitMask를 이용한 방법이다. 원소 index에서 1인 비트 연산을 찾아 원소가 존재하는 index를 result에 넣는 것이다.

 

 

public class PowerSet {
    public static final <E> Collection<Set<E>> of(Set<E> s) {
        List<E> src = new ArrayList<>(s);
        if(src.size() > 30)
            throw new IllegalArgumentException(
                    "집합의 원소가 너무 많습니다(최대 30개). : " + s);
        return new AbstractList<Set<E>>() {
            @Override
            public Set<E> get(int index) {
                Set<E> result = new HashSet<>();
                for(int i = 0; index != 0; i++, index >>= 1)
                    if((index & 1) == 1)
                        result.add(src.get(i));
                return result;
            }

            @Override
            public int size() {
                return 1 << src.size();
            }

            @Override
            public boolean contains(Object o) {
                return o instanceof Set && src.containsAll((Set) o);
            }
        };
    }
}

 

Stream은 일종의 방법이지 반드시 써야 하는 것은 아니다. 위에서 봤듯이 스트림이 무조건 좋은 것은 아니다. 과도한 형변환과 스트림의 변환은 속도를 느리게 할 수도 있다. for문과 stream, Collection 등.. 을 적절히 사용하여 가독성 좋은 코드를 짜는 것이 현명하다.

 

스트림 병렬화는 주의해서 적용하라.

동시성 프로그래밍을 하기 위해서는 안저성과 응답 가능 상태를 유지하기 위해 애써야 한다. 스트림에서 병렬화를 옳게 사용하기 위한 방법을 알아보자.

 

아래 코드는 저자가 parallel()을 호출하여 실행해보았지만 1시간 반이나 지나 강제종료하였다고 한다. 이유가 무엇일까?

스트림 라이브러리가 파이프라인을 병렬화하는 방법을 찾아내지 못했기 때문이다. 환경이 아무리 좋더라도 데이터 소스가 Stream.iterate거나 중간 연산으로 limit를 쓰면 파이프라인 병렬화로는 성능 개선을 기대 할 수 없다.

public static void main(String[] args) {
    primes().map(p -> TWO.pow(p.intValueExact()).subtract(ONE))
            .filter(mersenne -> mersenne.isProbablePrime(50))
            .limit(20)
            .forEach(System.out::println);
}

static Stream<BigInteger> primes() {
    return Stream.iterate(TWO, BigInteger::nextProbablePrime);
}

스트림 파이프라인이 limit다룰 때 CPU에 코어가 남는다면 limit이후에 원소를 처리 한후 결과를 버리게 된다. 위 코드에서 19번째를 계산한다음 CPU코어 남게 되면 20번째 이후의 것도 계산을 하게 된다. 이 계산들이 끝나지 않기 때문에 오래걸리게 된 것이다.

 

스트림 소스는 ArrayList, HashMap, HashSet, ConcurrentHashMap의 인스턴스거나 배열, int 범위, long 범위 일 때 병렬화의 효과가 가장 좋다. 

• 원하는 크기로 정확하게 손쉽게 나눌 수 있어 다수의 스레드에 분배하기 좋다.
• 지역 참조성이 뛰어나다. 지역 참조성이 낮으면 데이터가 주 메모리에서 캐시 메모리로 전송되어 오기를 기다리며 대부분 시간을 멍하니 보내게 된다.

스트림의 종단 연산의 동작 방식 역시 병렬수행 효율에 영향을 준다. reduce 메서드 같이 max, min, count, sum같이 완성된 형태로 제공되거나 anyMatch, allMatch, noneMatch처럼 조건에 맞으면 바로 반환되는 메서드도 병렬화에 적합하다.

 

스트림을 잘못 병렬화하면 (응답 불가를 포함해) 성능이 나빠질 뿐만 아니라 결과 자체가 잘못되거나 예상 못한 동작이 발생할 수 있다. Stream의 reduce 연산에 건네자는 accumulator와 combiner함수는 반드시 결합법칙을 만족하고, 간섭받지 않고, 상태를 갖지 않아야 한다.

 

소수 계산 스트림 파이브라인 병렬화 버전

static long pi(long n) {
    return LongStream.rangeClosed(2, n)
            .parallel()
            .mapToObj(BigInteger::valueOf)
            .filter(i -> i.isProbablePrime(50))
            .count();
}