翻译 | Java8 Stream API 教程

本文翻译自The Java 8 Stream API Tutorial

简介

  本教程志在细致入微、深入底层，你将体验从Stream的创建开始（creation）到并行执行（parallel execution）的完整过程，以此体会Stream API的实际用处。

  为了理解下面的文章，读者需要掌握Java 7基础知识（Lambda表达式、Optional、方法引用）以及熟悉Stream API，如果你并不熟悉它们甚至一无所知，建议你先阅读我们之前的文章-Java8 新特性 以及 Java 8 Streams 介绍。

创建Stream

  创建一个Stream实例有多种方式，每种创建方式对应Stream的一个来源。但单个Stream实例每次创建之后，其来源将无法修改，这意味着Stream实例具备源头不可变性，不过我们却可以从单个源创建多个Stream实例。

Empty Stream - 空Stream

  方法empty()被用于创建一个Empty Stream：

Stream<String> streamEmpty  = Stream.empty;

  上述代码段创建的Empty Stream通常被用于避免null对象或零元素对象的streams（streams with no element）返回结果为null：

public Stream<String> streamOf(List<String> list){
  return lsit == null || list.isEmpty() ? Stream.empty() : list.streams();
}

Stream of Collection - 集合Steram

  我们可以创建任意Collection接口衍生类（Collection->List、Set、Queue）的Streams：

Collections<String> collection = Arrays.asList("a", "b", "c");
Stream<Stirng> streamOfCollection = collection.stream();

Stream of Array - 数组Stream

  接下来的这段代码展示的是数组Stream：

Stream<String> streamOfArray = Stream.of("a", "b", "c");

  当然我们可以先创建熟悉的数组类型，再以它为源创建Stream，而且我们可以选择Stream中包含的元素数量：

String[] arr = new String[]{"a", "b", "c"};
Stream<String> streamOfArrayFull = Arrays.stream(arr);
Stream<String> streamOfArrayPart = Arrays.stream(arr, 1, 3);

Stream.builder() - 构建器

  当builder被用于指定参数类型时，应被额外标识在声明右侧，否则方法build()将创建一个Stream(Object)实例：

Stream<String> streamBuilder = Stream.<String>builder().add("a").add("b").add("c").build();

Stream.generator() - 生成器

  方法generator()接受一个供应器Supplier用于元素生成。由于生产流(resulting stream)被定义之后属于无限流（即无止境地不断生产），开发者必须指定stream拥有流的目标大小，否则方法generator()将持续生产直到jvm内存到达顶值（memory limit）：

Stream<String> streamOfGenerated = Stream.generate( () -> "element").limit(10);

  上述代码将创建十个内容为“element”的生成流。

Stream.iterate() - 迭代器

  另一种创建无限流的方法是通过调用方法iterate()，同样的它也需要使用方法limit()对目标流的元素大小进行限制：

Stream<Integer> streamItreated = Stream.iterate(40, n -> n + 2).limit(20);

  迭代流即采用迭代的方法作为元素生产方式，类似于高中数学中的f(x)，f(f(x))，etc。上述例子中，生成流的第一个元素是迭代器iterate()中的第一个元素40，从第二个元素开始的每个新元素都与上个元素有关，在此例中，生成流中的元素为：40、42、44、…78、80。

Stream of Primitives - 基元流

  Java8提供了创建三大基础数据类型（int、long、double）stream的方式。由于Stream是一个类接口，我们无法采用泛型传参的方式声明基础数据类型的stream，因此三个特殊的接口就被创造出来了：IntStream、LongStream、DoubleStream。
使用它们能够避免不必要的自动装箱¹以提高生产效率。

IntStream intStream = IntStream.range(1, 3);
LongStream longStream = LongStream.rangeClosed(1, 3);

  方法range(int startInclusive, int endInclusive)创建了一个有序流（从startInclusive到endInclusive）。它使后面的值每个增加1，但却不包括最后一个参数，即此方法的结果是具备上限的。方法rangeClosed(int startInclusive, int endInclusive)与range()大致相同，但它却包含了最后一个值。
这两个方法用于生成三大基本数据类型的stream。

  此外，Java8之后，类Random也提供了拓展方法用于生成基础数据类型的stream。例如，下述代码创建了一个含有三个随机值的DoubleStream：

Random random = new Random();
DoubleStream doubleStream = random.doubles(3);

Stream of String - 字符串流

  String类型也可以作为生成stream的源，这得益于方法chars()的帮助，此外由于JDK中没有CharStream接口，IntStream也被用来表示字符流（stream of chars）

