这个包主要提供元素的streams函数操作，比如对collections的map,reduce.

例如：

int sum = widgets.stream()
 .filter(b -> b.getColor() == RED)
 .mapToInt(b -> b.getWeight())
 .sum(); 

本例中的widgets是Stream的源，类型为Collection

** Streams和collections的不同之处**

1. Stream没有存储。即不是一个存储结构，而是通过管道操作从Array,IO channel等转换而来。

2. Stream本质上是泛函。Streams上的操作不会修改源数据，比如filter操作只是将符合要求的数据放在一个新的Stream,而不是真的删除源数据。

3. Laziness-seeking，懒查询。

4. 大小不受限制，collections 有大小限制，streams没有。比如，limit或者findfirst等Short-circuiting操作会在有限的时间内完成无限的大小的streams操作。

5. 不可逆的，streams中的元素只能被操作一次。就像流水一样，一去不复返。如果你希望在源数据中获取一个新的Stream,必须在源数据上重新开始。

** Stream创建**

怎样创建一个Stream呢，有下面几种方法：

1. Collection的Stream()和paralleStream()方法。

2. Arrays.stream(Object[])

3. Stream 类的静态方法，比如 Stream.of(Object[]),IntStream.range(int, int) or Stream.iterate(Object, UnaryOperator);

4. File的lines操作， BufferedReader.lines();

5. Files中的一些方法获取file path的stream.例如

static Stream<Path> list(Path dir)
static Stream<Path> walk(Path start, FileVisitOption... options) 

6.Random.ints()返回IntStream

7.JDK中也有很多方法产生Stream,比如

BitSet.stream(),

Pattern.splitAsStream(java.lang.CharSequence),

JarFile.stream()

java.util.stream包中的接口和类 

1.接口

BaseStream<T,S extends BaseStream<T,S>> 

Streams的Base interface,支持各种并行和线性聚合操作。

Collector<T,A,R> ，可以将元素聚集在一个容器中，容器类型可变，比如List,Map. 是一个Stream处理完毕之后的结束操作。

DoubleStream ，处理double类型的stream

DoubleStream.Builder , 可变的builder。

IntStream ，处理int类型的stream.

IntStream.Builder , 可变的builder

LongStream ,处理Long类型的Stream.

LongStream.Builder ,可变的builder

Stream<T> - –数据流，支持并行和线性聚合操作。

Stream.Builder<T> —-Stream的可变builder.

2.类

执行Collector reduce操作的类，比如将元素聚合为collection,list等。

StreamSupport，创建和处理Stream的低级工具。

3.Enum

Collector.Characteristics—Collector的属性，可以用于优化reduce 操作。

Collector.Characteristics CONCURRENT：表示collector的result container可以被多个线程同时处理。

Collector.Characteristics UNORDERED：collection操作不保证操作元素的原有顺序

Collector.Characteristics IDENTITY_FINISH：类型转换时不会检查

*** Stream的操作和管道**

Stream操作是管道操作，操作分为中间操作和结束操作两种。整个管道操作的流程为数据源->中间操作->结束操作。

中间操作 比如：Stream.filter、Stream.map,可以有0个或者多个中间操作。

结束操作 ，比如Sream.foreach、Stream.reduce，Stream.collect,结束操作是必须的。

中间操作返回的仍然是stream,这些操作经常是“懒惰的”。比如filter操作，并非真的对源数据进行过滤，而是重新创建一个Stream。而且当多个中间操作时，不是真的每次都去遍历源数据，而是在结束操作时，才会去遍历源数据，从而执行这些中间操作。在遍历中，操作的懒惰性表现在，并不是每次都是遍历所有的数据，比如“找出长度大于1000的第一个String”,找到之后就会返回。

结束操作的目的是将Stream转换为想要的结果或者结构。当结束操作之后， 当前管道结束，不可以再重复使用。如果你仍然想操作源数据，那你必须创建一个新的Stream.

中间操作又分为 无状态 和 有状态 两种。比如map和filter是无状态操作，也就是各元素间的操作是无关的，可单独进行。而distinct和sorted 是有状态操作，在操作时需要用到前元素，也就是元素之间的处理是相关的。

有状态的操作必须全部执行完毕的时候才能装入结果中。

1.并行操作

JDK的默认操作都是线性的，或者说单线程的。并行操作可以通过Collection. parallel Stream()和BaseStream.parallel()实现。

例如：

int sumOfWeights = widgets.parallelStream()
 .filter(b -> b.getColor() == RED)
 .mapToInt(b -> b.getWeight())
 .sum(); 

检查一个Stream的执行方式可以调用 isParallel() 方法。

但并不是所有的数据都是可以使用并行操作的 。要执行的操作必须是可以并行执行的，也就是多个线程操作之后，和顺序执行的结果是一致的，即 线程安全 。

int[] wordLength = new int[12];
Stream.of("It", "is", "your", "responsibility").parallel().forEach(s -> { if (s.length() < 12) wordLength[s.length()]++;
}); 

