TinyStream과 내부 반복

Stream API를 사용하기만 하면 메서드 체이닝이 편리하다는 인상에 머물 수 있습니다. 작은 스트림을 직접 구현하면 각 연산이 무엇을 저장하고 언제 실행되는지, 타입이 어떻게 바뀌는지, 반복을 누가 소유하는지 분명해집니다. 이 장의 첫 구현은 이해하기 쉬운 불변 즉시 파이프라인이며, 연습문제에서 Iterator 기반 지연 순서로 확장합니다.

of(source)는 팩터리, filter와 map은 중간 연산, toList와 count는 최종 연산 역할을 합니다. 실제 JDK Stream은 한 번 최종 연산을 실행하면 재사용할 수 없고 지연 평가을 사용하지만, 학습 구현은 스냅샷 값으로 새 객체를 반환해 재사용 가능합니다. 이름이 같아도 수명 계약이 다름을 명시합니다.

가변 체이닝 오염

다음 구현은 filter()가 내부 values를 덮어쓰고 this를 반환합니다. 기반에서 두 파이프라인을 분기했다고 생각해도 첫 필터가 기반 상태를 바꿔 두 번째 분기가 축소된 원본을 봅니다. map으로 원소 타입이 바뀌는 연산은 같은 제네릭 인스턴스를 안전하게 반환하기도 어렵습니다.

bad/MutableSelfReturningStream.java

import java.util.ArrayList;
import java.util.List;
import java.util.function.Predicate;

public final class MutableSelfReturningStream<T> {
    private List<T> values;

    MutableSelfReturningStream(List<T> source) {
        values = new ArrayList<>(source);
    }

    MutableSelfReturningStream<T> filter(Predicate<? super T> predicate) {
        values.removeIf(predicate.negate());
        return this;
    }

    List<T> toList() {
        return values;
    }

    public static void main(String[] args) {
        MutableSelfReturningStream<Integer> base = new MutableSelfReturningStream<>(List.of(1, 2, 3, 4));
        List<Integer> even = base.filter(value -> value % 2 == 0).toList();
        List<Integer> large = base.filter(value -> value >= 3).toList();
        System.out.println(even);
        System.out.println(large);
    }
}

플루언트 API가 반드시 가변해야 하는 것은 아닙니다. 연산마다 새 파이프라인 값을 반환하면 기반을 안전하게 분기할 수 있습니다. 원소 목록도 스냅샷으로 고정하면 호출자가 원본 목록을 나중에 수정해도 파이프라인 의미가 변하지 않습니다.

실제 Stream이 단일 사용인 이유에는 자원 관리와 지연 순회 상태가 포함됩니다. 사용자 정의 API가 재사용 가능하다고 해서 JDK Stream도 같은 방식으로 보관하면 안 됩니다. Supplier<Stream<T>>를 보관해 필요할 때 새 스트림을 만드는 패턴이 더 맞습니다.

불변 중간 연산

아래 TinyStream은 생성자를 비공개로 두고 of() 팩터리로 원본 스냅샷을 만듭니다. 필터와 map은 즉시 반복문을 실행한 뒤 새 TinyStream을 반환합니다. map의 결과 타입 R이 새로운 제네릭 인스턴스에 반영됩니다.

src/ImmutableEagerTinyStream.java

import java.util.ArrayList;
import java.util.List;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Predicate;

public final class ImmutableEagerTinyStream<T> {
    private final List<T> values;

    private ImmutableEagerTinyStream(List<? extends T> values) {
        this.values = List.copyOf(values);
    }

    static <T> ImmutableEagerTinyStream<T> of(List<? extends T> source) {
        return new ImmutableEagerTinyStream<>(source);
    }

    ImmutableEagerTinyStream<T> filter(Predicate<? super T> predicate) {
        List<T> selected = new ArrayList<>();
        for (T value : values) {
            if (predicate.test(value)) {
                selected.add(value);
            }
        }
        return new ImmutableEagerTinyStream<>(selected);
    }

    <R> ImmutableEagerTinyStream<R> map(Function<? super T, ? extends R> mapper) {
        List<R> mapped = new ArrayList<>(values.size());
        for (T value : values) {
            mapped.add(mapper.apply(value));
        }
        return new ImmutableEagerTinyStream<>(mapped);
    }

