5.转换Stream
转换Stream其实就是把一个Stream通过某些行为转换成一个新的Stream。
eg：
List<Integer> nums = Lists.newArrayList(1,1,null,2,3,4,null,5,6,7,8,9,10); System.out.println(“sum is:” + nums.stream().filter(num -> num != null) .distinct().mapToInt(num -> num * 2) .peek(System.out::println).skip(2).limit(4).sum() );
说明：给定一个Integer类型的List，获取其对应的Stream对象，然后进行过滤掉null，再去重，再每个元素乘以2，再每个元素被消费的时候打印自身，在跳过前两个元素，最后去前四个元素进行加和运算。
性能问题：
在对于一个Stream进行多次转换操作，每次都对Stream的每个元素进行转换，而且是执行多次，这样时间复杂度就是一个for循环里把所有操作都做掉的N（转换的次数）倍。但是事实上不是这样的，转换操作都是lazy的，多个转换操作只会在聚合（reduce）操作的时候融合起来，一次循环完成。我们可以这样简单的理解，Stream里有个操作函数的集合，每次转换操作就是把转换函数放入这个集合中，在聚合操作的时候循环Stream对应的集合，然后对每个元素执行所有的函数。
5.1 distinct()
Stream<T>distinct()
对于Stream中包含的元素进行去重操作（去重逻辑依赖元素的equals方法），新生成的Stream中没有重复的元素。（根据.equals行为排除所有重复的元素。）
distinct()方法示意图：
5.2 filter( )
Stream<T> filter(Predicate<? super T> predicate)
对于Stream中包含的元素使用给定的过滤函数进行过滤操作，新生成的Stream只包含符合条件的元素。（排除所有与断言不匹配的元素。）
filter()方法示意图：
5.3 map( )
<R> Stream<R> map(Function<? super T,? extends R> mapper)
DoubleStream mapToDouble(ToDoubleFunction<? super T> mapper)
IntStreammapToInt(ToIntFunction<? super T> mapper)
LongStream mapToLong(ToLongFunction<? super T> mapper)
对于Stream中包含的元素使用给定的转换函数进行转换操作，新生成的Stream只包含转换生成的元素。（通过Function对元素执行一对一的转换）
这个方法有三个对于原始类型的变种方法，分别是：mapToInt()，mapToLong()和mapToDouble()。比较好理解，如mapToInt就是把原始Stream转换成一个新的Stream，这个新生成的Stream中的元素都是int类型。此三个变种方法，可以免除自动装箱/拆箱的额外消耗；
map()方法示意图：
5.4 flatMap( )
<R> Stream<R>flatMap(Function<? super T,? extends Stream<? extends R>> mapper)
DoubleStream flatMapToDouble(Function<? super T,? extends DoubleStream> mapper)
IntStream flatMapToInt(Function<? super T,? extends IntStream> mapper)
LongStream flatMapToLong(Function<? super T,? extends LongStream> mapper)
和map类似，不同的是其每个元素转换得到的是Stream对象，会把子Stream中的元素压缩到父集合中。（通过FlatMapper将每个元素转变为无或更多的元素。）
flatMap()方法示意图：
5.5 peek( )
Stream<T>peek(Consumer<? super T> action)
（返回一个流的元素组成的流,另外在每个元素上执行所提供的行动产生的流元素消耗。）
生成一个包含原Stream的所有元素的新Stream，同时会提供一个消费函数（Consumer实例），新Stream每个元素被消费的时候都会执行给定的消费函数.（对每个遇到的元素执行一些操作。主要对调试很有用。）
peek()方法示意图：
5.6 limit( )
Stream<T>limit(long maxSize)
对一个Stream进行截断操作，获取其前N个元素；如果原Stream中包含的元素个数小于N，那就获取其所有的元素。（保证后续的操作所能看到的最大数量的元素。）
limit()方法示意图：
5.7 skip( )
Stream<T> skip(long n)
返回一个丢弃原Stream的前N个元素后剩下元素组成的新Stream，如果原Stream中包含的元素个数小于N，那么返回空Stream。（取N个元素后面的所有元素）
skip()方法示意图：
6.Reduce（聚合）Stream
聚合（也称为折叠）接受一个元素序列为输入，反复使用某个合并操作，把序列中的元素合并成一个汇总的结果。eg：查找一个数字列表的总和或者最大值，或者把这些数字累积成一个List对象。
Stream接口有一些通用的聚合操作，eg：reduce()和collect()；
也有一些特定用途的聚合操作，eg：sum(),max()和count()。
Note：sum()方法不是所有的Stream对象都有的，只有IntStream、LongStream和DoubleStream是实例才有。
聚合操作:
1)可变聚合：把输入的元素们累积到一个可变的容器中，eg:Collection或者StringBuilder。
2)其他聚合：除去可变汇聚剩下的，一般都不是通过反复修改某个可变对象，而是通过把前一次的聚合（汇聚）结果当成下一次的入参，反复如此。eg:reduce()，count()，allMatch()。
6.1 可变聚合（collect）
collect()方法可以把Stream中的所有元素收集到一个结果容器中（eg：Collection）
<R,A>Rcollect(Collector<? super T,A,R> collector)
<R> R collect(Supplier<R> supplier, BiConsumer<R,? super T> accumulator, BiConsumer<R,R> combiner)
方法参数说明：Supplier supplier是一个工厂函数，用来生成一个新的容器；BiConsumer accumulator也是一个函数，用来把Stream中的元素添加到结果容器中；BiConsumer combiner还是一个函数，用来把中间状态的多个结果容器合并成为一个（并发的时候会用到）。
eg:对一个元素是Integer类型的List，先过滤掉全部的null，然后把剩下的元素收集到一个新的List中。
List<Integer> nums = Lists.newArrayList(1,1,null,2,3,4,null,5,6,7,8,9,10); List<Integer> numsWithoutNull = nums.stream().filter(num -> num != null) .collect(() -> new ArrayList<Integer>(), (list, item) -> list.add(item), (list1, list2) -> list1.addAll(list2));
说明：第一个函数生成一个新的ArrayList实例；
第二个函数接受两个参数，第一个是前面生成的ArrayList对象，第二个是stream中包含的元素，函数体就是把stream中的元素加入ArrayList对象中。第二个函数被反复调用直到原stream的元素被消费完毕；
第三个函数也是接受两个参数，这两个都是ArrayList类型的，函数体就是把第二个ArrayList全部加入到第一个中。
Java8还给我们提供了Collector的工具类Collectors，其中已经定义了一些静态工厂方法，
eg：Collectors.toCollection()收集到Collection中；
Collectors.toList()收集到List中；
Collectors.toSet()收集到Set中。
Collectors英文文档地址：
http://docs.oracle.com/javase/8/docs/api/java/util/stream/Collectors.html
eg：
List<Integer> nums = Lists.newArrayList(1,1,null,2,3,4,null,5,6,7,8,9,10); List<Integer> numsWithoutNull = nums.stream().filter(num -> num != null) .collect(Collectors.toList());
6.2其他聚合（或汇聚）
reduce()、count()，sum()等方法。
reduce()方法有三种形式：
Optional<T>reduce(BinaryOperator<T> accumulator)
T reduce(T identity, BinaryOperator<T> accumulator)
<U> U reduce(U identity, BiFunction<U,? super T,U> accumulator, BinaryOperator<U> combiner)
eg：
一个参数的reduce方法：
/** *这个函数有两个参数，第一个参数是上次函数执行的返回值（也称为中间结果），第二个参数是stream中的元素，这个函数把这两个值相加，得到的和会被赋值给下次执行这个函数的第一个参数。Note：**第一次执行的时候第一个参数的值是Stream的第一个元素，第二个参数是Stream的第二个元素。 *这个方法返回值类型是Optional，这是Java8防止出现NPE的一种可行方法。 */ List<Integer> ints = Lists.newArrayList(1,2,3,4,5,6,7,8,9,10); System.out.println("ints sum is:" + ints.stream().reduce((sum, item) -> sum + item).get());
两个参数的 reduce方法：
/** *不同的是：它允许用户提供一个循环计算的初始值，如果Stream为空，就直接返回该值。 *而且这个方法不会返回Optional，因为其不会出现null值。 */ List<Integer> ints = Lists.newArrayList(1,2,3,4,5,6,7,8,9,10); System.out.println("sum is:" + ints.stream().reduce(0, (sum, item) -> sum + item));
count()方法示例：
List<Integer> ints = Lists.newArrayList(1,2,3,4,5,6,7,8,9,10); System.out.println("sum is:" + ints.stream().count());
7. 应用示例
1）Stream API示例
eg：
public class MyStreams { private enum Status { OPEN, CLOSED }; /** * Task类有一个分数的概念（或者说是伪复杂度），其次是还有一个值可以为OPEN或CLOSED的状态. */ private static final class Task { private final Status status; private final Integer points; Task( final Status status, final Integer points ) { this.status = status; this.points = points; } public Integer getPoints() { return points; } public Status getStatus() { return status; } @Override public String toString() { return String.format( "[%s, %d]", status, points ); } } } /** * Stream API极大简化了集合框架的处理 */ public class StreamDemo { public static void main(String[] args) { final Collection< Task > tasks = Arrays.asList( new Task( Status.OPEN, 5 ), new Task( Status.OPEN, 13 ), new Task( Status.CLOSED, 8 ) ); //1.获取tasks中状态为OPEN的总数和 getOpenTotalPoint(); //2.计算所有状态的总和 getTotalPoints(); //3.根据状态分组 getListByStatus(); //4.计算整个集合中每个task分数（或权重）的平均值 //5.从文本文件中逐行读取数据这样典型的I/O操作也很适合用Stream API来处理。 final Path path = new File( filename ).toPath(); try( Stream< String > lines = Files.lines( path, StandardCharsets.UTF_8 ) ) { //对一个stream对象调用onClose方法会返回一个在原有功能基础上新增了关闭功能的stream对象， //当对stream对象调用close()方法时，与关闭相关的处理器就会执行。 lines.onClose( () -> System.out.println("Done!") ).forEach( System.out::println ); } } /** 1.获取tasks中状态为OPEN的总数和 思路： 第一，task集合被转换化为其相应的stream表示。然后，filter操作过滤掉状态为CLOSED的task。 下一步，mapToInt操作通过Task::getPoints这种方式调用每个task实例的getPoints方法把Task的stream转化为Integer的stream。 最后，用sum函数把所有的分数加起来，得到最终的结果。 */ public static void getOpenTotalPoint(){ final long totalPointsOfOpenTasks = tasks .stream() .filter( task -> task.getStatus() == Status.OPEN ) .mapToInt( Task::getPoints ) .sum(); System.out.println( "OPEN--Total points: " + totalPointsOfOpenTasks );//输出结果：OPEN--Total points:18 } /** * stream另一个有价值的地方是能够原生支持并行处理。 * 2.计算所有状态的总和 */ public static void getTotalPoints(){ //这个示例和第一个示例很相似，但这个例子的不同之处在于这个程序是并行运行的， //其次使用reduce方法来算最终的结果。 final double totalPoints = tasks .stream() .parallel() .map( task -> task.getPoints() ) // 或 map( Task::getPoints ) .reduce( 0, Integer::sum ); System.out.println( "Total points (all tasks): " + totalPoints );//输出结果：Total points (all tasks): 26.0 } /** 按照某种准则来对集合中的元素进行分组。 3.根据状态分组 */ public static void getListByStatus(){ final Map< Status, List< Task > > map = tasks .stream() .collect( Collectors.groupingBy( Task::getStatus ) ); System.out.println( map );//输出结果：{CLOSED=[[CLOSED, 8]], OPEN=[[OPEN, 5], [OPEN, 13]]} } /** 4.计算整个集合中每个task分数（或权重）的平均值 */ public static void getPercentage(){ final Collection< String > result = tasks .stream() // Stream< String > .mapToInt( Task::getPoints ) // IntStream .asLongStream() // LongStream .mapToDouble( points -> points / totalPoints ) // DoubleStream .boxed() // Stream< Double > .mapToLong( weigth -> ( long )( weigth * 100 ) ) // LongStream .mapToObj( percentage -> percentage + "%" ) // Stream< String> .collect( Collectors.toList() ); // List< String > System.out.println( result );//输出结果：[19%, 50%, 30%] } }
2）生成斐波那契数列示例（利用Stream API，可以设计更加简单的数据接口。）
生成斐波那契数列，完全可以用一个无穷流表示（受限Java的long型大小，可以改为BigInteger）。
eg：
class FibonacciSupplier implements Supplier<Long> { long a = 0; long b = 1; @Override public Long get() { long x = a + b; a = b; b = x; return a; } } public class FibonacciStream { public static void main(String[] args) { Stream<Long> fibonacci = Stream.generate(new FibonacciSupplier()); fibonacci.limit(10).forEach(System.out::println); //如果想取得数列的前10项，用limit(10)，如果想取得数列的第20~30项,用skip()， //通过collect()方法把Stream变为List。该List存储的所有元素就已经是计算出的确定的元素了. List<Long> list = fibonacci.skip(20).limit(10).collect(Collectors.toList()); } }
Note：用Stream表示Fibonacci数列，其接口比任何其他接口定义都要来得简单灵活并且高效。
3）计算π可以利用π的展开式：π/4 = 1 - 1/3 + 1/5 - 1/7 + 1/9 - ...
eg：
/** * 把π表示为一个无穷Stream */ class PiSupplier implements Supplier<Double> { double sum = 0.0; double current = 1.0; boolean sign = true; @Override public Double get() { sum += (sign ? 4 : -4) / this.current; this.current = this.current + 2.0; this.sign = ! this.sign; return sum; } } public class StreamDemo { public static void main(String[] args) { /* * 这个级数从100项开始可以把π的值精确到3.13~3.15之间 */ Stream<Double> piStream = Stream.generate(new PiSupplier()); piStream.skip(100).limit(10).forEach(System.out::println); } }
输出结果：
3.1514934010709914
3.1317889675734545
3.1513011626954057
3.131977491197821
3.1511162471786824
3.1321589012071183
3.150938243930123
3.132333592767332
3.1507667724908344
3.1325019323081857
4）利用欧拉变换对级数进行加速，
可以利用下面的公式：
/** * 用代码实现就是把一个流变成另一个流 */ class EulerTransform implements Function<Double, Double> { double n1 = 0.0; double n2 = 0.0; double n3 = 0.0; @Override public Double apply(Double t) { n1 = n2; n2 = n3; n3 = t; if (n1 == 0.0) { return 0.0; } return calc(); } double calc() { double d = n3 - n2; return n3 - d * d / (n1 - 2 * n2 + n3); } } public class StreamDemo { public static void main(String[] args) { /* * 可以在10项之内把π的值计算到3.141~3.142之间： */ Stream<Double> piStream2 = Stream.generate(new PiSupplier()); piStream2.map(new EulerTransform()) .limit(10) .forEach(System.out::println); } }
输出结果：
0.0
0.0
3.166666666666667
3.1333333333333337
3.1452380952380956
3.13968253968254
3.1427128427128435
3.1408813408813416
3.142071817071818
3.1412548236077655
可以多次使用上面的加速器,下面输出结果自己测试哦。
/* 20项之内可以计算出极其精确的值 */ Stream<Double> piStream3 = Stream.generate(new PiSupplier()); piStream3.map(new EulerTransform()) .map(new EulerTransform()) .map(new EulerTransform()) .map(new EulerTransform()) .map(new EulerTransform()) .limit(20) .forEach(System.out::println);
8.流（Stream）的串行与并行
一个流就像一个地带器。这些值“流过”（模拟水流）然后他们离开。一个流可以只被遍历一次，然后被丢弃。流也可以无限使用。
流能够是 串行的 或者 并行的。 它们可以使用其中一种方式开始，然后切换到另外的一种方式；使用stream.sequential()（切换串行）或stream.parallel()（切换并行）来达到这种切换。串行流在一个线程上连续操作。而并行流就可能一次出现在多个线程上。
Note:
重要的是要意识到并行不是毫无代价的。从性能的立场它不是无代价的，不能简单的将顺序流替换为并行流，且不做进一步思考就期望得到相同的结果。在并行化一个流以前，需要考虑很多特性，关于流、它的操作以及数据的目标方面。eg：访问顺序确实对我有影响吗？我的函数是无状态的吗？我的流有足够大，并且我的操作有足够复杂，这些能使得并行化是值得的吗？
如何通过并行Stream来提升性能？
//如何并行Stream来提升性能 class StreamDemo { public static void main(String[] args) { //创建一个没有重复元素的大表 int max =1000000; List<String> val=new ArrayList<>(max); for(int i=0;i<max;i++){ UUID id=UUID.randomUUID(); val.add(id.toString()); } //算一下排序这个Stream要耗时多久 //串行排序： getSequentialStreamTime();// 串行耗时: 899 ms //并行排序 getParallelStreamTime();// 并行排序耗时: 472 ms //通过比较两段方法代码几乎是一样的，但是并行版的快了50%之多， //唯一需要做的改动就是将stream()改为parallelStream()。 } public static void getSequentialStreamTime(){ long t0 = System.nanoTime(); long count = values.stream().sorted().count(); System.out.println(count); long t1 = System.nanoTime(); long millis = TimeUnit.NANOSECONDS.toMillis(t1 - t0); System.out.println(String.format("sequential sort took: %d ms", millis)); } public static void getParallelStreamTime(){ long t0 = System.nanoTime(); long count = values.parallelStream().sorted().count(); System.out.println(count); long t1 = System.nanoTime(); long millis = TimeUnit.NANOSECONDS.toMillis(t1 - t0); System.out.println(String.format("parallel sort took: %d ms", millis)); } }
四.Stream与Collection区别
Collection是关于静止的数据结构，而Stream是有关动词算法和计算的。
Collection是主要面向内存，存储在内存中；Stream主要是面向CPU，通过CPU实现计算的。
理解说明：
将一个影片存储在DVD盘上，这是一个集合，因为它包含整个电影的字节数据结构，而这个影片被放在互联网上，我们通过视频软件去观看它时，它实际是被流化了，它变成了一个字节流，流是与时间有关的概念，而数据结构是与时间无关，不会随着时间变化变化，流正好相反，随着时间不断地动态变化，如同水流一样潺潺不断。
所以，集合与流的主要区别是是否需要被计算，集合是一个内存数据结构，集合中每个元素在加入到集合之前已经被计算了，相反，流是在即时要求即时计算。
使用集合需要开发者主动去遍历，使用一个遍历循环，这称为外部遍历。
使用一个流库需要使用内部遍历，它自己为你遍历元素，然后将结果保存在某处，你只要提供一个函数，它就会用这个函数对元素处理完成。
eg：
List<String> transactionIds = new ArrayList<>(); for(Transaction t: transactions){ transactionIds.add(t.getId()); //外部遍历 } List<Integer> transactionIds = transactions.stream() .map(Transaction::getId) //内部遍历 .collect(toList()) }
Stream操作不同于Collection操作有两个根本点：
1）管道Pipelining: 许多流Stream操作返回流Stream自身，这就允许对其操作可以像链条一样排列，变成一个管道，这其中也会激活比如懒加载和short-circuiting操作。
2）内部迭代：相比于集合Collection是显式迭代(需要我们编码完成迭代)，Stream操作是在其内部完成迭代操作。
五.为什么不在集合类实现元素迭代等操作，而是定义了全新的Stream API？
以下Oracle官方给出的解释：
1）集合类持有的所有元素都是存储在内存中的，非常巨大的集合类会占用大量的内存，而Stream的元素却是在访问的时候才被计算出来，这种“延迟计算”的特性有点类似Clojure的lazy-seq，占用内存很少。
2）集合类的迭代逻辑是调用者负责，通常是for循环，而Stream的迭代是隐含在对Stream的各种操作中，eg：map()。
要理解“延迟计算”，不妨创建一个无穷大小的Stream。
分析：如果要表示自然数集合，显然用集合类是不可能实现的，因为自然数有无穷多个。但是Stream可以做到。
eg：
/** * 自然数集合的规则非常简单，每个元素都是前一个元素的值+1。 * 反复调用get()，将得到一个无穷数列，利用这个Supplier，可以创建一个无穷的Stream */ class NaturalSupplier implements Supplier<Long> { long num = 0; public Long get() { this.num = this.num + 1; return this.num; } }
class StreamDemo { public static void main(String[] args) { /** * 对这个Stream做任何map()、filter()等操作都是完全可以的，这说明Stream API对Stream进行转换并生成一个新的Stream并非实时计算，而是做了延迟计算。 * 当然，对这个无穷的Stream不能直接调用forEach()，这样会无限打印下去。但是我们可以利用limit()变换，把这个无穷Stream变换为有限的Stream。 */ Stream<Long> natural = Stream.generate(new NaturalSupplier()); natural.map((x) -> { return x * x; }).limit(10).forEach(System.out::println); } }
