5. Java 8 中的 Stream<T> 类型
5.1. 示例-01 - Stream 类
对 Observable 流的操作与 Stream 流有许多相似之处。一个区别在于，Stream 中的元素必须等到整个流获取完毕后才能进行处理，而 Observable 流中的元素在获取到后即可立即进行处理（观察），无需等待整个 Observable 流的获取完成。 另一个区别在于，获取 Stream 后，其值是通过从 Stream 中逐个拉取来使用的。而对于 Observable，情况则不同。一旦它发出一个值，该值就会被推送到其订阅者那里。
有多个类实现了流（Stream）的概念。这里我们介绍 Stream<T> 类：
Stream 类提供了 39 个方法。我们将介绍其中几个。请看以下代码：
 |
package dvp.java8.streams;
import java.util.List;
import dvp.data.Personne;
import dvp.data.Personnes;
public class Exemple01 {
public static void main(String[] args) {
// list of persons
List<Personne> personnes = Personnes.get();
// display 1
personnes.stream().forEach(p -> {
System.out.println(p);
});
System.out.println("----------------");
// display 2
personnes.stream().forEach(System.out::println);
}
}
- 第 11 行：我们实例化了一个人员列表；
- 第 13 行：从该列表创建一个 Stream。所有集合都可以通过这种方式转换为 Stream。这使我们能够利用该类的所有方法，从而比使用循环更简洁地处理集合中的元素。此外，在可能的情况下，我们还可以利用元素的并行处理；
- 第 13 行：[Stream.forEach] 方法的签名如下：
 |
我们可以看到，该方法的参数是第 4.4 节中介绍的函数式接口 [Consumer<T>]——这是一个仅包含一个方法的接口，该方法作用于类型 T 且不返回任何值。
- 在代码中：
personnes.stream().forEach(p -> {
System.out.println(p);
});
- [people.stream()] 生成一个包含 [Person] 类型元素的流，该流被传递给 [forEach] 方法。参数 p 的类型为 [Person]，而提供的 lambda 函数会打印该人的信息；
上述代码可简化如下（第 18 行）：
personnes.stream().forEach(System.out::println);
我们不再将 lambda 函数的值作为参数传递，而是传递现有方法的引用，本例中即 System.out 类的 println 方法。当然，该方法必须具有正确的签名，本例中即 [Consumer.accept] 方法的签名：void accept(T t)。如前所述，[accept] 方法的参数类型为 [Person]；
我们得到以下结果：
一旦数据流被处理过，就无法再次处理。若需再次处理，必须重新构建该数据流。以下代码 [Example01b] 演示了这一过程：
package dvp.java8.streams;
import java.util.stream.Stream;
import dvp.data.Personne;
import dvp.data.Personnes;
public class Exemple01b {
public static void main(String[] args) {
// people flows
Stream<Personne> personnes = Personnes.get().stream();
// display 1
personnes.forEach(p -> {
System.out.println(p);
});
System.out.println("----------------");
// display 2
personnes.forEach(System.out::println);
}
}
- 第 11 行：为了优化代码，我们决定只构建一次流。所得结果如下：
{"nom":"jean","age":20,"poids":70.0,"sexe":"HOMME"}
{"nom":"marie","age":10,"poids":30.0,"sexe":"FEMME"}
{"nom":"camille","age":30,"poids":55.0,"sexe":"FEMME"}
----------------
Exception in thread "main" java.lang.IllegalStateException: stream has already been operated upon or closed
at java.util.stream.AbstractPipeline.sourceStageSpliterator(Unknown Source)
at java.util.stream.ReferencePipeline$Head.forEach(Unknown Source)
at dvp.java8.streams.Exemple02b.main(Exemple02b.java:18)
每次使用流时，都必须对其进行初始化，即使该流之前已被初始化过。
5.2. 示例-02 - 流中元素的并行处理
 |
请看以下代码：
package dvp.java8.streams;
import java.util.List;
import dvp.data.Personne;
import dvp.data.Personnes;
public class Exemple02 {
public static void main(String[] args) {
// list of persons
List<Personne> personnes = Personnes.get();
// display 1
personnes.stream().forEach(Exemple02::affiche);
System.out.println("-----------------");
// display 2
personnes.stream().parallel().forEach(Exemple02::affiche);
}
public static void affiche(Personne p) {
System.out.printf("Personne %s sur thread %s%n", p, Thread.currentThread().getName());
}
}
- 第 19–21 行：[display] 方法将一个人的 JSON 字符串以及执行显示操作的线程名称打印到控制台；
- 第 13 行：显示人员列表。请注意，[forEach] 方法的参数是前一个静态方法的引用；
- 第 16 行：我们执行相同操作，但使用 [parallel] 方法时，要求流中的元素在多个线程中并行处理。并非所有处理都能并行进行。在此，我们必须假设显示顺序无关紧要，因为在并行处理中，线程的执行顺序无法保证。另请注意一种在流（Streams）和可观察对象（Observables）中将无处不在的语法：
- （待续）
- 流生成元素 e1，这些元素被传递给 m1 方法；
- flux.m1 进而成为一个元素流 e2，该流为 m2 方法提供数据；
- flux.m1.m2 是一个元素流 e3，该流将数据传递给 m3 方法；
随着初始流经过处理，元素 e1、e2、e3 的类型可能会发生变化。
执行此代码将得到以下结果：
我们可以看到，并行执行（第 5–7 行）发生在三个不同的线程上，且并未遵循第 1–3 行中元素的顺序。在本文中，我们不会过多关注流中元素的并行处理，因为这需要讨论使此类处理成为可能的条件。 随后我们发现，能够并行执行的操作寥寥无几。其中一项天然适合并行处理的操作是计算流中数值元素的总和，下面我们将对此进行演示。
5.3. 示例-03 - 流元素的并行处理
 |