比如上面的wordlength操作就是非线程安全的。并行操作的线程安全问题可以参考博客：

2.非干扰性

通常在结束操作之前，源数据都是可以修改的。除了一些Concurrent数据，还有iterator,spliterator操作。

下面的例子就是正确的：

List<String> l = new ArrayList(Arrays.asList("one", "two"));
 Stream<String> sl = l.stream();
 l.add("three"); String s = sl.collect(joining(" ")); 

3.无状态的行为

在streams操作中使用无状态的行为，即操作的结果不受其他状态的影响，就比如上面提高的parallel操作要考虑多线程的行为，是否有线程安全问题。

4.collect,reduce操作通常是线程安全的，可参考博客

5.顺序

对于有顺序的源数据，比如list,set,Stream操作之后仍然会按照顺序排（ 线性操作 ）。

如果是非固定顺序的数据源，则不保证其最后处理的顺序。

对于线性Streams,源的固定顺序对性能没有特别影响。

对于并行Streams,不强调顺序的话显然性能会更好。比如filter,distinct,groupingBy操作，非顺序操作的效率就更好，可参考博客

另外，如果数据本身是固定顺序的，但是开发者本身不在意是否按顺序处理，那么可以通过unordered()处理之后再操作，会提高某些有状态的并行操作和结束操作的效率。

下面穿插一些线程安全的操作例子：

List< Integer >  datalist =new ArrayList<>(1000);List<Integer> datalist_collect=new ArrayList<>();//无状态的并行操作，即元素的操作和其他元素状态无关，线程安全
 long sum = IntStream.range(0,1000).parallel().filter(s->s%2==0).count();  System .out.println("并行:"+sum);//datalist为非线程安全的，因此属于有状态并行操作，非线程安全 IntStream.range(0,1000).parallel().forEach(data->datalist.add(data));
 System.out.println("并行foreach:"+datalist.size()); 
//collect操作线程安全datalist_collect=IntStream.range(0,1000).parallel().collect(ArrayList::new, (c,e)->c.add(e), (c1,c2)->c1.addAll(c2)); System.out.println("并行collect:"+datalist_collect.size());
 long sum2 = IntStream.range(0,1000).filter(s->s%2==0).count(); System.out.println("串行:"+sum2);
 List<Integer> intlist=IntStream.range(0,10).collect(ArrayList::new, (c,e)->c.add(e), (c1,c2)->c1.addAll(c2)); System.out.println("并行数据开始："); //并行操作时无顺序 intlist.parallelStream().map(e->e+1).forEach(System.out::println);
 System.out.println("串行数据开始：");//串行操作保证顺序
 intlist.stream().forEach(System.out::println); System.out.println("串行数据结束：");
运行结果如下-----：
并行:500并行foreach:980并行collect:1000串行:500并行数据开始：78106541932串行数据开始：0123456789串行数据结束 

6.Reduction 操作

Reduction操作将元素通过合并操作还原为一个result,比如总数，最大值，元素存入list等。

Reduction操作的函数有很多，比如 reduce(),collect(),sum(),max(),count() 等。

Reduction操作不仅看起来简洁，而且可以使用并行来实现，效率高，不存在效率安全问题，比如

int sum = 0;
 for (int x : numbers) { sum += x;
 } 

可以写为：

int sum = numbers.parallelStream().reduce(0, Integer::sum); 

这里介绍一下reduce()的定义， 该方法的 定义1 为：

T reduce(T identity,
 BinaryOperator<T> accumulator) 

该方法使用提供的identity和accumulator来获取result.等同于下面的写法，其实identity就是计算的初始值：

T result = identity; for (T  element  : this stream)
 result = accumulator.apply(result, element) return result; 

但于 for循环 不同的是，reduce操作不局限于线性运行。但对于同一个对象T,其accumulator的运算结果必须相同。

Reduce()方法的 定义2 是：

Optional<T> reduce(BinaryOperator<T> accumulator) 

这个操作等同于：

boolean foundAny = false;
 T result = null; for (T element : this stream) { if (!foundAny) {
 foundAny = true;
 result = element;
 } else
 result = accumulator.apply(result, element);
 } return foundAny ? Optional.of(result) : Optional.empty(); 

Accumulator的操作和上个方法一样，详见官网：

这里主要介绍 Optional返回值 的作用，可以参考文章

官方文档

Reduce() 定义3 ：

<U> U reduce(U identity,
 BiFunction<U,? super T,U> accumulator,
 BinaryOperator<U> combiner) 

等同于：

U result = identity; for (T element : this stream)
 result = accumulator.apply(result, element) return result; 

此方法通常用于将map和reduce操作的结合来使用。

下面介绍第一种reduction操作—– stream.collect()

collect()和reduce()不同的是，reduce是从初始值开始，计算得到一个Optional结果，可以转换为一个值。

而collect操作则是将结果放supplier中，supplier即提供一个类或者lamda表达式。

这里说明一下Supplier

collect()方法 定义1 :

