상태를 가진 람다와 재사용 문제

앞의 글들에서 filter와 map에 넘긴 함수는 모두 지금 들어온 원소 하나만 보고 답을 냈습니다. value -> value % 2 == 0에 4를 넣으면 언제, 몇 번째로 호출하든 true가 나옵니다. 그런데 람다 본문은 바깥에 있는 필드나 컬렉션, 카운터도 읽고 바꿀 수 있습니다. 그런 람다는 같은 값을 넣어도 지금까지 무엇을 처리했느냐에 따라 다른 답을 냅니다. 이런 람다를 상태를 가진(stateful) 람다라고 부릅니다.

상태가 언제나 잘못된 것은 아닙니다. 중복을 없애려면 이미 본 값을 기억해야 하고, 누적 합계를 내려면 지금까지의 합을 들고 있어야 합니다. 문제가 되는 것은 상태가 살아 있는 기간, 즉 수명입니다. 한 번의 실행에서만 쓰여야 할 상태가 실행이 끝난 뒤에도 남아 있으면, 같은 함수를 다시 쓸 때 앞 실행의 기억이 결과에 섞여 듭니다. 이 글에서는 그런 일이 어떻게 일어나는지 확인하고, 상태의 수명을 실행 한 번에 맞추는 방법을 익힙니다.

예제에는 JDK의 Stream도 등장합니다. list.stream().filter(조건).toList()는 앞 글의 TinyStream에서 of(list).filter(조건).toList()라고 쓰던 것과 같은 모양이고, Stream 자체는 32장에서 자세히 다룹니다. 지금은 원소를 하나씩 조건 함수에 넣어 보고 통과한 것만 목록으로 모은다고 읽으면 충분합니다.

같은 조건 함수를 두 번 쓰면 결과가 달라진다

다음 Predicate는 지금까지 본 값을 필드 seen에 모아 두고, 처음 보는 값에만 true를 돌려줍니다. Set.add는 원소가 새로 들어가면 true, 이미 있으면 false를 돌려주기 때문에 return seen.add(value); 한 줄이 그대로 중복 제거 조건이 됩니다. 이 조건 함수 하나를 두 스트림에 차례로 넘겨 보겠습니다.

import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.function.Predicate;

public final class ReusableStatefulPredicate<T> implements Predicate<T> {
    private final Set<T> seen = new HashSet<>();

    @Override
    public boolean test(T value) {
        return seen.add(value);
    }

    public static void main(String[] args) {
        ReusableStatefulPredicate<String> distinct = new ReusableStatefulPredicate<>();
        System.out.println(List.of("a", "a", "b").stream().filter(distinct).toList());
        System.out.println(List.of("a", "b", "c").stream().filter(distinct).toList());
    }
}

두 번째 입력에는 중복이 없습니다. 조건 함수가 스트림마다 새로 시작한다면 두 번째 줄에는 세 원소가 모두 남아야 합니다.

[a, b]
[c]

첫 줄은 기대한 대로 중복된 a 하나가 빠졌습니다. 그런데 두 번째 입력 a, b, c에는 중복이 하나도 없는데도 c만 남았습니다. 실행마다 seen이 어떻게 바뀌는지 따라가 보면 이유가 보입니다.

seen은 객체의 필드라서 첫 스트림이 끝나도 비워지지 않습니다. 그래서 두 번째 스트림은 빈 집합이 아니라 첫 실행이 남긴 a, b를 안고 시작하고, 그 둘을 이미 본 값으로 여겨 거절합니다. 예외도 경고도 없이 결과만 조용히 틀리기 때문에, 출력을 꼼꼼히 비교하기 전에는 알아차리기 어려운 버그입니다.

이 인스턴스를 여러 스레드가 함께 쓰면 문제가 하나 더 생깁니다. HashSet은 여러 스레드가 동시에 수정하는 상황을 고려하지 않은 컬렉션입니다. 그래서 두 스레드가 동시에 add를 호출하면 내부 상태가 어긋나, 같은 값이 두 번 통과하거나 추가한 값이 사라지는 일이 생길 수 있습니다.

동시 컬렉션으로 바꿔도 남는 문제