请看以下代码（示例 03a）：
package dvp.java8.streams;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
public class Exemple03a {
public static void main(String[] args) {
final long limite = 10_000_000L;
// number of processors
System.out.printf("La JVM a détecté [%s] coeurs sur votre machine%n", Runtime.getRuntime().availableProcessors());
// list of numbers
long début = new Date().getTime();
List<Long> nombres = new ArrayList<>();
for (long i = 0; i < limite; i++) {
nombres.add(i);
}
System.out.printf("création de la liste des %s nombres en %s ms%n", limite, new Date().getTime() - début);
// sum of numbers - sequential method
début = new Date().getTime();
long somme = nombres.stream().reduce(0L, (s, i) -> s + i);
System.out.printf("somme séquentielle : somme=%s, durée (ms)=%s%n", somme, new Date().getTime() - début);
}
}
- 在第 22 行，我们使用了 [reduce] 方法，其签名如下：
 |  |
- [reduce] 方法处理类型为 T 的元素；
- [reduce] 方法对流中的所有元素应用相同的处理：累加器的初始值作为第一个参数提供。实现函数式接口 [BinaryOperator] [2] 的方法作为第二个参数提供：该方法基于每个元素和累加器，返回累加器的新值。累加器的最终值即为 [reduce] 方法返回的值。 代码 [3] 演示了这一机制。[apply] 方法是函数式接口 [BinaryOperator] [2] 的方法；
让我们回到示例代码：
- 第 12 行：我们显示 JVM 检测到的核心数；
- 第 15–18 行：创建一个包含 1000 万个数字的列表；
- 第 22 行：使用单线程顺序计算这些数字的和；
我们得到以下结果：
现在，让我们将代码的第 22 行替换为以下内容（示例 03b）：
long somme = nombres.stream().parallel().reduce(0L, (s, i) -> s + i);
我们指示流中的元素使用多个线程并行处理。之所以能这样做，是因为数字相加的顺序并不重要。因此，我们可以将 n1 个数字分配给线程 T1，n2 个数字分配给线程 T2，……最后将这些不同线程提供的结果相加。随后，我们将得到以下结果：
因此，性能提升几乎可以忽略不计。在接下来的示例中，这种情况经常会出现。线程管理本身就非常耗时。每个核心执行的操作必须足够复杂，才能使性能提升变得明显。以下示例（Example03c）对此进行了说明：
package dvp.java8.streams;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.function.BinaryOperator;
public class Exemple03c {
public static void main(String[] args) {
final long limite = 10_000L;
// number of processors
System.out.printf("La JVM a détecté [%s] coeurs sur votre machine%n", Runtime.getRuntime().availableProcessors());
// list of numbers
long début = new Date().getTime();
List<Long> nombres = new ArrayList<>();
for (long i = 0; i < limite; i++) {
nombres.add(i);
}
System.out.printf("création de la liste des %s nombres en %s ms%n", limite, new Date().getTime() - début);
// sum of numbers - sequential method
début = new Date().getTime();
BinaryOperator<Long> bo = (s, i) -> {
try {
Thread.sleep(1);
} catch (InterruptedException e) {
}
return s + i;
};
long somme = nombres.stream().reduce(0L, bo);
System.out.printf("somme séquentielle : somme=%s, durée (ms)=%s%n", somme, new Date().getTime() - début);
}
}
- 第 30 行：我们再次使用 [reduce] 方法，并将第 23–29 行中的方法引用作为参数传递给它；
- 第 28 行：[bo] 方法返回其两个参数的和；
- 第24–27行：人为地让线程等待1毫秒，以模拟高强度工作；
随后我们得到以下结果：
现在，如果我们将第 30 行替换为以下内容：
long somme = nombres.stream().parallel().reduce(0L, bo);
我们将得到以下结果：
我们可以清楚地看到，通过并行执行求和计算所获得的性能提升。对于处理 8 个数字：
- 顺序线程等待8次1毫秒，即8毫秒；
- 8个并行线程（为简化起见）同时各等待1毫秒，因此处理这8个数字总共耗时1毫秒；
因此，我们可以预期并行执行的速度将比顺序执行快 8 倍。本例中的情况大致如此。
5.4. 示例-04 - 过滤数据流
 |
请看以下代码：
package dvp.java8.streams;
import java.util.List;
import dvp.data.Personne;
import dvp.data.Personnes;
public class Exemple04 {
public static void main(String[] args) {
// list of persons
List<Personne> personnes = Personnes.get();
// displays
System.out.println("age < 28 ----------------------");
personnes.stream().filter(p -> p.getAge() < 28).forEach(p -> {
System.out.println(p);
});
System.out.println("poids < 50 ----------------------");
personnes.stream().filter(p -> p.getPoids() < 50).forEach(p -> {
System.out.println(p);
});
System.out.println("age < 28 ----------------------");
personnes.stream().filter(p -> p.getAge() < 28).forEach(System.out::println);
System.out.println("poids < 50 ----------------------");
personnes.stream().filter(p -> p.getPoids() < 50).forEach(System.out::println);
}
}
- 第 14 行：[Stream.filter] 方法的签名如下：
 |