IntStream streamOfChars = "abc".chars();

  下例中通过特征的正则表达式将一个字符串割裂成（break into）其子串。

Stream<String> streamOfString = 
  Pattern.compile(", ").spitAsStream("a", "b", "c");

Stream of File - 文件流

  Java NIO²类文件允许通过方法lines()生成文本文件的Stream。文本的每一行都会变成stream的一个元素：

Path path = Paths.get("C:\\file.txt");
Stream<String> streamOfString = Files.lines(path);
Stream<String> streamWithCharset = Files.lines(path, Charset.forName("utf-8"));

ps：在方法lines()中也可以通过Charset设置文件编码。

Referencing a Stream - 引用stream

  只要调用生成操作（中间操作）就会实例化一个stream并生成一个可获取的引用，但执行终端操作会使得stream无法访问。为了证明这一点，我们不妨先忘记它，毕竟实践是检验真理的唯一标准。
以下代码如果不考虑冗长的话将是有效的：

Stream<String> stream = Stream.of("a", "b", "c").filter(element -> element.contains("b"));
Optional<String> anyElement = stream.findAny();

  但是倘若我们在执行终端操作后重新使用相同的引用，则会不可避免的触发IllegalStateException。

Optional<String> firstElement = stream.findFirst();

  IllegalStateException是一个运行时异常（RuntimeException），即编译器将不会提示此错误。因此必须记得，JAVA8 不允许重复使用stream
这一设计是合乎逻辑的，因为stream从设计上旨在提供一个将有限操作（指函数体中元素的相关操作）的序列，而不是存储元素。
因此想让以前的代码正常工作我们得先改一改：

List<String> elements =
  Stream.of("a", "b", "c").filter(element -> element.contains("b"))
    .collect(Collectors.toList());
Optional<String> anyElement = elements.stream().findAny();
Optional<String> firstElement = elements.stream().findFirst();

Stream Pipeline - 流的管道

  想要执行源数据集的操作集并聚合它们，你需要以下三个部分——源（Source)、中间操作（Intermediate operations）和终结操作（terminal operation）。
中间操作返回的是一个新的可操作stream。举个例子，为了在一个包含少量元素Stream的基础之上新建Stream，我们可以调用方法skip()：

Stream<String> oneModifiedStream = Stream.of("abcd", "bbcd", "cbcd").skip(1);

  如果需要多次修改，则可以采用多次中间操作。假如我们还需要将Stream中每个字符串替换为其子串subString(0, 3)，则可以使用skip()和map()相连的方式完成：

Stream<String> twiceModifiedStream = stream.skip(1).map(element -> element.subString(0, 3));

  正如你所见，上例中map()使用Lambda表达式作为其参数对stream中的各元素进行处理。
stream本身是毫无价值的，编程人员最感兴趣的其实是终结操作（terminal operation），它可以是一个元素也可以是一个行为。只有在终结操作里才能对每个stream进行使用。正确的且最方便的stream操作方式就是Stream Pipeline，即stream源->中间操作->终结操作。如例：

List<String> list = Arrays.asList("abc1", "abc2", "abc3");
long size = list.stream().skip(1)
  .map(element -> element.substring(0, 3)).sorted().count();

Lazy Invocation - 懒式调用

  中间操作是懒式调用的，这意味着只有在终结操作需要它们的时候中间操作才会被唤醒。
为了证明这个事实，假象我们有个方法wasCalled()，每当它被唤醒时使内部变量counter自增。

private long counter;
private void wasCalled() {
  counter++;
}

  接下来让我们在filter()操作中唤起wasCalled():

List<String> list = Arrays.asList(“abc1”, “abc2”, “abc3”);
counter = 0;
Stream<String> stream = list.stream().filter(element -> {
    wasCalled();
    return element.contains("2");
});

  由于有三个变量，想象中filter()中的代码块将被执行三次，wasCalled()执行三次之后counter的值应为3，但是执行之后counter并未发生改变，仍然为0，也就是说filter()一次也没有被唤醒，这个原因就是缺失了终结操作（terminal operation）。
那接下来我们不妨再上述代码的基础之上添加一次map()操作和一个终结操作——findFirst()，并采用打日志的方式帮助我们了解方法调用时机及顺序。

Optional<String> stream = list.stream().filter( element -> {
    log.info("filter() was called!");
    return element.contains("2");
  }).map(element -> {
    log.info("map() was called!");
    return element.toUpperCase();
  }).findFirst();

  日志结果显示filter()被唤醒了两次，而map()仅仅被调用一次，这是由于管道流是垂直执行的。在此例中第一个元素不满足filter()的要求，因此filter()被调用第二次以查找合适的结果，通过之后即进行map()操作，此时就没有第三次机会执行filter()操作了。findFirst()就能找出源数据集中第一个含有“2”的字符串的全大写字符串了。因此，懒调用使得不必相继调用两个中间操作（filter()和map()）才能完成任务了。