스레드 문제를 보면 HashSet을 스레드 안전한 집합으로 바꾸고 싶어집니다. 필드를 다음처럼 고치고, import 문도 java.util.concurrent.ConcurrentHashMap으로 바꿔 다시 실행해 보겠습니다.

private final Set<T> seen = ConcurrentHashMap.newKeySet();

출력은 여전히 [a, b]와 [c]입니다. 동시 컬렉션이 막아 주는 것은 여러 스레드가 동시에 수정할 때 집합이 깨지는 일이지, 지난 실행이 남긴 값을 지우는 일이 아니기 때문입니다. 스레드 안전성과 '실행마다 상태를 새로 시작하는 것'은 서로 다른 문제이고, 하나를 고쳐도 다른 하나는 그대로 남습니다.

JDK Stream의 distinct()에는 이런 문제가 없습니다. distinct()가 쓰는 '이미 본 값' 집합은 그 스트림을 실행할 때 만들어졌다가 실행이 끝나면 함께 사라지고, 스트림 자체도 한 번만 실행할 수 있기 때문입니다. 다만 distinct()는 원소 전체의 equals로만 중복을 판단합니다. 같은 id를 가진 사용자는 하나만 남기는 것처럼 키를 기준으로 거르려면 직접 만들어야 하는데, 그 방법은 이 글의 뒷부분에서 다룹니다.

바깥 카운터로 번호 붙이기

상태를 가진 람다의 또 다른 흔한 예는 원소에 순번을 붙이는 코드입니다. 반복문에서 하던 대로 지역 변수 하나를 두고 람다 안에서 늘리려고 하면 컴파일러가 막습니다.

import java.util.List;

public final class LocalCounterIndex {
    public static void main(String[] args) {
        List<String> words = List.of("a", "b", "c");
        int index = 0;
        List<String> numbered = words.stream()
                .map(word -> index++ + ":" + word)
                .toList();
        System.out.println(numbered);
    }
}

이 코드는 실행까지 가지 못합니다. javac가 람다 안의 index++를 가리키며 다음 오류를 냅니다.

LocalCounterIndex.java:8: error: local variables referenced from a lambda expression must be final or effectively final
                .map(word -> index++ + ":" + word)
                             ^
1 error

람다가 캡처하는 지역 변수는 사실상 final(effectively final), 즉 초기화한 뒤 다시 대입하지 않는 변수여야 합니다. 람다는 변수 자체가 아니라 캡처한 순간의 값을 복사해 들고 다닙니다. 그래서 변수를 나중에 바꿔도 람다는 그 변화를 볼 수 없고, 자바는 이런 혼란을 막으려고 재대입 자체를 금지합니다. index++는 index에 새 값을 대입하므로 이 규칙에 걸립니다. 자세한 내용은 30장에서 람다와 익명 클래스를 비교한 글에서 다뤘습니다.

그래서 규칙을 피하려고 AtomicInteger나 길이가 1인 배열을 캡처하는 코드를 자주 보게 됩니다. 변수는 그대로 두고 변수가 가리키는 객체 안의 값만 바꾸니 컴파일은 됩니다. 하지만 이제 번호의 뜻이 '원소의 위치'에서 '람다가 몇 번째로 호출됐는가'로 바뀝니다. 순차 실행에서는 두 값이 우연히 같아서 차이가 드러나지 않습니다. 다음 프로그램은 같은 코드를 병렬로 실행해 보고, 번호를 위치에서 계산하는 방법과 비교합니다.

import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.IntStream;

public final class CapturedCounterIndex {
    public static void main(String[] args) {
        List<String> words = List.of("a", "b", "c", "d", "e", "f", "g", "h");

        AtomicInteger next = new AtomicInteger();
        List<String> sequential = words.stream()
                .map(word -> next.getAndIncrement() + ":" + word)
                .toList();
        System.out.println(sequential);

        next.set(0);
        List<String> parallel = words.parallelStream()
                .map(word -> next.getAndIncrement() + ":" + word)
                .toList();
        System.out.println(parallel);

        List<String> byPosition = IntStream.range(0, words.size())
                .parallel()
                .mapToObj(i -> i + ":" + words.get(i))
                .toList();
        System.out.println(byPosition);
    }
}

실행 결과의 한 예입니다. 두 번째 줄은 실행할 때마다 달라질 수 있습니다.