    void forEach(Consumer<? super T> consumer) {
        for (T value : values) {
            consumer.accept(value);
        }
    }

    long count() {
        return values.size();
    }

    List<T> toList() {
        return values;
    }

    public static void main(String[] args) {
        ImmutableEagerTinyStream<Integer> base = of(List.of(1, 2, 3, 4));
        List<String> even = base
                .filter(value -> value % 2 == 0)
                .map(value -> "N" + value)
                .toList();
        List<Integer> large = base.filter(value -> value >= 3).toList();
        System.out.println(even);
        System.out.println(large);
        System.out.println("baseCount=" + base.count());
    }
}

메서드 체이닝은 각 반환값을 다음 수신 객체로 쓰는 중첩 호출입니다. 줄 break는 실행 경계가 아니며, 이 즉시 구현에서는 필터 호출이 끝난 뒤 map이 시작됩니다. 지연 스트림에서는 중간 호출이 연산 메타데이터만 쌓고 최종에서 순회합니다.

toList()가 내부 불변 목록을 그대로 반환하는 것은 안전합니다. 가변 ArrayList를 보관한다면 방어적 복사가 필요합니다. 불변 파이프라인은 별도 잠금 없이 여러 판독기가 사용할 수 있지만 전달된 함수가 가변 캡처를 가진 경우 실행 전체가 스레드-안전하다는 뜻은 아닙니다.

for 반복문과 Iterator는 호출자가 hasNext, next, break 시점을 제어하는 외부 반복입니다. forEach, 필터, map 파이프라인은 컬렉션 라이브러리가 순회를 소유하고 호출자는 원소 동작만 제공합니다. 단락 평가가 필요하면 boolean을 반환하는 방문자나 findFirst 같은 전용 최종 연산이 있어야 합니다.

src/InternalExternalIteration.java

import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.function.Predicate;

public final class InternalExternalIteration {
    @FunctionalInterface
    interface VisitStep<T> {
        boolean visit(T value);
    }

    static <T> List<T> externalUntil(List<T> source, Predicate<? super T> stop) {
        List<T> visited = new ArrayList<>();
        Iterator<T> iterator = source.iterator();
        while (iterator.hasNext()) {
            T value = iterator.next();
            if (stop.test(value)) {
                break;
            }
            visited.add(value);
        }
        return List.copyOf(visited);
    }

    static <T> void forEachUntil(Iterable<T> source, VisitStep<? super T> step) {
        for (T value : source) {
            if (!step.visit(value)) {
                return;
            }
        }
    }

    static <T> List<T> internalUntil(Iterable<T> source, Predicate<? super T> stop) {
        List<T> visited = new ArrayList<>();
        forEachUntil(source, value -> {
            if (stop.test(value)) {
                return false;
            }
            visited.add(value);
            return true;
        });
        return List.copyOf(visited);
    }

    static <T> T findFirst(List<T> source, Predicate<? super T> predicate) {
        for (T value : source) {
            if (predicate.test(value)) {
                return value;
            }
        }
        return null;
    }

    public static void main(String[] args) {
        List<Integer> source = List.of(1, 2, 3, 4);
        System.out.println(externalUntil(source, value -> value == 3));
        System.out.println(internalUntil(source, value -> value == 3));
        System.out.println(findFirst(source, value -> value % 2 == 0));
    }
}

예제의 forEachUntil()은 라이브러리가 순회를 소유하고 콜백의 boolean 결과를 계속·중단 신호로 해석하는 전용 내부 반복 연산입니다. 표준 Iterable.forEach()가 받는 Consumer에는 중단 신호가 없으므로, 중간 종료가 필요하면 이처럼 계약을 확장하거나 findFirst() 같은 전용 최종 연산을 제공해야 합니다.

내부 반복은 라이브러리가 순차, 병렬, 지연 실행 전략을 선택할 여지를 줍니다. 대신 콜백이 호출 횟수와 스레드, 순서에 의존하지 않아야 합니다. 외부 반복문은 복잡한 break와 continue가 필요할 때 더 직접적일 수 있습니다.