- [filter] 方法期望作为参数接收第 4.2 节中介绍的功能接口 [Predicate] 的一个实例，该接口需要实现的唯一方法是：boolean test(T t);
- [filter] 方法返回满足 Predicate 条件的流中的元素。因此，它用于过滤流；
请看以下代码：
package dvp.java8.streams;
import java.util.List;
import dvp.data.Personne;
import dvp.data.Personnes;
public class Exemple04 {
public static void main(String[] args) {
// list of persons
List<Personne> personnes = Personnes.get();
// displays
System.out.println("age < 28 ----------------------");
personnes.stream().filter(p -> p.getAge() < 28).forEach(p -> {
System.out.println(p);
});
System.out.println("poids < 50 ----------------------");
personnes.stream().filter(p -> p.getPoids() < 50).forEach(p -> {
System.out.println(p);
});
System.out.println("age < 28 ----------------------");
personnes.stream().filter(p -> p.getAge() < 28).forEach(System.out::println);
System.out.println("poids < 50 ----------------------");
personnes.stream().filter(p -> p.getPoids() < 50).forEach(System.out::println);
}
}
- 第 14-16 行：显示年龄小于 28 岁的人；
- 第18-20行：显示体重小于50的人；
- 第22行：与第14–16行功能相同，但表达更简洁；
- 第24行：与第18–20行功能相同，但表达更简洁；
执行结果如下：
5.5. 示例-05 - 从 Stream<T1> 创建 Stream<T2>
 |
请看以下代码：
package dvp.java8.streams;
import java.util.List;
import dvp.data.Personne;
import dvp.data.Personnes;
public class Exemple05 {
public static void main(String[] args) {
// list of persons
List<Personne> personnes = Personnes.get();
// displays
System.out.println("Personne --> String ----------------------");
personnes.stream().map(p -> p.getNom()).forEach(System.out::println);
System.out.println("Personne --> Integer ----------------------");
personnes.stream().map(p -> p.getAge()).forEach(System.out::println);
}
}
- 第 13 行：[Stream.map] 方法的签名如下：
 |
[Stream.map] 方法的参数是第 4.3 节中介绍的函数接口 [Function] 的一个实例，该接口需要实现的唯一方法是：R apply(T t)。 我们可以看到，给定类型 T，[apply] 函数会返回类型 R。因此，[Stream.map] 方法将从类型 T 的流（此处“类型 T 的流”在技术上虽不完全准确，但我们将保留此表述，指由类型 T 的元素组成的流）生成类型 R 的流。
现在让我们分析示例中的代码：
- 第 14 行：从一个 person p 中，我们只保留名字。因此我们得到一个 String 类型的流；
- 第 14 行：从一个 person p 中，我们仅保留姓名。因此，我们得到一个 Integer 类型的流；
所得结果如下：
5.6. 示例-06 - Stream<T> 类的其他方法
 |