[0:a, 1:b, 2:c, 3:d, 4:e, 5:f, 6:g, 7:h]
[6:a, 7:b, 4:c, 5:d, 2:e, 3:f, 0:g, 1:h]
[0:a, 1:b, 2:c, 3:d, 4:e, 5:f, 6:g, 7:h]

첫 줄은 순차 실행이라 번호와 위치가 일치합니다. 두 번째 줄의 parallelStream()은 목록을 여러 조각으로 나눠 여러 스레드에서 동시에 처리합니다. toList()가 결과를 원래 순서대로 모아 주므로 단어는 a부터 h까지 그대로지만, 번호는 어느 스레드가 먼저 getAndIncrement()를 호출했느냐에 따라 나눠 가졌습니다. 그래서 g가 0번을 받는 식으로 번호가 섞이고, 섞이는 모양도 실행할 때마다 바뀝니다.

AtomicInteger가 스레드 안전하다는 사실은 여기서 도움이 되지 않습니다. 같은 번호가 두 번 나오거나 번호가 빠지는 일은 막아 주지만, 어떤 원소가 몇 번을 받을지는 여전히 스레드 실행 순서가 정합니다. 앞 절에서 동시 컬렉션이 지난 실행의 값을 지워 주지 못한 것과 같은 이야기입니다.

세 번째 줄처럼 번호를 원본의 위치에서 계산하면 실행 방식과 상관없이 결과가 같습니다. IntStream.range(0, words.size())는 0부터 7까지의 정수를 차례로 내보내는 스트림이고, mapToObj는 각 인덱스 i를 words.get(i)와 묶어 문자열로 바꿉니다. 번호가 곧 위치이므로 어느 스레드가 어떤 순서로 처리하든 어긋날 수 없습니다. 병렬 실행이 필요 없는 코드라면 인덱스를 쓰는 평범한 for 반복문도 같은 일을 하는 좋은 방법입니다.

람다와 익명 클래스가 붙잡는 객체

상태는 람다 안에서 바꾸는 값에만 해당하는 이야기가 아닙니다. 함수가 무엇을 캡처하느냐에 따라 어떤 객체가 함께 살아남는지도 달라집니다. 30장에서 봤듯이 람다는 새 this를 만들지 않으므로, 인스턴스 메서드 안에서 만든 람다의 this는 그 메서드를 실행하고 있는 바깥 객체입니다. 반면 익명 클래스 본문의 this는 익명 객체 자신을 가리킵니다. 같은 입력을 두 방식으로 처리해 비교해 보겠습니다.

import java.util.List;
import java.util.function.Function;

public final class LambdaAnonymousPipelineScope {
    private final String name;

    LambdaAnonymousPipelineScope(String name) {
        this.name = name;
    }

    Function<String, String> lambdaFormatter() {
        return value -> this.name + ":" + value;
    }

    Function<String, String> anonymousFormatter() {
        return new Function<>() {
            private final String name = "anonymous";

            @Override
            public String apply(String value) {
                return this.name + ":" + LambdaAnonymousPipelineScope.this.name + ":" + value;
            }
        };
    }

    public static void main(String[] args) {
        LambdaAnonymousPipelineScope owner = new LambdaAnonymousPipelineScope("pipeline");
        System.out.println(List.of("a", "b").stream().map(owner.lambdaFormatter()).toList());
        System.out.println(List.of("a", "b").stream().map(owner.anonymousFormatter()).toList());
    }
}

두 포매터는 같은 owner에서 만들어지고 같은 입력 a, b를 받습니다. 그러니 두 줄의 차이는 본문의 this.name이 어느 객체의 필드를 읽느냐에서만 나옵니다.

[pipeline:a, pipeline:b]
[anonymous:pipeline:a, anonymous:pipeline:b]

첫 줄에서 람다의 this.name은 바깥 객체 owner의 필드이므로 pipeline이 나옵니다. 익명 클래스 안에서 this.name은 익명 객체가 직접 선언한 필드라서 anonymous이고, 바깥 객체의 필드는 LambdaAnonymousPipelineScope.this.name처럼 바깥 클래스 이름을 붙여야 읽을 수 있습니다. 그래서 두 번째 줄에는 두 이름이 차례로 모두 들어갑니다. 아래 도표는 입력 "a" 하나에 대해 두 방식이 어느 필드를 읽는지 나란히 보여 줍니다.