Order of Execution - 执行顺序

  从性能的角度考虑，正确的执行顺序是采用上文提到的流式管道（Stream Pipeline）：

long size = list.stream().map(element -> {
    wasCalled();
    return element.substring(0, 3);
}).skip(2).count();

  执行这段代码将使counter自增长3次，这意味着stream的方法map()将被调用3次，但最终size的值为1。这意味着结果流（resulting stream）中仅仅只有一个元素，毫无疑问在三次消息处理中程序跳过了两次处理。
如果我们改变skip()和map()的执行顺序，counter将只自增长一次。也即是map()只被调用一次：

long size = list.stream().skip(2).map(element -> {
    wasCalled();
    return element.substring(0, 3);
}).count();

  以上示例告诉我们一个规则：用于减少流中元素数量的中间操作，应当放置在处理操作之前。因此，保证在你的Stream Pipeline规则中按照这样的顺序编码：skip() –> filter() –> distinct()

Stream Reduction - 流的聚合

  API提供了大量的终端操作用以聚合一个stream为一种数据类型或变量。比如：count()、max()、min()、sum()，但是这些方法都是预定义的。但如果用户需要自定义一个stream的聚合操作呢？官方提供了两个方法用以实现此类需求：reduce() 和 collect()。

reduce()方法

  此方法提供了三种变种，不同之处是它们的签名以及返回类型。reduce()方法具有下列参数：
identify（标识器） - 累积器的初始值或当stream为空时的默认值。
accumulator（累积器） - 提供设定聚合元素之逻辑的功能，每次规约(reducing)累积器都会创建一个新的值，新值的大小等于stream的大小，并且只有上一个值是可用的。这非常有助于提升性能。
combiner（组合器） - 提供聚合accumulator（累积器）中元素的功能，combiner是唯一一个能从不同线程以并行模式聚合累积器中结果的方法。
好，让我们来实战一下吧：

OptionalInt reduced =
    IntStream.range(1, 4).reduce((a, b) -> a + b);

reduced = 6 = 1 + 2 + 3。

int reducedTwoParams =
    IntStream.range(1, 4).reduce(10, (a, b) -> a + b);

reducedTwoParams = 16 = 10 + 1 + 2 + 3。

int reducedParams = Stream.of(1, 2, 3)
  .reduce(10, (a, b) -> a + b, (a, b) -> {
     log.info("combiner was called");
     return a + b;
  });

  这一结果与上文中的16一样，并且不会打出日志，因为combiner没有被唤起。为了唤醒combiner，stream应当是并行的：

int reducedParallel = Arrays.asList(1, 2, 3).parallelStream()
    .reduce(10, (a, b) -> a + b, (a, b) -> {
       log.info("combiner was called");
       return a + b;
    });

  此时，结果变为36，并且combiner被唤起了两次。规约（reduce）运转的算法为：每当stream中的元素通过identify（标识器）时accumulator（累积器）均被调用，最终累积器调用了3次。上述行为是并行完成的，因此造成了（10+1=11; 10+2=12; 10+3=13;）。最终combiner（组合器）混合了三次的结果，通过两次迭代完成运算（12+13=25; 25+11=36;）。

collect()方法

  stream的规约也可以被其他的终结方法执行——collect()。它接收了一个名为collector的参数，此参数注明规约的流程。官方已经创建了预定义的收集器，我们可以在这些收集器的帮助下访问它们。
下面我们将看到使用List作为所有stream的来源：

List<Product> productList = Arrays.asList(new Product(23, "potatoes"),
  new Product(14, "orange"), new Product(13, "lemon"),
  new Product(23, "bread"), new Product(13, "sugar"));

转换一个stream为Collection集合（Collection、List、Set、Queue、etc）。

List<String> collectorCollection = 
  productList.stream().map(Product::getName).collect(Collectors.toList());

规约为String类型：

String listToString = productList.stream().map(Product::getName)
  .collect(Collectors.joining(", ", "[", "]"));

  join()方法拥有三个参数（delimiter, prefix, suffix），使用join()最便捷之处在于程序员不需要考虑stream的起始与结束甚至界定符，Collector会考虑到这些的。
计算stream中所有数字元素的平均值