我们通过以下代码演示了 Stream 类中 39 种方法中的几种：
package dvp.java8.streams;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import dvp.data.Personne;
import dvp.data.Personnes;
import java.util.Comparator;
import java.util.List;
import java.util.stream.Collectors;
import java.util.stream.DoubleStream;
import java.util.stream.IntStream;
import java.util.stream.Stream;
public class Exemple06 {
// mapper jSON
static private ObjectMapper jsonMapper = new ObjectMapper();
public static void main(String[] args) throws JsonProcessingException {
// list of persons
List<Personne> personnes = Personnes.get();
// all people
affiche("all", personnes);
// the 1st person
affiche("findFirst", personnes.stream().findFirst().get());
// any person
affiche("findAny", personnes.stream().findAny().get());
// people without the 1st
affiche("skip 1", personnes.stream().skip(1L).collect(Collectors.toList()));
// the first 2 people
affiche("limit 2", personnes.stream().limit(2L).collect(Collectors.toList()));
// the number of people
affiche("count", personnes.stream().count());
// the oldest person
affiche("age max", personnes.stream().max(Comparator.comparingInt(Personne::getAge)).get());
// the lightest person
affiche("poids min", personnes.stream().min(Comparator.comparingDouble(Personne::getPoids)).get());
// last person in alphabetical order of name
affiche("nom max", personnes.stream().max((p1, p2) -> p1.getNom().compareToIgnoreCase(p2.getNom())).get());
// total age of all persons
affiche("âge total (reduce)", personnes.stream().map(p -> p.getAge()).reduce(0, (a1, a2) -> a1 + a2));
// people by ascending age
affiche("personnes par âge croissant",
personnes.stream().sorted(Comparator.comparingInt(Personne::getAge)).collect(Collectors.toList()));
// are there any people over 100?
affiche("des personnes de + de 100 ans (anyMatch)", personnes.stream().anyMatch(p -> p.getAge() > 100));
// are all people at most 100 years old?
affiche("des personnes de + de 100 ans (noneMatch)", personnes.stream().noneMatch(p -> p.getAge() > 100));
// are all people over 8 years old?
affiche("des personnes de + de 8 ans (allMatch)", personnes.stream().allMatch(p -> p.getAge() > 8));
// group people by gender
affiche("personnes regroupées par sexe", personnes.stream().collect(Collectors.groupingBy(p -> p.getSexe())));
// remove duplicate elements from a list
affiche("distinct", Stream.of(1, 2, 1).distinct().collect(Collectors.toList()));
// of a Stream<Stream<T>>, we make a Stream<T>
affiche("flatMap", Stream.of(1, 2, 3).flatMap(i -> Stream.of(i, i + 10)).collect(Collectors.toList()));
// of a Stream<Stream<Integer>>, we make a IntStream and calculate its sum
affiche("flatMapToInt", Stream.of(1, 2, 3).flatMapToInt(i -> IntStream.of(i, i + 10)).sum());
// of a Stream<Stream<Integer>>, we make a DoubleStream and then an array
affiche("flatMapToDouble", Stream.of(1, 2, 3).flatMapToDouble(i -> DoubleStream.of(i, i * 1.2)).toArray());
// max of a stream of integers
affiche("reduce Integer::max", Stream.of(1, 10, 8).reduce(Integer::max).get());
// min of a Double
affiche("reduce Integer::min", Stream.of(1.5, 10.4, 8.9).reduce(Double::min).get());
// average of a stream of integers
affiche("IntStream average", IntStream.of(1, 10, 8).average().getAsDouble());
// statistics for a stream of integers
affiche("IntStream summaryStatistics", IntStream.of(1, 10, 8).summaryStatistics());
}
public static <T> void affiche(String message, T value) throws JsonProcessingException {
System.out.println(String.format("%s ----", message));
System.out.println(jsonMapper.writeValueAsString(value));
}
}
- 第 72、75 行：显示该方法第二个参数的 JSON 字符串；
- 第 24 行：显示所有人的 JSON 字符串。结果如下：
5.6.1. [findFirst]
// la 1ère personne
affiche("findFirst", personnes.stream().findFirst().get());
[findFirst] 方法返回流中的第一个元素（如果存在）。其签名如下：
 |
返回值类型为 Optional<T>，这是 Java 8 中引入的一种类型：
 |
Optional<T> 类允许对空指针进行不同的处理。如果某个方法需要返回可能为空的类型 T，则可以选择返回 Optional<T> 类型。通过 [Optional<T>.isPresent()] 方法，可以判断该方法是否返回了值。以下代码 [Example06b] 演示了 Optional<T> 部分的工作原理：
package dvp.java8.streams;
import java.util.Optional;
import com.fasterxml.jackson.core.JsonProcessingException;
public class Exemple06b {
public static void main(String[] args) throws JsonProcessingException {
// optional without value
Optional<Integer> o1 = m1();
System.out.println(o1.isPresent());
affiche(o1);
// optional with value
Optional<Integer> o2 = m2();
System.out.println(o2.isPresent());
affiche(o2);
}
private static void affiche(Optional<Integer> o1) {
try {
// retrieve the value of the Optional
// throws 1 exception if no value
System.out.println(o1.get());
} catch (Throwable th) {
System.out.printf("%s : %s%n", th.getClass().getName(), th.getMessage());
}
}
public static Optional<Integer> m1() {
// no value
return Optional.empty();
}
public static Optional<Integer> m2() {
// a value
return Optional.of(10);
}
}
结果如下：
false
java.util.NoSuchElementException : No value present
true
10
让我们回到演示 [findFirst] 方法的代码：
// la 1ère personne
affiche("findFirst", personnes.stream().findFirst().get());
- 第 2 行：为了简化代码，我们对 [findFirst] 方法生成的 Optional<Person> 对象调用了 [get] 方法。符合代码规范的做法是在调用 [get] 方法之前，先调用 [Optional<Person>.isPresent()] 方法；
结果如下：
5.6.2. [findAny]
// n'importe quelle personne
affiche("findAny", personnes.stream().findAny().get());
[findAny] 方法的签名如下：
 |
[findAny] 方法可以返回流中的任意元素。在测试过程中，我们观察到顺序执行会返回流中的第一个元素，而并行执行确实可以返回任意元素。以下代码 [Example06c] 对此进行了演示：
package dvp.java8.streams;
import java.util.List;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import dvp.data.Personne;
import dvp.data.Personnes;
public class Exemple06c {
// mapper jSON
static private ObjectMapper jsonMapper = new ObjectMapper();
public static void main(String[] args) throws JsonProcessingException {
// list of persons
List<Personne> personnes = Personnes.get();
// everyone
affiche("all", personnes);
// any person
affiche("findAny parallèle", personnes.stream().parallel().findAny().get());
// any person
affiche("findAny séquentiel", personnes.stream().findAny().get());
}
public static <T> void affiche(String message, T value) throws JsonProcessingException {
System.out.println(String.format("%s ----", message));
System.out.println(jsonMapper.writeValueAsString(value));
}
}
- 第 22 行：findAny 并行执行；
- 第 24 行：findAny 按顺序执行；
所得结果如下：
- 第4行：并行执行返回了人员列表的第二个元素。也可能是其他元素；
- 第6行：顺序执行返回了人员列表中的第一个元素；
[findAny] 方法的使用似乎仅在流的并行处理中才有意义。
5.6.3. [skip]
// les personnes sans la 1ère
affiche("skip 1", personnes.stream().skip(1L).collect(Collectors.toList()));
[skip] 方法的签名如下：
 |