지연 파이프라인 상태

지연 순서는 원본 Iterator를 필터 반복자와 map 반복자로 감쌉니다. 중간 연산에서는 원본을 읽지 않고 최종 연산이 next를 요청할 때 필요한 만큼만 계산합니다. findFirst나 take와 결합하면 전체 입력을 보지 않을 수 있습니다.

Iterator는 가변 순회 커서이므로 같은 반복자를 여러 최종이 공유하면 재사용 문제가 생깁니다. 순서가 Supplier<Iterator<T>>를 보관하면 최종마다 새 커서를 만들 수 있습니다. 원본 자체가 일회성 네트워크 스트림이면 공급자로 재생할 수 없으므로 단일 사용 계약을 택해야 합니다.

연습 문제

Sequence.of(List)가 새 반복자 팩터리를 보관하고, 필터와 map이 래퍼 반복자 팩터리를 반환하게 합니다. take(n) 최종 연산은 n개만 요청해야 합니다. 매퍼 호출 횟수를 세어 원본 전체가 평가되지 않았음을 확인합니다.

해답 보기

src/LazySequenceSolution.java

import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.NoSuchElementException;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Function;
import java.util.function.Predicate;
import java.util.function.Supplier;

public final class LazySequenceSolution<T> {
    private final Supplier<Iterator<T>> iteratorFactory;

    private LazySequenceSolution(Supplier<Iterator<T>> iteratorFactory) {
        this.iteratorFactory = iteratorFactory;
    }

    static <T> LazySequenceSolution<T> of(List<? extends T> source) {
        List<T> snapshot = List.copyOf(source);
        return new LazySequenceSolution<>(snapshot::iterator);
    }

    LazySequenceSolution<T> filter(Predicate<? super T> predicate) {
        return new LazySequenceSolution<>(() -> new Iterator<>() {
            private final Iterator<T> source = iteratorFactory.get();
            private T next;
            private boolean ready;

            private void prepare() {
                while (!ready && source.hasNext()) {
                    T candidate = source.next();
                    if (predicate.test(candidate)) {
                        next = candidate;
                        ready = true;
                    }
                }
            }

            @Override
            public boolean hasNext() {
                prepare();
                return ready;
            }

            @Override
            public T next() {
                prepare();
                if (!ready) {
                    throw new NoSuchElementException();
                }
                T value = next;
                ready = false;
                next = null;
                return value;
            }
        });
    }

    <R> LazySequenceSolution<R> map(Function<? super T, ? extends R> mapper) {
        return new LazySequenceSolution<>(() -> new Iterator<>() {
            private final Iterator<T> source = iteratorFactory.get();

            @Override
            public boolean hasNext() {
                return source.hasNext();
            }

            @Override
            public R next() {
                return mapper.apply(source.next());
            }
        });
    }

    List<T> take(int limit) {
        if (limit < 0) {
            throw new IllegalArgumentException("negative limit");
        }
        List<T> result = new ArrayList<>(limit);
        Iterator<T> iterator = iteratorFactory.get();
        while (result.size() < limit && iterator.hasNext()) {
            result.add(iterator.next());
        }
        return List.copyOf(result);
    }

    public static void main(String[] args) {
        AtomicInteger mapped = new AtomicInteger();
        LazySequenceSolution<String> sequence = LazySequenceSolution
                .of(List.of(1, 2, 3, 4, 5, 6))
                .filter(value -> value % 2 == 0)
                .map(value -> {
                    mapped.incrementAndGet();
                    return "N" + value;
                });
        System.out.println(sequence.take(2));
        System.out.println("mapped=" + mapped.get());
        System.out.println(sequence.take(1));
    }
}

해답은 최종마다 새 반복자 체인을 만들기 때문에 순서를 재사용할 수 있습니다. 두 번째 take에서 매퍼가 다시 실행되는 점도 계약의 일부입니다. 캐시가 필요하면 지연과 메모이제이션을 별도로 설계합니다.

가변 체이닝 오염

불변 중간 연산

외부 반복과 내부 반복

지연 파이프라인 상태

연습 문제

이 페이지의 목차