java - 在 Java 8 Streams 上实现自定义中间操作

Question

我正在尝试研究如何在 Java 8 Stream 上实现自定义中间操作。看来我被锁在门外了:(

具体来说，我想获取一个流并返回每个条目，直到 并包括 第一个具有特定值的条目。之后我想停止生成任何东西 - 使其短路。

它正在对输入数据运行一系列验证检查。我想在出现第一个错误时停止，如果有的话，但我想在途中整理警告。而且因为这些验证检查可能很昂贵 - 例如涉及数据库查找 - 我只想运行所需的最小集合。

所以代码应该是这样的:

Optional<ValidationResult> result = validators.stream()
    .map(validator -> validator.validate(data))
    .takeUntil(result -> result.isError()) // This is the bit I can't do
    .reduce(new ValidationResult(), ::mergeResults);

似乎我应该能够使用 ReferencePipeline.StatefulOp 做一些事情，除了它是所有包范围，所以我不能扩展它。所以我想知道实现这一目标的正确方法是什么？或者是否有可能？

另请注意 - 这需要在 Java 8 中，而不是 9+，因为我们出于各种不相关的原因还没有做到这一点。

干杯

Answer 1

一般来说，自定义操作会需要和Spliterator接口(interface)打交道。它通过添加特征和大小信息以及将元素的一部分拆分为另一个拆分器(因此得名)的能力扩展了 Iterator 的概念。它还通过只需要一种方法来简化迭代逻辑。

public static <T> Stream<T> takeWhile(Stream<T> s, Predicate<? super T> condition) {
    boolean parallel = s.isParallel();
    Spliterator<T> spliterator = s.spliterator();
    return StreamSupport.stream(new Spliterators.AbstractSpliterator<T>(
        spliterator.estimateSize(),
        spliterator.characteristics()&~(Spliterator.SIZED|Spliterator.SUBSIZED)) {
            boolean active = true;
            Consumer<? super T> current;
            Consumer<T> adapter = t -> {
                if((active = condition.test(t))) current.accept(t);
            };

            @Override
            public boolean tryAdvance(Consumer<? super T> action) {
                if(!active) return false;
                current = action;
                try {
                    return spliterator.tryAdvance(adapter) && active;
                }
                finally {
                    current = null;
                }
            }
        }, parallel).onClose(s::close);
}

为了保持流的属性，我们首先查询并行状态，为新流重新建立它。此外，我们注册了一个将关闭原始流的关闭操作。

主要工作是实现一个Spliterator来装饰之前流状态的spliterator。

除了 SIZED 和 SUBSIZED 之外，这些特征都被保留了下来，因为我们的操作导致了不可预测的大小。原始大小仍然通过，现在将用作估计值。

此解决方案在操作期间存储传递给 tryAdvance 的 Consumer，以便能够使用相同的适配器消费者，避免为每个消费者创建一个新消费者迭代。这是有效的，因为它保证永远不会同时调用 tryAdvance。

并行性是通过拆分完成的，它继承自AbstractSpliterator。这种继承的实现将缓冲一些元素，这是合理的，因为为像 takeWhile 这样的操作实现更好的策略确实很复杂。

所以你可以像这样使用它

    takeWhile(Stream.of("foo", "bar", "baz", "hello", "world"), s -> s.length() == 3)
        .forEach(System.out::println);

将打印

foo
bar
baz

或

takeWhile(Stream.of("foo", "bar", "baz", "hello", "world")
    .peek(s -> System.out.println("before takeWhile: "+s)), s -> s.length() == 3)
    .peek(s -> System.out.println("after takeWhile: "+s))
    .forEach(System.out::println);

将打印

before takeWhile: foo
after takeWhile: foo
foo
before takeWhile: bar
after takeWhile: bar
bar
before takeWhile: baz
after takeWhile: baz
baz
before takeWhile: hello

这表明它没有处理超出必要的部分。在 takeWhile 阶段之前，我们必须遇到第一个不匹配的元素，之后，我们只会遇到直到那个的元素。