[skip] 方法用于跳过流中的前 n 个元素。如上文文档所述，并行调用此方法几乎无法带来性能提升，甚至可能导致性能下降。事实上，为了跳过前 n 个元素，线程被迫进行协调，这抵消了并行处理带来的性能优势。
[skip] 方法返回一个 Stream<Person>，该流通过 [collect] 方法转换为 List<Person>，其签名如下：
 |
[collect] 方法的参数是一个 [Collector] 类型的实例，该类型的签名较为复杂。通常，[Collector] 类型有预定义的实现，因此您通常无需自行实现。此处使用的实现是 [Collectors.toList()]。 [Collectors] 是一个包含大量静态方法的类，这些方法实现了 [Collector<T,A,R>] 类型。当您想要将一个流 (Stream) 转换为标准的 Java 集合时，这里是您首先应该查阅的地方：
 |
我们稍后将使用其中的一些方法。
执行后得到以下结果：
列表中的第一个元素（jean）已被省略。
5.6.4. [limit]
// les 2 premières personnes
affiche("limit 2", personnes.stream().limit(2L).collect(Collectors.toList()));
[limit] 方法的签名如下：
 |
[limit] 方法允许您仅保留流中的前 n 个元素。它不适用于并行处理。
执行结果如下：
5.6.5. [count]
// le nombre de personnes
affiche("count", personnes.stream().count());
[count] 方法的签名如下：
 |
[count] 方法返回流中元素的数量。如以下代码（Example06d1）所示，该方法的并行执行不会带来性能提升：
package dvp.java8.streams;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.stream.Stream;
public class Exemple06d1 {
public static void main(String[] args) {
final long limite = 10_000_000L;
// number of processors
System.out.printf("La JVM a détecté [%s] coeurs sur votre machine%n", Runtime.getRuntime().availableProcessors());
// list of numbers
long début = new Date().getTime();
List<Long> nombres = new ArrayList<>();
for (long i = 0; i < limite; i++) {
nombres.add(i);
}
System.out.printf("création de la liste des %s nombres en %s ms%n", limite, new Date().getTime() - début);
// counting numbers - sequential method
Stream<Long> sNombres = nombres.stream();
début = new Date().getTime();
long count = sNombres.count();
System.out.printf("comptage séquentiel : compteur=%s, durée (ms)=%s%n", count, new Date().getTime() - début);
}
}
- 第 11–22 行：创建一个包含 1000 万个数字的流；
- 第22–24行：对流进行计数；
执行结果如下：
如果我们将代码的第 22 行替换为以下内容（Example06d2）：
Stream<Long> sNombres = nombres.stream().parallel();
我们将得到以下结果：
5.6.6. [最大值, 最小值]
// la personne la + âgée
affiche("age max", personnes.stream().max(Comparator.comparingInt(Personne::getAge)).get());
[max] 方法的签名如下：
 |
[max] 方法使用作为参数传递的比较器，返回流中的最大值。 Comparator 是一个函数接口，其唯一需要实现的方法具有以下签名：int compare(T o1, T o2)。该方法必须在 o1 < o2 时返回 -1，在 o1.equals(o2) 时返回 0，在 o1 > o2 时返回 +1。Comparator 函数接口提供了许多默认的静态方法，用于在最常见的情况下实现 Comparator 接口。因此，在以下语句中：
affiche("age max", personnes.stream().max(Comparator.comparingInt(Personne::getAge)).get());
我们使用了静态方法 [Comparator.comparingInt]，其签名如下：
 |
ToIntFunction 类型是一个函数式接口：
 |
ToIntFunction 函数式接口的 [applyAsInt] 方法会将类型 T 转换为 int 类型。让我们回到我们的代码：
affiche("age max", personnes.stream().max(Comparator.comparingInt(Personne::getAge)).get());
此处 [Comparator.comparingInt] 方法的实际参数必须是 Person --> int 类型的 lambda 表达式。我们传入了 [Person.getAge] 方法的引用，该方法具有此签名。最终，我们将获得年龄最大的人。我们得到一个 Optional<Person> 类型，并通过 [Optional.get] 方法从中提取值。结果如下：
如下例所示，并行计算最大值并不会带来任何性能提升：（示例06e1）：
package dvp.java8.streams;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.Date;
import java.util.List;
import java.util.Random;
import java.util.stream.Stream;
public class Exemple06e1 {
public static void main(String[] args) {
// data
// final long limit = 100L;
// final boolean verbose = true;
final long limite = 10_000_000L;
final boolean verbose = false;
// number of processors
System.out.printf("La JVM a détecté [%s] coeurs sur votre machine%n", Runtime.getRuntime().availableProcessors());
// list of numbers
long début = new Date().getTime();
List<Long> nombres = new ArrayList<>();
for (long i = 0; i < limite; i++) {
nombres.add(new Random().nextLong());
}
System.out.printf("création de la liste des %s nombres en %s ms%n", limite, new Date().getTime() - début);
// max numbers - sequential method
Stream<Long> sNombres = nombres.stream();
Comparator<Long> compLong = (l1, l2) -> {
if (verbose) {
// thread
System.out.printf("[%s]", Thread.currentThread().getName());
}
// comparison
long v1 = l1.longValue();
long v2 = l2.longValue();
if (v1 < v2) {
return -1;
} else {
if (v1 == v2) {
return 0;
} else {
return +1;
}
}
};
début = new Date().getTime();
// long max = sNombres.max(Comparator.naturalOrder()).get();
long max = sNombres.max(compLong).get();
System.out.printf("%nmax séquentiel : max=%s, durée (ms)=%s%n", max, new Date().getTime() - début);
}
}
- 第 29 行：我们有一个类型为 Long 的随机整数流；
- 第 30–47 行：lambda 变量 compLong 实现了 Comparator<Long> 接口。该接口通常由第 49 行的 [Comparator.naturalOrder()] 方法实现。但在此处，我们希望显示执行线程（第 31–33 行）。因此我们自行实现了该接口；
- 第 50 行：查找最大值；
我们得到以下结果：
 |