수명의 관점에서 중요한 점은 두 함수가 모두 바깥 객체를 붙잡고 있다는 것입니다. owner.lambdaFormatter()가 돌려준 함수를 어딘가에 오래 보관하면, 프로그램이 owner를 더는 쓰지 않더라도 그 함수가 살아 있는 한 owner는 가비지 컬렉션 대상이 되지 않습니다. 메서드 참조 this::format도 마찬가지로 this를 붙잡습니다. 반면 정적 메서드 참조는 붙잡을 객체가 없으므로, 오래 보관되는 함수 목록에 넣어도 다른 객체의 수명을 늘리지 않습니다. 그렇다고 동작이 실제로 인스턴스의 설정에 의존하는데 이 이유만으로 억지로 static으로 바꿀 필요는 없습니다.

사실상 final 규칙도 같은 관점에서 읽어야 합니다. 이 규칙은 지역 변수에 다른 값을 대입하는 것만 막을 뿐, 변수가 가리키는 객체의 내용이 바뀌는 것은 막지 않습니다. 예를 들어 List<String> log를 캡처한 Consumer는 그 목록에 계속 값을 추가할 수 있고, 메서드가 끝난 뒤에도 함수가 살아 있는 한 목록도 함께 살아 있습니다.

순회하는 동안 원본을 바꾸지 않기

java.util.stream 패키지 문서는 filter나 map에 넘기는 함수, 즉 행동 매개변수(behavioral parameter)에 두 가지를 요구합니다. 하나는 실행 중에 스트림의 원본을 수정하지 않는 것으로, 이를 비간섭(non-interference)이라고 합니다. 다른 하나는 대부분의 경우 실행 도중 바뀔 수 있는 상태에 결과가 의존하지 않는 것으로, 무상태(stateless)라고 합니다. 앞의 두 절이 무상태 요구를 어긴 예였다면, 이번에는 비간섭 요구를 어긴 예를 보겠습니다. 다음 코드는 짝수를 만날 때마다 10을 곱한 값을 원본 목록 끝에 추가합니다.

import java.util.ArrayList;
import java.util.ConcurrentModificationException;
import java.util.List;

public final class ModifySourceDuringStream {
    public static void main(String[] args) {
        List<Integer> source = new ArrayList<>(List.of(1, 2, 3, 4));
        try {
            source.stream()
                    .filter(value -> value % 2 == 0)
                    .forEach(value -> source.add(value * 10));
        } catch (ConcurrentModificationException e) {
            System.out.println("예외: " + e.getClass().getName());
        }
        System.out.println(source);
    }
}

catch는 예외의 클래스 이름만 찍습니다. 마지막 줄은 예외가 난 뒤 원본 목록에 무엇이 남았는지를 보여 줍니다.

예외: java.util.ConcurrentModificationException
[1, 2, 3, 4, 20, 40]

20과 40은 원본에 실제로 추가됐지만, 스트림은 ConcurrentModificationException으로 끝났습니다. 새로 들어간 20과 40도 짝수인데 200과 400이 없으니, 추가된 원소는 순회 대상이 아니었다는 것도 알 수 있습니다. ArrayList는 원소를 추가하거나 지울 때마다 수정 횟수를 하나씩 늘려 기록합니다. 이 스트림은 순회를 마친 뒤 그 횟수가 시작할 때와 같은지 확인하고, 달라졌으면 결과를 믿을 수 없다고 보고 예외를 던집니다. 이처럼 수정을 알아채는 즉시 실패시키는 방식을 빠른 실패(fail-fast)라고 합니다.

빠른 실패는 최선을 다하는 검사일 뿐 모든 경우를 잡아 준다는 약속이 아닙니다. 패키지 문서도 실행 중에 원본을 수정하면 예외나 틀린 결과, 명세와 다른 동작이 생길 수 있다고만 말합니다. 그러니 예외가 나지 않았다고 해서 원본을 바꾸는 코드가 안전한 것은 아닙니다. 참고로 문서가 말하는 '실행 중'은 최종 연산을 호출한 순간부터 끝날 때까지입니다. list.stream()으로 스트림을 만든 뒤 최종 연산을 호출하기 전에 원소를 추가하는 것은 간섭이 아니며, 그 원소도 결과에 포함됩니다.