<R> R collect(Supplier<R> supplier,
 BiConsumer<R,? super T> accumulator,
 BiConsumer<R,R> combiner) 

该方法可以对stream的元素做可变的reduction 操作，可变的reduction操作指的是这个操作的结果是一个可变的结果容器，比如ArrayList,并且操作过程中是修改结果的状态而不是替换结果中的内容。

这个方法的操作等同于：

R result = supplier.get(); for (T element : this stream)
 accumulator. accept (result, element); return result; 

类似

T reduce(T identity,
 BinaryOperator<T> accumulator) 

Collect操作是线程安全的，并且是一个结束操作 。

Collect方法中的supplier用于创建一个结果容器，在并行操作中，这个方法可能会被多次调用，每次返回一个最新的值。

accumulator —用于将一个元素放入到result container,但要求其符合结合律，互不干涉，无状态性。

combiner—用于合并两个值。必须与accumulator函数一致。

collect()的 定义2 ：

<R,A> R collect(Collector<? super T,A,R> collector) 

通过一个Collector完成可变的container操作。

Collector<? super T,A,R> collector 的例子有：将元素放入Collection中；用StringBuffer把String连起来；

计算元素的min,max,average或者sum。 Collectors提供了很多类似的方法。详情可以参考博客java8中Collectors的使用方法举例和Function

Mutable reduction

我把这个称之为“动态还原”，一个动态还原操作会把元素集合到一个动态的结果容器里，比如Collection, StringBuilder ,因为这些容器在Stream中处理元素。

举例说明：

将一个stream里面的String连接为一个String,可以这样操作：

String concatenated = strings.reduce("", String::concat) 

这个操作也可以并行执行。但是这样的操作往往会损耗性能，因为会执行多次String的copy操作 。更有效的方法是将集合操作的结果放入StringBuilder,StringBuilder是一个动态容器(mutable container)。

因此上面的例子修改为下面的方法，效率会更高

String[] testStrings={"a","b","c","d","e"}; 
 String testStringBuilder=Stream.of(testStrings).parallel().collect(StringBuilder::new, StringBuilder:: append ,StringBuilder::append). toString (); 

我这里附上四种String连接的方法和测试结果，从结果测试来看，的确StringBuilder在处理已经存在的String数据时速度很快，但原来旧的for循环反而比Stream快。当然这里的数据量并不是很大，实际选择时需要自己测试，根据情况选，这篇文章的分析可以参考

String[] testStrings={"a","b","c","d","e","a","b","c","d","e","a","b","c","d","e","a","b","c","d","e","a","b","c","d","e","a","b","c","d","e","a","b","c","d","e","a","b","c","d","e","a","b","c","d","e","a","b","c","d","e","a","b","c","d","e","a","b","c","d","e","a","b","c","d","e"}; 

四种方法由快到慢的顺序是：

方法1：

StringBuilder sumbuilder=new StringBuilder(); for(int i=0;i<testStrings.length;i++)
 sumbuilder.append(testStrings[i]); 

方法2:

String sumString=""; for(int i=0;i<testStrings.length;i++)
 sumString+=testStrings[i]; 

方法3：

String testStringBuilder=Stream.of(testStrings).collect(StringBuilder::new, StringBuilder::append,StringBuilder::append).toString(); 

方法4：

String concatenated = Stream.of(testStrings).reduce("", String::concat); 

对于组装一个list来说，下面的方法是等效的：

方法1：

ArrayList<String> strings = stream.collect(() -> new ArrayList<>(),(c, e) -> c.add(e.toString()), (c1, c2) -> c1.addAll(c2)); 

方法2：

ist<String> strings = stream.map(Object::toString)
 .collect(ArrayList::new, ArrayList::add, ArrayList::addAll); 

方法3：

List<String> strings = stream.map(Object::toString)
 .collect(Collectors.toList()); 

其中，方法3利用了Collectors的一些方法，这些在写代码的过程中非常有用。

比如，还可以利用Collectors转换map:

Collector<Employee, ?, Integer> summingSalaries
 = Collectors.summingInt(Employee::getSalary);
Map<Department, Integer> salariesByDept
 = employees.stream().collect(Collectors.groupingBy(Employee::getDepartment,
 summingSalaries)); 

动态还原操作的要求满足等效性，无论操作拆分与否：

A a1 = supplier.get();
 accumulator.accept(a1, t1);
 accumulator.accept(a1, t2); R r1 = finisher.apply(a1); // result without splitting
 A a2 = supplier.get();
 accumulator.accept(a2, t1); A a3 = supplier.get();
 accumulator.accept(a3, t2); R r2 = finisher.apply(combiner.apply(a2, a3)); // result with splitting 

最后做个总结，在你希望更换掉原有的for循环操作时，如何很在乎运行的时间或者说性能，那么使用Streams操作时，务必做相应的性能检测。

如果对时间的要求不是很高，并且需要对数据进行管道操作等，可以选择Streams方法，代码看起来简洁明了，而且操作简单，但数据量较大时，其实和for循环的性能相当。