现在，如果我们将第27行替换为以下内容（示例06e2）：
Stream<Long> sNombres = nombres.stream().parallel();
我们将得到以下结果：
 |
因此，并行执行的速度较慢。如果我们将数字数量增加到 1000 万，并设置 verbose=false，则得到以下结果：
对于顺序执行：
对于并行执行，其速度仍然较慢。
我们以类似的方式使用 [Stream.min] 方法：
// la personne la + légère
affiche("poids min", personnes.stream().min(Comparator.comparingDouble(Personne::getPoids)).get());
5.6.7. [reduce]
// l'âge total de toutes les personnes
affiche("âge total (reduce)", personnes.stream().map(p -> p.getAge()).reduce(0, (a1, a2) -> a1 + a2));
[reduce] 方法已在第 5.3 节中介绍。上文第 2 行计算了所有人的年龄之和。结果如下：
5.6.8. [排序]
// les personnes par âge croissant
affiche("personnes par âge croissant",
personnes.stream().sorted(Comparator.comparingInt(Personne::getAge)).collect(Collectors.toList()));
// les personnes par ordre alphabétique des noms
List<Personne> lPersonnes=personnes.stream().sorted((p1, p2) -> p1.getNom().compareTo(p2.getNom())).collect(Collectors.toList());
affiche("personnes par ordre alphabétique des noms", lPersonnes);
[sorted] 方法（第 3 行和第 5 行）的签名如下：
 |
[sorted] 方法的参数是第 5.6.6 节中为 min 和 max 方法描述的 [Comparator] 类型。它允许根据作为参数传递的比较器，按其定义的顺序对 Stream 进行排序。 我们已经看到，[Comparator] 接口提供了若干默认静态方法，用于实现常见的比较器，特别是针对数字和字符串的比较器。在此，我们使用 [Comparator.comparingInt] 方法，该方法接受一个 ToIntFunction 类型的参数，这是一个用于 [applyAsInt] 方法的功能接口，其签名如下： int applyAsInt(T t)。 在此，第 3 行传递给 [Comparator.comparingInt] 方法的实际参数是 [Person.age] 方法的引用，该方法返回该人的年龄。
[Comparator] 接口未提供用于比较字符串的静态方法。在第 5 行，我们自行构造了一个实现该接口唯一方法的 lambda 表达式：int compare(T t1, T t2)
(p1, p2) -> p1.getNom().compareTo(p2.getNom())
该 lambda 表达式用于比较人员的姓名。所得结果如下：
如下面的代码（Example06f1）所示，该排序似乎无法并行执行：
package dvp.java8.streams;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.Date;
import java.util.List;
import java.util.Random;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
public class Exemple06f1 {
// mapper jSON
static ObjectMapper jsonMapper = new ObjectMapper();
public static void main(String[] args) throws JsonProcessingException {
// data
final long limite = 100L;
final boolean verbose = true;
// final long limit = 10_000_000L;
// final boolean verbose = false;
// number of processors
System.out.printf("La JVM a détecté [%s] coeurs sur votre machine%n", Runtime.getRuntime().availableProcessors());
// list of numbers
long début = new Date().getTime();
List<Integer> nombres = new ArrayList<>();
for (long i = 0; i < limite; i++) {
nombres.add(new Random().nextInt(1000));
}
System.out.printf("création de la liste des %s nombres en %s ms%n", limite, new Date().getTime() - début);
// number sorting - sequential method
Stream<Integer> sNombres = nombres.stream();
début = new Date().getTime();
Comparator<Integer> compInt = (i1, i2) -> {
if (verbose) {
// thread
System.out.printf("[%s]", Thread.currentThread().getName());
}
// comparison
int v1 = i1.intValue();
int v2 = i2.intValue();
if (v1 < v2) {
return +1;
} else {
if (v1 == v2) {
return 0;
} else {
return -1;
}
}
};
if (verbose) {
affiche("nombres", sNombres.sorted(compInt).collect(Collectors.toList()));
}
System.out.printf("tri séquentiel : durée (ms)=%s%n", new Date().getTime() - début);
}
public static <T> void affiche(String message, T value) throws JsonProcessingException {
System.out.println(String.format("%s ----", message));
System.out.println(jsonMapper.writeValueAsString(value));
}
}
- 第30–36行：我们生成一个随机数序列；
- 第 32 行：我们将 compInt 闭包（第 38-55 行）传递给 [sorted] 方法。该闭包将数字按降序排序，并显示执行该闭包的线程。
所得结果如下：
 |