결과를 모으는 쪽을 바꾸는 것은 원본 간섭이 아닙니다. 최종 연산 forEach에서 원본과 다른 목록에 값을 추가하는 코드는 흔히 볼 수 있고, 순차 실행에서는 문제없이 동작합니다. 간섭이 되는 것은 결과를 담는 곳이 곧 읽고 있는 원본일 때입니다. 다만 그런 바깥 목록도 여러 스레드가 함께 쓰면 경합이 생기므로, 결과는 toList()처럼 스트림이 직접 모아 주게 하는 편이 낫습니다.

원본을 바꾸는 작업과 원본을 읽는 작업을 분리하면 이런 문제를 피할 수 있습니다. 다음 Pipeline은 실행을 시작할 때 원본의 스냅샷(snapshot), 즉 그 순간의 원소 목록을 복사해 두고 복사본만 순회합니다. 결과는 별도의 목록에 모았다가 수정할 수 없는 목록으로 돌려줍니다.

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

public final class NonInterferingPipeline {
    record Pipeline<T, R>(Predicate<? super T> filter, Function<? super T, ? extends R> mapper) {
        List<R> execute(List<? extends T> source) {
            List<T> snapshot = List.copyOf(source);
            List<R> result = new ArrayList<>();
            for (T value : snapshot) {
                if (filter.test(value)) {
                    result.add(mapper.apply(value));
                }
            }
            return List.copyOf(result);
        }
    }

    public static void main(String[] args) {
        List<Integer> source = new ArrayList<>(List.of(1, 2, 3, 4));
        Pipeline<Integer, String> pipeline = new Pipeline<>(
                value -> value % 2 == 0,
                value -> "N" + value);
        List<String> first = pipeline.execute(source);
        source.add(6);
        List<String> second = pipeline.execute(source);
        System.out.println(first);
        System.out.println(second);
        System.out.println(source);
    }
}

main은 같은 pipeline을 원본에 6을 추가하기 전과 후에 한 번씩 실행합니다. 마지막 줄은 두 번의 실행이 끝난 뒤의 원본입니다.

[N2, N4]
[N2, N4, N6]
[1, 2, 3, 4, 6]

first는 원본이 [1, 2, 3, 4]일 때의 결과이고, source.add(6) 뒤에 다시 실행한 second에는 N6이 들어 있습니다. 파이프라인은 조건 함수와 변환 함수라는 정의만 들고 있고, 원본은 실행할 때마다 그 순간의 것을 읽기 때문입니다. 실행이 원본을 바꾸지 않으므로 마지막 줄의 원본에는 main에서 직접 추가한 6만 늘었습니다.

List.copyOf()가 고정하는 것은 목록의 구성, 즉 어떤 원소가 어떤 순서로 들어 있는가입니다. 원소 객체까지 복사하지는 않으므로, 원소가 가변 객체라면 실행 도중 다른 코드가 원소의 필드를 바꾸는 것은 파이프라인에도 그대로 보입니다. 또 다른 스레드가 원본을 수정하는 도중에 복사하면 복사 자체가 예외로 끝나거나 어긋난 내용을 읽을 수 있으므로, 여러 스레드가 원본을 함께 쓴다면 별도의 동기화가 필요합니다. 그리고 List.copyOf()는 null 원소가 있으면 NullPointerException을 던진다는 점도 알아 두세요.

상태가 꼭 필요하면 실행마다 새로 만들기

키 기준 중복 제거처럼 상태가 정말로 필요한 경우에는 상태를 없앨 수 없습니다. 대신 상태가 언제 만들어지고 언제 버려지는지를 실행 한 번에 맞추면 됩니다. 핵심은 '규칙의 정의'와 '그 규칙을 한 번 실행하는 데 쓰는 상태'를 나누는 것입니다. 아래 distinctBy는 조건 함수를 바로 돌려주지 않고, 호출할 때마다 새 HashSet을 가진 조건 함수를 만들어 주는 공급자, 즉 Supplier<Predicate<T>>를 돌려줍니다.

