能否实现可在获取结果时提前终止流处理的Java Collector？

Great question! The short answer is yes, but it’s not straightforward with the standard Collector API alone—collectors are inherently designed to accumulate all elements in a stream. However, we can use a combination of stream short-circuiting operations or custom implementations to stop processing as soon as we have a definitive result (like hitting a NaN during average calculation).

Let’s walk through practical solutions for your example:

1. Simple Approach: Check for Termination Condition First #

The easiest way to handle your average/NaN scenario is to first check if the stream contains a NaN, and only calculate the average if it doesn’t. Since streams can’t be reused, wrap your stream in a Supplier to generate a fresh stream for each operation:

// Wrap the stream in a Supplier to avoid "stream has already been operated upon" errors
Supplier<Stream<Double>> streamSupplier = () -> Stream.of(1.0, 2.0, Double.NaN, 3.0, 4.0);

// Check if any element is NaN (short-circuits at the first NaN)
boolean hasNaN = streamSupplier.get().anyMatch(Double::isNaN);

Double result;
if (hasNaN) {
    result = Double.NaN;
} else {
    // Calculate average only if there are no NaNs
    result = streamSupplier.get().average().orElse(Double.NaN);
}

System.out.println(result); // Outputs NaN

This uses anyMatch, a short-circuiting terminal operation that stops traversing the stream as soon as it finds a NaN. The tradeoff is it traverses the stream twice (once for the check, once for the average), but this is acceptable for most use cases unless your stream elements are expensive to generate.

2. Advanced: Custom Collector with Short-Circuiting Spliterator #

If you need to avoid traversing the stream twice and want to truly stop processing as soon as the termination condition is met, you can combine a custom Spliterator with a collector. The Spliterator will control when the stream stops, and the collector will track the state.

Step 1: Custom Spliterator to Handle Termination #

This spliterator checks a termination flag before advancing to the next element:

class ShortCircuitSpliterator<T> implements Spliterator<T> {
    private final Spliterator<T> source;
    private volatile boolean terminated;

    public ShortCircuitSpliterator(Spliterator<T> source) {
        this.source = source;
    }

    public void terminate() {
        this.terminated = true;
    }

    @Override
    public boolean tryAdvance(Consumer<? super T> action) {
        if (terminated) {
            return false; // Stop stream processing
        }
        return source.tryAdvance(action);
    }

    @Override
    public Spliterator<T> trySplit() {
        return null; // Disable parallel processing for simplicity
    }

    @Override
    public long estimateSize() {
        return source.estimateSize();
    }

    @Override
    public int characteristics() {
        return source.characteristics() & ~Spliterator.SIZED;
    }
}

Step 2: Custom Collector for Average Calculation #

This collector tracks the sum and count of elements, and triggers the spliterator to terminate if a NaN is encountered:

import java.util.Collections;
import java.util.Set;
import java.util.function.*;
import java.util.stream.Collector;

class ShortCircuitAverageCollector implements Collector<Double, double[], Double> {
    private final ShortCircuitSpliterator<Double> spliterator;

    public ShortCircuitAverageCollector(ShortCircuitSpliterator<Double> spliterator) {
        this.spliterator = spliterator;
    }

    @Override
    public Supplier<double[]> supplier() {
        return () -> new double[2]; // Index 0 = count, Index 1 = sum
    }

    @Override
    public BiConsumer<double[], Double> accumulator() {
        return (acc, value) -> {
            if (Double.isNaN(value)) {
                spliterator.terminate(); // Stop stream immediately
                acc[1] = Double.NaN;
                return;
            }
            if (!Double.isNaN(acc[1])) {
                acc[0]++;
                acc[1] += value;
            }
        };
    }

    @Override
    public BinaryOperator<double[]> combiner() {
        return (a, b) -> {
            if (Double.isNaN(a[1]) || Double.isNaN(b[1])) {
                return new double[]{1, Double.NaN};
            }
            return new double[]{a[0] + b[0], a[1] + b[1]};
        };
    }

    @Override
    public Function<double[], Double> finisher() {
        return acc -> {
            if (Double.isNaN(acc[1])) {
                return Double.NaN;
            }
            return acc[0] == 0 ? Double.NaN : acc[1] / acc[0];
        };
    }

    @Override
    public Set<Characteristics> characteristics() {
        return Collections.emptySet();
    }
}

Step 3: Use the Custom Collector #

import java.util.stream.Stream;
import java.util.stream.StreamSupport;

public class Main {
    public static void main(String[] args) {
        Stream<Double> originalStream = Stream.of(1.0, 2.0, Double.NaN, 3.0, 4.0);
        ShortCircuitSpliterator<Double> spliterator = new ShortCircuitSpliterator<>(originalStream.spliterator());
        Stream<Double> shortCircuitStream = StreamSupport.stream(spliterator, false);

        Double result = shortCircuitStream.collect(new ShortCircuitAverageCollector(spliterator));
        System.out.println(result); // Outputs NaN, and stops processing after the NaN
    }
}

This approach processes the stream only once and stops immediately when a NaN is found. Note that we disabled parallel processing; adding parallel support would require thread-safe state management (like AtomicBoolean for the termination flag).

Key Takeaways #

• The standard Collector API doesn’t natively support short-circuiting, but we can pair it with short-circuiting stream operations (like anyMatch) for simple use cases.
• For true early termination, custom Spliterator + collector combinations work, but add complexity.
• For Java 9+, you could also use takeWhile to stop the stream before calculating the average, but you’d still need to check if the stream was terminated early to return NaN instead of the partial average.

内容的提问来源于stack exchange，提问作者Michael Kay