如果在之前的代码中，我们将第36行替换为以下内容（示例06f2）：
Stream<Integer> sNombres = nombres.stream().parallel();
我们将得到以下结果：
 |
令人惊讶的是，我们发现这串数字是在单线程下进行排序的。并没有并行处理。还是说我漏掉了什么？
5.6.9. [anyMatch, noneMatch, allMatch]
// y-a-t-il des personnes de + de 100 ans ?
affiche("des personnes de + de 100 ans (anyMatch)", personnes.stream().anyMatch(p -> p.getAge() > 100));
// est-ce que toutes les personnes ont au plus 100 ans ?
affiche("des personnes de + de 100 ans (noneMatch)", personnes.stream().noneMatch(p -> p.getAge() > 100));
// est-ce que toutes les personnes ont + de 8 ans
affiche("des personnes de + de 8 ans (allMatch)", personnes.stream().allMatch(p -> p.getAge() > 8));
第 2、4 和 6 行：方法 [anyMatch, noneMatch, allMatch] 接受 Predicate 类型作为参数，如第 4.2 节所述。因此，它们执行过滤操作。这三个方法都返回一个布尔值：
- 如果流中至少有一个元素满足过滤条件，则 anyMatch 返回 true；
- 如果流中没有元素满足过滤条件，noneMatch 返回 true；
- 如果流中的所有元素都满足过滤条件，则 allMatch 返回 true；
所得结果如下：
5.6.10. [collect(Collectors.groupingBy)]
// on regroupe les personnes par sexe
affiche("personnes regroupées par sexe", personnes.stream().collect(Collectors.groupingBy(p -> p.getSexe())));
[collect] 方法在第 5.6.3 节中已介绍过。其参数是 [Collector] 接口的一个实现。[Collectors] 类提供了一系列实现 [Collector] 接口的静态方法。 到目前为止，我们一直使用 [Collectors.toList()] 方法。在此，我们将使用静态方法 [Collectors.groupingBy]，该方法会从流中创建一个字典。其签名如下：
 |
[groupingBy] 方法会从一个 Stream<T> 创建一个 Map<K,List<T>>。键 K 由 [groupingBy] 方法的 Function<T,K> 类型参数提供，该函数的唯一方法具有以下签名：K apply(T t)。如果我们要创建一个以人的性别为索引的映射，就必须提供一个能根据人生成性别的函数。 在此，我们将 [Person.getGender] 方法的引用作为 [groupingBy] 方法的实际参数传入。所得结果如下：
第 2 行包含一个以 MAN 和 WOMAN 为键的字典的 JSON 字符串。
如下例（Example06g1）所示，并行计算并不会带来性能提升：
package dvp.java8.streams;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.function.Function;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
public class Exemple06g1 {
// mppeur jSON
static ObjectMapper jsonMapper = new ObjectMapper();
public static void main(String[] args) throws JsonProcessingException {
// data
final long limite = 100L;
final boolean verbose = true;
// final long limit = 10_000_000L;
// final boolean verbose = false;
// number of processors
System.out.printf("La JVM a détecté [%s] coeurs sur votre machine%n", Runtime.getRuntime().availableProcessors());
// list of numbers
long début = new Date().getTime();
List<Integer> nombres = new ArrayList<>();
for (long i = 0; i < limite; i++) {
nombres.add(new Random().nextInt(1000));
}
System.out.printf("création de la liste des %s nombres en %s ms%n", limite, new Date().getTime() - début);
// grouping numbers by hundred - sequential method
Stream<Integer> sNombres = nombres.stream();
Function<Integer, Integer> groupByCent = n -> {
if (verbose) {
System.out.printf("[%s]", Thread.currentThread().getName());
}
return n / 100;
};
début = new Date().getTime();
// Map<Integer, List<Integer>> lNombres = sNombres.collect(Collectors.groupingBy(number -> number / 100));
Map<Integer, List<Integer>> lNombres = sNombres.collect(Collectors.groupingBy(groupByCent));
System.out.printf("%nregroupement séquentiel : durée (ms)=%s%n", new Date().getTime() - début);
// results
if (verbose) {
affiche("nombres regroupés", lNombres);
}
}
public static <T> void affiche(String message, T value) throws JsonProcessingException {
System.out.println(String.format("%s ----", message));
System.out.println(jsonMapper.writeValueAsString(value));
}
}
- 第 23–38 行：构建一个数字流；
- 第 47 行：将数字按百位分组。第 39–44 行中的 lambda 函数用于显示执行线程；
执行结果如下：
 |
如果在代码中，我们将第38行替换为以下代码（Example06g2）：
Stream<Integer> sNombres = nombres.stream().parallel();
我们将得到以下结果：
 |
我们可以看到，分组操作的并行执行导致了性能下降。
5.6.11. [distinct]
// supression des éléments en double d'une liste
affiche("distinct", Stream.of(1, 2, 1).distinct().collect(Collectors.toList()));
[distinct] 方法的签名如下：
 |
它用于从流中移除重复项。[Stream.of] 方法（第 2 行）具有以下签名：
 |