import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.function.Function;
import java.util.function.Predicate;
import java.util.function.Supplier;

public final class DistinctByKeyFactory {
    static <T, K> Supplier<Predicate<T>> distinctBy(Function<? super T, ? extends K> keyExtractor) {
        return () -> {
            Set<K> seen = new HashSet<>();
            return value -> seen.add(keyExtractor.apply(value));
        };
    }

    public static void main(String[] args) {
        Supplier<Predicate<String>> byFirstLetter = distinctBy(word -> word.charAt(0));

        List<String> first = List.of("apple", "avocado", "banana").stream()
                .filter(byFirstLetter.get())
                .toList();
        List<String> second = List.of("apricot", "blueberry", "cherry").stream()
                .filter(byFirstLetter.get())
                .toList();
        System.out.println(first);
        System.out.println(second);
    }
}

두 스트림은 같은 byFirstLetter에서 get()을 따로 호출해 조건 함수를 하나씩 얻습니다. 두 번째 입력의 apricot은 첫 입력의 apple과 첫 글자가 같다는 점을 눈여겨보세요.

[apple, banana]
[apricot, blueberry, cherry]

byFirstLetter는 '첫 글자가 같은 단어는 처음 것만 남긴다'는 규칙의 정의일 뿐, 그 자체로는 집합을 갖고 있지 않습니다. 스트림을 실행할 때마다 byFirstLetter.get()을 호출하면 그 순간 새 seen을 가진 조건 함수가 만들어집니다. 그래서 두 번째 실행의 apricot은 첫 실행에서 본 a에 막히지 않습니다. 만약 get()을 한 번만 호출해 얻은 조건 함수를 두 스트림에 함께 넘겼다면, 첫 실행이 seen에 a와 b를 남겨 두 번째 줄은 [cherry]가 됐을 것입니다.

실무 코드에서 자주 보는 filter(distinctByKey(User::id))도 같은 원리로 동작합니다. 이런 distinctByKey 메서드는 호출될 때마다 새 집합을 만들어 조건 함수를 돌려주므로, 스트림 식 안에서 바로 호출하면 그 집합은 그 스트림 하나에서만 쓰입니다. 하지만 그 결과를 필드에 저장해 두고 여러 스트림에서 다시 쓰는 순간, 첫 절의 문제가 그대로 돌아옵니다.

이렇게 만든 조건 함수에도 남는 제한이 하나 있습니다. '처음 나온 것을 남긴다'는 말은 원소가 앞에서부터 하나씩 차례로 들어온다는 전제 위에서만 성립합니다. 그래서 이런 조건 함수는 순차 실행용이라고 이름이나 문서에 밝혀 두고, 병렬 실행에서도 같은 의미가 필요하다면 그 의미를 지켜 주는 연산을 따로 써야 합니다.

더 깊이: 병렬 실행에서 '처음 나온 원소'가 흔들리는 이유

Stream.distinct()는 순서가 있는 스트림이라면 중복된 원소 가운데 만남 순서(encounter order)상 처음 나온 것을 남기도록 정해져 있습니다. 병렬로 실행할 때도 이 약속을 지키기 위해, 스트림 라이브러리는 조각별 결과를 순서에 맞게 합치는 방법을 따로 갖고 있습니다. 직접 만든 조건 함수에는 그런 장치가 없습니다. 집합을 ConcurrentHashMap.newKeySet()으로 바꾸면 여러 스레드가 동시에 add를 호출해도 집합이 깨지지는 않지만, 같은 키를 가진 원소 가운데 먼저 add에 도착한 것이 남습니다. 어느 쪽이 먼저 도착할지는 스레드 실행 순서가 정하므로, 같은 입력에서도 실행할 때마다 다른 원소가 남을 수 있습니다. 스트림 패키지 문서의 무상태 설명도 동기화된 집합을 쓰는 병렬 map을 예로 들며, 같은 입력의 결과가 실행마다 달라질 수 있다고 설명합니다.

