Java 8：停止检查所有流元素的缩减操作

Question

我试图了解是否有一种方法可以在不检查整个流程的情况下终止减少操作，我无法找到一种方法。

用例大致如下：让有一长串需要折叠成Accumulator的整数s。每个元素检查都可能很昂贵，因此在Accumulator中，我对传入的Accumulator执行检查，看看我们是否需要执行昂贵的操作-如果我们不需要，那么我只需返回累加器。

这对于小列表（er）显然是一个很好的解决方案，但大列表会产生不必要的流元素访问成本，我希望避免这种情况。

这里是一个代码草图-仅假设序列缩减。

class Accumulator {
    private final Set<A> setA = new HashSet<>;
    private final Set<B> setB = new HashSet<>;
}

class ResultSupplier implements Supplier<Result> {

    private final List<Integer> ids;

    @Override
    public Result get() {
        Accumulator acc = ids.stream().reduce(new Accumulator(), f(), (x, y) -> null);

        return (acc.setA.size > 1) ? Result.invalid() : Result.valid(acc.setB);
    }

    private static BiFunction<Accumulator, Integer, Accumulator> f() {
        return (acc, element) -> {
            if (acc.setA.size() <= 1) {
                // perform expensive ops and accumulate results
            }
            return acc;
        };
    }
}

除了必须遍历整个流之外，我还不喜欢另一个事实——我必须检查两次相同的条件（即setA大小检查）。

我考虑过map（）和collect（）操作，但它们看起来更像是相同的，没有发现它们实质性地改变了这样一个事实，即如果不检查整个流，我就无法完成折叠操作。

此外，我的想法是，想象中的takeWhile（p：（A）=

请记住，我是FP的新手，所以——有没有办法让这个工作像我预期的那样？我是否设置了整个问题，或者这是设计上的限制？




Answer 1

没有真正的FP解决方案，仅仅是因为您的整个累加器不是FP。在这方面我们无法帮助您，因为我们不知道它实际上在做什么。我们所看到的是它依赖于两个可变集合，因此不能成为纯FP解决方案的一部分。

如果您接受这些限制，并且没有干净的方法来使用流API，那么您可能会努力使用简单的方法。这种简单的方法包含一个有状态的谓词，它不是最好的，但有时是不可避免的：

public Result get() {
    int limit = 1;
    Set<A> setA=new HashSet<>();
    Set<B> setB=new HashSet<>();
    return ids.stream().anyMatch(i -> {
        // perform expensive ops and accumulate results
        return setA.size() > limit;
    })? Result.invalid(): Result.valid(setB);
}

但我想指出的是，考虑到您的特定逻辑，即当集合变得太大时，您的结果被视为无效，您尝试不处理太多的元素是对错误情况的优化。你不应该浪费精力去优化它。如果有效结果是处理所有元素的结果，则处理所有元素…

Answer 2

当然，会有一个有趣的、纯粹的FP答案，它可能会帮助以您想要的方式解决这个问题。

与此同时，当简单的解决方案在实用主义上势在必行并且您的原始数据源是List时，为什么还要使用FP呢？无论如何，它已经完全实现了，并且您将使用串行缩减，而不是并行缩减。改写这个：

@Override
public Result get() {
    Accumulator acc = new Accumulator();

    for (Integer id : ids) {
        if (acc.setA.size() <= 1) {
            // perform expensive ops and accumulate results
        }

        // Easy:
        if (enough)
            break;
    }

    return (acc.setA.size > 1) ? Result.invalid() : Result.valid(acc.setB);
}

Answer 3

匿名用户

而不是从ID开始。stream（）您可以

1. 使用ID。拆分器（）
2. 将生成的拆分器包装到具有易失性布尔标志的自定义拆分器中
3. 如果标志更改，请让自定义拆分器的tryAdvance返回false
4. 将自定义拆分器转换为具有流支持的流。流（分离器

添加一些静态助手方法以保持其功能。

由此产生的API可以看到

Accumulator acc = terminateableStream(ids, (stream, terminator) ->
   stream.reduce(new Accumulator(terminator), f(), (x, y) -> null));

此外，我的想法是想象中的takeWhile（p：（A）=

如果条件依赖于累加器状态而不是流成员，它确实有效。这基本上是我上面概述的方法。

JDK提供的takeWhile中可能会禁止它，但是使用拆分器的自定义实现可以自由地采用有状态的方法。