java8 stream和Parallel Streams使用注意事项-一休教程

当java8的第一个早期access版本出现时，最重要的演变似乎是lambdas。这一点现在正在改变，许多开发人员现在似乎认为流是最有价值的Java8特性。这是因为他们相信通过改变程序中的一个单词（用parallelStream替换stream），他们将使这些程序并行工作。许多Java8的传道者已经展示了这方面的惊人例子。这有什么问题吗？不，不是什么。很多事情：

并行运行可能是一个好处，也可能不是。这取决于您使用此功能的目的。
Java8 parallel streams 并行流可以使程序运行得更快。不管怎样。或者更慢。
将流看作是以低成本实现并行处理的一种方法，这将妨碍开发人员理解真正发生的事情。流与并行处理没有直接联系。
上述问题大多是基于一个误解：并行处理和并发处理不是一回事。大多数关于Java8“自动并行化”的例子实际上都是并发处理的例子。
把map、filter和其他操作看作“内部迭代”完全是无稽之谈（尽管这不是java8的问题，但是我们使用它的方式有问题）。

那么，什么是streams流?

根据维基百科：

“流是列表的潜在无限模拟，由归纳定义给出：

data Stream a = Cons a (Stream a)

使用流生成和计算需要延迟求值，或者隐式地使用延迟求值的语言，或者通过创建和强制使用渴望的语言的thunk。”

最重要的一点是，Java是Wikipedia所称的“eager”语言，这意味着Java在评估事物时通常是严格的（而不是懒惰的）。例如，如果您在Java中创建了一个列表，那么在创建列表时将计算所有元素。这可能会让您感到惊讶，因为您可能会创建一个空列表并在之后添加元素。这仅仅是因为列表是可变的（并且您正在用对某物的引用替换空引用），或者您正在从附加新元素的旧列表创建一个新列表。

列表是由产生其元素的东西创建的。例如：

List<Integer> list = Arrays.asList(1, 2, 3, 4, 5);

这里生产者是一个数组，数组的所有元素都经过严格的计算。

也可以以递归的方式创建列表，例如以1开头的列表，其中所有元素都等于1加上前一个元素，小于6。在Java<8中，这可以转化为：

List<Integer> list = new ArrayList<Integer>();
for(int i = 0; i < 6; i++) {
  list.add(i);
}

有人可能会说for循环是Java中罕见的惰性计算示例之一，但结果是一个列表，其中所有元素都被计算。

如果我们想将一个函数应用于这个列表的所有元素，会发生什么？我们可以循环进行。例如，如果希望将所有元素增加2，则可以执行以下操作：

for(int i = 0; i < list.size(); i++) {
  list.set(i, list.get(i) * 2);
}

但是，这不允许使用更改元素类型的操作，例如将所有元素增加10%。以下解决方案解决了此问题：

List<Double> list2 = new ArrayList<Double>();
for(int i = 0; i < list.size(); i++) {
  list2.add(list.get(i) * 1.2);
}

此表单允许对每种语法使用Java 5中的：

List<Double> list2 = new ArrayList<>();
for(Integer i : list) {
  list2.add(i * 1.2);
}

或Java 8语法：

List<Double> list2 = new ArrayList<>();
list.forEach(x -> list2.add(x * 1.2));

到目前为止，还不错。但是如果我们想把这个值增加10%，然后再除以3呢？简单的答案是：

List<Double> list2 = new ArrayList<>();
list.forEach(x -> list2.add(x * 1.2));
List<Double> list3 = new ArrayList<>();
list2.forEach(x -> list3.add(x / 3));

这远不是最佳的，因为我们在列表上迭代了两次。更好的解决方案是：

List<Double> list2 = new ArrayList<>();
for(Integer i : list) {
  list2.add(i * 1.2 / 3);
}

暂时把自动装箱/拆箱问题放在一边。在Java 8中，这可以写成：

List<Double> list2 = new ArrayList<>();
list.forEach(x -> list2.add(x * 1.2 / 3));

但是等等。。。这只可能是因为我们看到绑定到列表的使用者的内部结构，所以我们可以手动组合操作。如果我们有：

List<Double> list2 = new ArrayList<>();
list.forEach(consumer1);
List<Double> list3 = new ArrayList<>();
list2.forEach(consumer2);

我们怎么会知道怎么写呢？不可能。在Java8中，使用者接口有一个默认方法，然后。我们很可能会用这种方式来组成消费者：

list.forEach(consumer1.andThen(consumer2));

但这将导致一个错误，因为和被定义为：

default Consumer<T> andThen(Consumer<? super T> after) {
  Objects.requireNonNull(after);
  return (T t) -> { accept(t); after.accept(t); };
}

这意味着我们不能使用and来组合不同类型的消费者。

