点击图片访问我的个人博客
1. 简介
本教程志在细致入微、深入底层，你将体验从Stream的创建开始（creation）到并行执行（parallel execution）的完整过程，以此体会Stream API的实际用处。
为了理解下面的文章，读者需要掌握Java 7基础知识（Lambda表达式、Optional、方法引用）以及熟悉Stream API，如果你并不熟悉它们甚至一无所知，建议你先阅读我们之前的文章-Java8 新特性 以及 Java 8 Streams 介绍。
2. 创建Stream
创建一个Stream实例有多种方式，每种创建方式对应Stream的一个来源。但单个Stream实例每次创建之后，其来源将无法修改，这意味着Stream实例具备源头不可变性，不过我们却可以从单个源创建多个Stream实例。
2.1 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();
}
2.2 Stream of Collection - 集合Steram
我们可以创建任意Collection接口衍生类（Collection->List、Set、Queue）的Streams：
Collections<String> collection = Arrays.asList("a", "b", "c");
Stream<Stirng> streamOfCollection = collection.stream();
2.3 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);
2.4 Stream.builder() - 构建器
当builder被用于指定参数类型时，应被额外标识在声明右侧，否则方法build()将创建一个Stream(Object)实例：
Stream<String> streamBuilder = Stream.<String>builder().add("a").add("b").add("c").build();
2.5 Stream.generator() - 生成器
方法generator()接受一个供应器Supplier<T>用于元素生成。由于生产流(resulting stream)被定义之后属于无限流（即无止境地不断生产），开发者必须指定stream拥有流的目标大小，否则方法generator()将持续生产直到jvm内存到达顶值（memory limit）：
Stream<String> streamOfGenerated = Stream.generate( () -> "element").limit(10);
上述代码将创建十个内容为“element”的生成流。
2.6 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。
2.7 Stream of Primitives - 基元流
Java8提供了创建三大基础数据类型（int、long、double）stream的方式。由于Stream<T>是一个类接口，我们无法采用泛型传参的方式声明基础数据类型的stream，因此三个特殊的接口就被创造出来了：IntStream、LongStream、DoubleStream。
使用它们能够避免不必要的自动装箱1以提高生产效率。
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);
2.8 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");
2.9 Stream of File - 文件流
Java NIO2类文件允许通过方法lines()生成文本文件的Stream<String>。文本的每一行都会变成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设置文件编码。
3. 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();
4. Stream Pipeline - 流的管道
想要执行源数据集的操作集并聚合它们，你需要以下三个部分——源（Source)、中间操作（Intermediate operations）和终结操作（terminal operation）。
中间操作返回的是一个新的可操作stream。举个例子，为了在一个包含少量元素Stream的基础之上新建Stream，我们可以调用方法skip()：
Stream<String> oneModifiedStream = Stream.of("abcd", "bbcd", "cbcd").skip(1);
如果需要多次修改，则可以采用多次中间操作。假如我们还需要将Stream<String>中每个字符串替换为其子串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();
5. 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()）才能完成任务了。
6. 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()
7. Stream Reduction - 流的聚合
API提供了大量的终端操作用以聚合一个stream为一种数据类型或变量。比如：count()、max()、min()、sum()，但是这些方法都是预定义的。但如果用户需要自定义一个stream的聚合操作呢？官方提供了两个方法用以实现此类需求：reduce() 和 collect()。
7.1 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;）。
7.2 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<Person>。
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上获取到。祝福你身体健康，编码顺利！
附录
- 自动装箱： 编译器自动为语句进行语法解析，如类型补充等。详见Java 自动装箱与拆箱(Autoboxing and unboxing)
- JAVA NIO： New I/O的简称，与旧式的基于流的I/O方法相对，从名字看，它表示新的一套Java I/O标准。详见NIO与AIO
网友评论