double averagePrice = productList.stream()
  .collect(Collectors.averagingInt(Product::getPrice));

计算stream中所有数字元素的和

int summingPrice = productList.stream()
    .collect(Collectors.summingInt(Product::getPrice));

  方法averagingXX()、summingXX()和summarizingXX()适用于基础数据类型（int,long,double），也适用于它们的封装类（ Integer,Long,Double）。一个很有效的功能技术提供映射，因此开发者也不是一定需要在collect()方法之后使用map()操作才能完成映射的。
收集stream元素集的统计信息：

IntSummaryStatistics statistics = productList.stream()
    .collect(Collectors.summarizingInt(Product::getPrice));

  通过使用IntSummaryStatistics的生成实例，开发者能够通过请求toString()方法创建一个统计报告，结果将是一系列显而易见的结果：IntSummaryStatistics{count=5, sum=86, min=13, average=17,200000, max=23}。通过调用上述方法getCount()、getSum()、getMin()、getAverage()、getMax(),我们也很容易从对象中提取出count、sum、min、average的值，这是因为所有的值均可以从单个管道中获取。

采用指定方法组合stream中的元素:

Map<Integer, List<Product>> collectorMapOfLists = productList.stream()
  .collect(Collectors.groupingBy(Product::getPrice));

  此例中stream将根据group规则将所有元素规约成一个map。
根据一些描述对stream进行分组:

Set<Product> unmodifiableSet = productList.stream()
  .collect(Collectors.collectingAndThen(Collectors.toSet(),
  Collections::unmodifiableSet));

  这种相对特殊的情况里，collection将stream转化为一个Set，之后在此基础上创建了一个不可变的Set。

Custome collector（自定义收集器）:
  假若我们因为一些特定的原因需要创建自定义的收集器，那更简介轻快的方法是采用Collection的of()方法：

Collector<Product, ?, LinkedList<Product>> toLinkedList =
  Collector.of(LinkedList::new, LinkedList::add, 
    (first, second) -> { 
       first.addAll(second); 
       return first; 
    });
 
LinkedList<Product> linkedListOfPersons =
  productList.stream().collect(toLinkedList);

  在上例中，Collection的实例被规约成了一个LinkedList。

Parallel Streams - 并行流

  在Java8之前，并行化十分复杂。ExecutorService和FornJoin的出现大大降低了并行开发的复杂度，但它们都无不避免的关注在如何创建一个特征鲜明的executor，以及如何去运行它等等。Java8提倡了一种新的方式用于在函数类型中实现并行化。
  API提供并行流用以并行化执行操作。当stream的源是一个数组或者Collection时，在parallelStream()方法的帮助下可以实现并行化：

Stream<Product> streamOfCollection = productList.parallelStream();
boolean isParallel = streamOfCollection.isParallel();
boolean bigPrice = streamOfCollection
  .map(product -> product.getPrice() * 12)
  .anyMatch(price -> price > 200);

但如果stream的源不是数组或者集合类型时，parallel()方法就应该被使用了：

IntStream intStreamParallel = IntStream.range(1, 150).parallel();
boolean isParallel = intStreamParallel.isParallel();

上例中，Stream API自动使用了ForkJoin框架去完成并行操作。默认情况下，公共线程池将被使用，不会（至少暂时不会）给它单独分配线程。当stream处于并行状态时，应当注意可能产生阻塞的操作，当对时间效率有所追求且操作可并行时应当转换为并行stream（理由是假如某个任务大小远远多于其他任务，那它将更加耗时）。当然啦，并行模式也可以转换回串行模式，只要使用sequential()方法就能做到这点：

IntStream intStreamSequential = intStreamParallel.sequential();
boolean isParallel = intStreamSequential.isParallel();

Conclusions - 结论

  Stream API在对链式数据进行操作时体现了其强大性，但也易于理解。它通过引用的方法规约大容量的数据，构建了更健壮的程序，最主要的是提升了项目开发的生产力。
  在本文中stream均是未被关闭的（我们没有调用close()方法或者其他的终结操作），但在实际项目中，不要这样无节制的放纵stream的存在，这将逐步耗尽你的内存，造成内存泄漏程序崩溃的风险。
最后，本文所对应的示例代码你可以在github-core-java-8上获取到。祝福你身体健康，编码顺利！

附录

1. 自动装箱： 编译器自动为语句进行语法解析，如类型补充等。详见Java 自动装箱与拆箱(Autoboxing and unboxing)
2. JAVA NIO： New I/O的简称，与旧式的基于流的I/O方法相对，从名字看，它表示新的一套Java I/O标准。详见NIO与AIO