事实上，我们从一开始就错了。我们需要的是将列表绑定到函数以获得新列表，例如：

Function<Integer, Double> function1 = x -> x * 1.2;
Function<Double, Double> function2 = x -> x / 3;
list.bind(function1).bind(function2);

其中bind方法将在一个特殊的FList类中定义，如：

public class FList<T> {
  final List<T> list;

  public FList(List<T> list) {
    this.list = list;
  }

  public <U> FList<U> bind(Function<T, U> f) {
    List<U> newList = new ArrayList<U>();
    for (T t : list) {
      newList.add(f.apply(t));
    }
    return new FList<U>(newList);
  }
}

我们将在下面的例子中使用它：

new Flist<>(list).bind(function1).bind(function2);

我们遇到的唯一问题是绑定两次需要在列表上迭代两次。这是因为bind是严格计算的。我们需要的是一个懒惰的评估，这样我们就可以只迭代一次。

这里的问题是bind方法不是真正的绑定。它实际上是一种真正的约束和还原的组合。”“reducting”是对列表中的每个元素应用一个操作，从而将此元素与应用于前一个元素的相同操作的结果结合起来。因为从第一个元素开始时没有前一个元素，所以我们从一个初始值开始。例如，applying（x）->r+x，其中r是对上一个元素的操作结果，或对第一个元素应用0，得到列表中所有元素的总和。对每个元素applying（）->r+1，从r=0开始给出列表的长度。（这可能不是获得列表长度的更有效的方法，但它完全是功能性的！）

这里，操作是add（element），初始值是一个空列表。这仅仅是因为函数应用程序是经过严格评估的。

Java8流提供给我们的是相同的，但是是延迟计算的，这意味着在将函数绑定到流时，不涉及迭代！

将一个Function<T，U>绑定到一个Stream<T>会得到一个没有迭代的Stream。结果流不会被评估，这并不取决于初始流是用评估或未评估的数据构建的。

在函数语言中，将Function<T，U>绑定到Stream<T>本身就是一个函数。在Java8中，它是一个方法，这意味着它的参数是严格求值的，但这与结果流的求值无关。为了理解正在发生的事情，我们可以想象要绑定的函数存储在某个地方，它们成为新的（未计算的）结果流的数据生成器的一部分。

在Java8中，将FunctionT->U绑定到Stream<T>，从而产生Stream的方法称为map。将FunctionT->Stream绑定到Stream<T>，从而产生Stream的函数称为flatMap。

flatten在哪里？

大多数函数式语言还提供了一个将Stream<Stream转换为Stream的flatten扁平函数，但Java8流中缺少这个函数。这看起来并不是什么大麻烦，因为定义一个方法来实现这一点非常容易。例如，给定以下函数：

Function<Integer, Stream<Integer>> f = x -> Stream.iterate(1, y -> y + 1).limit(x);

Stream<Integer> stream = Stream.iterate(1, x -> x + 1);
Stream<Integer> stream2 = stream.limit(5).flatMap(f);

System.out.println(stream2.collect(toList()))

生产：

[1, 1, 2, 1, 2, 3, 1, 2, 3, 4, 1, 2, 3, 4, 5]

使用map而不是flatMap：

Stream<Integer> stream = Stream.iterate(1, x -> x + 1);
Stream<Integer> stream2 = stream.limit(5).map(f);

System.out.println(stream2.collect(toList()))

将产生一个流：

[java.util.stream.SliceOps$1@12133b1, java.util.stream.SliceOps$1@ea2f77,
java.util.stream.SliceOps$1@1c7353a, java.util.stream.SliceOps$1@1a9515, 
java.util.stream.SliceOps$1@f49f1c]

使用函数范式将这个整数流转换为整数流非常简单：只需将标识函数平面映射到它：

System.out.println(stream2.flatMap(x -> x).collect(toList()));

然而奇怪的是，没有向流中添加展平方法，因为知道map、flatMap、unit和flatten之间的强关系，其中unit是从T到Stream<T>的函数，由以下方法表示：

Stream<T> Stream.of(T... t)

何时评估stream流？

当我们对流应用一些称为终端操作的特定操作时，流被评估。这只能做一次。一旦终端操作应用于流，它就不再可用。终端操作包括：

forEach

forEachOrdered

toArray

reduce

collect

min

max

count

anyMatch

allMatch

noneMatch

findFirst

findAny

iterator

spliterator

有些方法是短路。例如，一旦找到第一个元素，findFirst就会返回。

非终端操作称为中间操作，可以是有状态的（如果元素的计算依赖于前一个元素的计算）或无状态的。中间操作包括：

filter

map

mapTo... (Int, Long or Double)

flatMap

flatMapTo... (Int, Long or Double)

