Java Stream相当于ReactiveX Observable＃scan

Question

ReactiveX有一个名为Scan的Very Neat运算符，它类似于reduce，只是它发出每个中间累加器。我将如何使用Java Streams做到这一点？Stream#reduce不是我想要的，因为它返回T：T reduce(T identity, BinaryOperator<T> accumulator)我想要返回Stream<T>的是流的每个项目，T它们是每次调用时都返回的accumulator：Stream.of(1, 2, 3)    .myScan(0, (accumulator, i) -> accumulator + i)    .collect(Collectors.toList()); // [1, 3, 6]我可以做一些reducing脚的事情，例如减少List，然后再转换回Stream，但这很丑陋。

Answer 1

流不支持此操作。您可以将其转换为迭代器，在其中进行迭代，然后转换回流，但是流没有内置的功能可以执行此操作。

Answer 2

尝试StreamEx。我认为它提供了您想要的提取API：

List<Integer> result = StreamEx.of(1, 2, 3).scanLeft((i, j) -> i + j);

System.out.println(result); // [1, 3, 6]

Answer 3

您可以尝试从其他不受欢迎的副作用中受益。由于您的目标是处理流并继续处理流，因此我们可以使用该map方法来实现这一点。以下代码利用了副作用。

public class Scan {

  public static void main(String[] args) {

    Accumulator accumulator = new Accumulator();

    Stream<Integer> originalStream = Stream.of(1, 2, 3);

    Stream<Integer> scannedStream = originalStream

      .map(i -> accumulator.accumulate(i));

    List<Integer> list = scannedStream

      .collect(Collectors.toList()); // [1, 3, 6]

    for (Integer i : list) {

      System.out.println(i);

    }

  }

  private static class Accumulator {

    private int value;

    public int accumulate(int i) {

      return value += i;

    }

  }

}

所述Accumulator可以通过替换Function以用于不同的扫描-operations。

但是您必须意识到限制和担忧（例如线程安全性）。