JDK 24부터는 이런 상태를 스트림 연산 안에서 다루는 정식 도구도 있습니다. Stream.gather()에 넘기는 Gatherer는 상태를 만드는 초기화 함수(initializer)를 따로 가지며, 파이프라인을 평가할 때마다 이 함수로 상태를 새로 만듭니다. 병렬로 평가하면 조각마다 따로 만듭니다. 이 글에서 직접 만든 '상태를 만드는 공급자'가 API 안으로 들어간 모양입니다. Gatherer는 JDK 22와 23에서 프리뷰를 거쳐 JDK 24에서 정식 기능이 됐습니다(JEP 485).

정리

• 람다가 바깥의 필드나 컬렉션, 카운터를 읽거나 바꾸면 결과가 입력뿐 아니라 지금까지의 처리 이력에 따라 달라집니다. 이런 상태가 실행이 끝난 뒤에도 남으면 같은 함수를 다시 쓸 때 결과가 틀립니다.
• 동시 컬렉션이나 AtomicInteger는 여러 스레드가 동시에 수정할 때 자료구조가 깨지는 것만 막습니다. 지난 실행의 값을 지워 주지도, 처리 순서를 정해 주지도 않습니다.
• 번호처럼 위치로 정해지는 값은 카운터 대신 원본의 인덱스에서 계산합니다.
• 바깥 객체의 필드를 읽는 람다와 this::format 같은 참조는 바깥 객체를 붙잡아 수명을 늘립니다. 사실상 final 규칙은 재대입만 막고 객체의 변경은 막지 않습니다.
• 실행 중에는 스트림의 원본을 수정하지 않습니다. 원본을 바꿔야 한다면 스냅샷을 순회하고 결과는 별도의 목록에 모읍니다.
• 상태가 꼭 필요하면 규칙의 정의와 상태를 나누고, 상태는 실행할 때마다 새로 만듭니다.

연습 문제

1. 실행마다 새 집합을 쓰는 DistinctPipeline

키 추출기를 받는 DistinctPipeline<T, K>를 만드세요. execute(List<T>)를 호출할 때마다 새 Set을 만들어, 같은 파이프라인 인스턴스로 두 번 실행해도 각 입력 안에서만 중복이 제거되어야 합니다. 결과에는 키마다 처음 나온 원소를 원래 순서대로 남깁니다.

import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.function.Function;

public final class PerExecutionDistinctSolution<T, K> {
    private final Function<? super T, ? extends K> keyExtractor;

    PerExecutionDistinctSolution(Function<? super T, ? extends K> keyExtractor) {
        this.keyExtractor = keyExtractor;
    }

    List<T> execute(List<? extends T> source) {
        Set<K> seen = new HashSet<>();
        List<T> result = new ArrayList<>();
        for (T value : source) {
            if (seen.add(keyExtractor.apply(value))) {
                result.add(value);
            }
        }
        return List.copyOf(result);
    }

    record User(String id, String name) {
    }

    public static void main(String[] args) {
        PerExecutionDistinctSolution<User, String> pipeline =
                new PerExecutionDistinctSolution<>(User::id);
        System.out.println(pipeline.execute(List.of(
                new User("1", "A"),
                new User("1", "A2"),
                new User("2", "B"))));
        System.out.println(pipeline.execute(List.of(
                new User("1", "Again"),
                new User("3", "C"))));
    }
}

[User[id=1, name=A], User[id=2, name=B]]
[User[id=1, name=Again], User[id=3, name=C]]

2. 키를 바꾸고 null 키를 넣어 보기

해답의 파이프라인으로 다음 결과를 예상해 보세요. 사용자는 id:name 형식으로 적었습니다.

1. 첫 입력 1:A, 1:A2, 2:B에서 키를 id 대신 name().length()로 바꾸면 무엇이 남나요?
2. id를 키로 쓰는 파이프라인에 null:A, null:B, 1:C를 넣고, 이어서 다시 호출한 execute에 null:D만 넣으면 각각 무엇이 남나요?
3. 입력 목록에 User 대신 null 원소가 들어 있으면 어떻게 되나요?

다음 글

메서드 참조의 네 가지 형태에서는 이 글에서도 몇 번 쓴 User::id나 this::format 같은 메서드 참조를 정식으로 다룹니다. 어떤 메서드 참조는 만들 때 객체를 붙잡고 어떤 것은 붙잡지 않는데, 이 차이가 이 글에서 본 수명 문제와 그대로 이어집니다.