distinct

sorted

peek

limit

skip

sequential

parallel

unordered

onClose

可以对一个流应用多个中间操作，但是只能使用一个终端操作。

那么并行处理呢？

自动并行通常不会给出预期的结果，至少有两个原因：

1. 速度的提高在很大程度上取决于任务类型和并行化策略。总的来说，最好的策略取决于任务的类型。

2. 速度的提高在很大程度上取决于环境。在某些环境中，通过并行化很容易降低速度。

无论什么样的任务要并行化，并行流应用的策略都是相同的，除非您自己设计这个策略，这将消除并行流的许多兴趣。并行化需要：

执行子任务的线程池，
将初始任务划分为子任务，
将子任务分配给线程，
整理结果。

不输入细节，所有这些都意味着一些开销。它将显示惊人的结果时：

有些任务意味着长时间阻塞，例如访问远程服务，或者
没有多少线程同时运行，特别是没有其他并行流。

如果所有子任务都意味着高强度的计算，那么潜在的增益将受到可用处理器数量的限制。默认情况下，Java8将使用与计算机上的处理器相同数量的线程，因此，对于密集型任务，结果高度依赖于其他线程可能同时执行的操作。当然，如果每个子任务基本上都在等待，那么收益可能看起来是巨大的。

最坏的情况是应用程序与其他应用程序一起在服务器或容器中运行，并且子任务并不意味着等待。在这种情况下（例如在J2EE服务器上运行），并行流通常比串行流慢。想象一下，一台服务器每秒处理数百个请求。同时评估多个流的可能性很大，因此这项工作已经并行化了。在业务级别上新的并行化层很可能会使事情变得更慢。

最糟糕的：业务应用程序很有可能看到开发环境的速度提高和生产的减少。这是最糟糕的情况。

Streams流有什么好处

Stream流是一个有用的工具，因为它们允许延迟计算。这在几个方面非常重要：

它们允许使用绑定的函数式编程风格。
它们允许通过删除迭代来获得更好的性能。迭代随着求值而发生。使用流，我们可以绑定几十个函数而无需迭代。
它们允许简单的任务并行化，包括长时间的等待。
Stream流可能是无限的（因为它们是懒惰的）。函数可以毫无问题地绑定到无限流。经过评估，一定有办法使它们有限。这通常是通过短路操作完成的。

什么Stream流不好

在处理密集型计算任务时，应高度谨慎地使用流。特别是，默认情况下，所有流将使用相同的ForkJoinPool，配置为使用与运行程序的计算机中的内核数量相同的线程。

如果对一个并行流的求值导致一个运行时间很长的任务，那么可以将其拆分为尽可能多的长时间运行的子任务，这些子任务将分发到池中的每个线程。从那里，没有其他并行流可以被处理，因为所有线程都将被占用。因此，对于计算密集型流评估，应该始终使用特定的ForkJoinPool，以免阻塞其他流。

为此，可以从流中创建一个Callable并将其提交到池中：

List<SomeClass> list = // A list of objects
Stream<SomeClass> stream = list.parallelStream().map(this::veryLongProcessing);
Callable<List<Integer>> task = () -> stream.collect(toList());
ForkJoinPool forkJoinPool = new ForkJoinPool(4);
List<SomeClass> newList = forkJoinPool.submit(task).get()

这样，其他并行流（使用它们自己的ForkJoinPool）就不会被这个流阻塞。换句话说，我们需要一个ForkJoinPool池来避免这个问题。

如果程序要在容器中运行，那么在使用并行流时必须非常小心。在这种情况下不要使用默认池，除非您确定容器可以处理它。在javaee容器中，不要使用并行流。

java8 stream和Parallel Streams使用注意事项

那么，什么是streams流?

flatten在哪里？

何时评估stream流？

那么并行处理呢？

Streams流有什么好处

什么Stream流不好

文章展示

Java9中新增的Collector

Java线程优先级

Java正则表达式API系列一

Java8接口–默认方法和静态方法

Java中的按值调用和按引用调用

Java 8中的Collectors API详解

排行榜展示

hs2姿势mod教程（hs2解锁姿势插件）

如何将Windows10电脑的默认下载位置更改为D盘

华夏网赚论坛（华夏免费版论坛）

哪里可以写黄文赚钱（哪里可以写黄文赚钱的软件）

上传91视频可以赚钱吗（91上传作品能赚钱吗）

美国赚钱可以拿回中国用吗（美国挣得钱能拿回中国吗）

java8 stream和Parallel Streams使用注意事项

那么，什么是streams流?

flatten在哪里？

何时评估stream流？

那么并行处理呢？

Streams流有什么好处

什么Stream流不好

相关文章

文章展示

排行榜展示