它允许您根据显式提供的值创建一个流。执行结果如下：
5.6.12. [flatMap]
// d'un Stream<Stream<T>>, on fait un Stream<T>
affiche("flatMap", Stream.of(1, 2, 3).flatMap(i -> Stream.of(i, i + 10)).collect(Collectors.toList()));
[flatMap] 方法的签名如下：
  |
[flatMap] 方法接受一个函数作为参数，该函数：
- 该函数将流中的类型为 T 的元素作为参数；
- 返回一个 Stream<R>；
如果我们不用 [flatMap] 方法，而是使用第 5.5 节中描述的 [map] 方法，结果将是一个 Stream<Stream<R>> 类型，其中初始流中的每个 T 类型元素都会产生一个 Stream<R> 元素。 另一方面，[flatMap] 方法返回的是 Stream<R> 类型。它将多个 Stream<R> 流合并为一个单一的流。这正是执行前面的代码所显示的结果：
[flatMap] 还有一些专门的变体：
// d'un Stream<IntStream>, on fait un IntStream dont on calcule la somme
affiche("flatMapToInt", Stream.of(1, 2, 3).flatMapToInt(i -> IntStream.of(i, i + 10)).sum());
[flatMapToInt] 方法的签名如下：
 |
[flatMapToInt] 方法接受一个函数作为参数，该函数返回以下类型的 IntStream：
 |
IntStream 是一个 int 类型的流。与 Stream<Integer> 类型相比，该类型更优，因为其处理过程避免了 Integer 与 int 类型之间的装箱/拆箱操作。该接口继承了 Stream<T> 类型的许多方法，并新增了其他方法，包括上文提到的 [sum] 方法，该方法用于计算 IntStream 中元素的总和。
以下代码演示了类似的 [flatMapToDouble] 方法的使用：
// d'un Stream<DoubleStream>, on fait un DoubleStream puis un tableau
affiche("flatMapToDouble", Stream.of(1, 2, 3).flatMapToDouble(i -> DoubleStream.of(i, i * 1.2)).toArray());
[DoubleStream.toArray] 方法允许您将 DoubleStream 类型转换为 double[] 类型。
这两个示例的结果如下：
以下示例展示了从 Stream<Long> 类型切换到 LongStream 类型所获得的性能提升（示例06i1）：
package dvp.java8.streams;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
public class Exemple06i1 {
public static void main(String[] args) {
final long limite = 10_000_000L;
// number of processors
System.out.printf("La JVM a détecté [%s] coeurs sur votre machine%n", Runtime.getRuntime().availableProcessors());
// list of numbers
long début = new Date().getTime();
List<Long> nombres = new ArrayList<>();
for (long i = 0; i < limite; i++) {
nombres.add(i);
}
System.out.printf("création de la liste des %s nombres en %s ms%n", limite, new Date().getTime() - début);
// sum of numbers - sequential method
début = new Date().getTime();
long somme = nombres.stream().reduce(0L, (s, i) -> s + i);
System.out.printf("somme séquentielle du Stream<Integer> : somme=%s, durée (ms)=%s%n", somme, new Date().getTime() - début);
}
}
- 第 22 行：计算 Long 类型数流的总和；
得到以下结果：
现在，让我们将第 22 行替换为以下内容（Example06i2）：
long somme = nombres.stream().mapToLong(n -> n.longValue()).sum();
Stream<Integer>.mapToLong 方法允许我们获取一个包含原始 long 类型的 LongStream，然后使用 sum 函数对其求和。最终得到以下结果：
性能提升显而易见。
5.6.13. 原始数字流方法
// max d'un flux de int
affiche("IntStream max", IntStream.of(1, 10, 8).max());
// min d'un flux de double
affiche("DoubleStream min", DoubleStream.of(1.5, 10.4, 8.9).min());
// moyenne d'un flux de int
affiche("IntStream average", IntStream.of(1, 10, 8).average().getAsDouble());
// statistiques d'un flux de int
affiche("IntStream summaryStatistics", IntStream.of(1, 10, 8).summaryStatistics());
基本类型（int、long、double）的流提供了针对这些类型的专用方法。执行上述代码的结果如下：
- 代码第 2 行的结果是一个 OptionalInt 类型，类似于 Optional<Integer> 类型。可以通过 [getAsInt()] 方法获取此对象中存储的值。可以通过 [isPresent()] 方法检查值是否存在。 结果中的第 2 行并不意味着 [OptionalInt] 类拥有名为 [asInt] 和 [present] 的字段。默认情况下，JSON 库会使用待序列化为 JSON 的对象的所有公共 getX 和 isY 方法。而在此处，确实存在 [getAsInt] 方法和 [isPresent] 方法，尽管 [asInt, present] 字段本身并不存在；
- 代码第 4 行的结果是一个 OptionalDouble 类型，类似于 Optional<Double> 类型；
- 代码第 6 行的结果是一个 OptionalDouble 类型，其值可通过 [getAsDouble()] 方法获取。[average] 方法计算数字流的平均值；
- 代码第 8 行的结果是一个 IntSummaryStatistics 类型，其定义如下：
 |
我们可以看到，生成的 IntSummaryStatistics 对象提供了关于数字流的各种统计信息，例如值个数、总和、最大值、最小值和平均值。