1.什么是Stream

Java8中， Collection新增了两个流方法，分别是stream()和parallelStream()。
Java8中添加了一个新的接口类Stream，相当于高级版的Iterator，它可以通过Lambda表达式对集合进行大批量数据操作，或者各种非常便利、高效的聚合数据操作。

2.为什么要使用Stream

在Java8之前，我们通常是通过for loop或者Iterator迭代来重新排序合并数据，又或者通过重新定义Collections.sorts的Comparator方法来实现，这两种方式对于大数据量系统来说，效率并不是很理想。
Stream的聚合操作与数据库sql胡聚合操作sorted、filter、map等类似。我们在应用层就可以高效地实现类似数据库sql的聚合操作了，而在数据操作方面，Stream不仅可以通过串行的方式实现数据操作，还可以通过并行的方式实现大批量数据，提高数据的处理效率。

3.Stream使用

/*** @author 公众号：IT三明治* @date 2022/1/18*/
public class StreamTest {public static void main(String[] args) {List<String> names = Arrays.asList("Sandwich", "Jack", "Sam", "Mark", "Jackson", "Seamus", "Tom");//找出S开头的最长的名字长度System.out.println(getMaxLengthStartWithS(names));System.out.println(getMaxLengthStartWithS1(names));}private static int getMaxLengthStartWithS(List<String> names) {return names.stream().filter(name -> name.startsWith("S")).mapToInt(String::length).max().orElse(0);}private static int getMaxLengthStartWithS1(List<String> names) {List<Integer> lengthList = new ArrayList<>();for (String name: names) {if (name.startsWith("S")) {lengthList.add(name.length());}}return Collections.max(lengthList);}
}

由上面两种方法实现相同的功能，使用for loop循环的方式，代码行数更多，Stream的方式更简洁。

4.Stream操作分类

Stream的操作可以分成两大类: 终结操作(Terminal operations)和中间操作(Intermediate operations)

• 中间操作会返回一个新的流，一个流可以后面跟随0个或者多个中间操作。其目的是打开流，做出某种程度的数据映射/过滤，然后会返回一个新的流，交给下一个操作使用。这类操作都是惰性化的(Lazy)，就是说，仅仅调用到这个类方法，并没有真正开始流的遍历。而是在终结操作开始的时候才真正开始执行。
  中间操作又可以分为无状态(Stateless)和有状态(Stateful)操作，无状态是指元素的处理不受之前元素的影响，有状态是指该操作只有拿到所有元素之后才开始执行。
• 终结操作是指返回最终的结果。一个流只能有一个终结操作，当这个操作执行后，这个流就结束了，无法再被操作。终结操作的执行才会真正开始流的遍历，并且会生成一个结果。
  终结操作又可以分为短路(Short-circuiting)和非短路(Unshort-circuiting)操作。 短路是指遇到某些符合条件的元素就可以得到最终结果；非短路是指必须处理完所有元素才能得到最终结果。

总结如下图所示

实例如下：

/*** @author 公众号：IT三明治* @date 2022/1/18*/
public class ColleagueStream {public static List<Colleague> dataInit() {return Arrays.asList(new Colleague("Sandwich", 18, "male"),new Colleague("Jack", 25, "male"),new Colleague("Kate", 16, "female"),new Colleague("Merry", 16, "female"),new Colleague("Michael", 30, "male"),new Colleague("Dannel", 16, "male"));}public static Map<String, List<Colleague>> groupBy(List<Colleague> colleagues) {return colleagues.stream().collect(Collectors.groupingBy(Colleague::getSex));}public static void main(String[] args) {List<Colleague> colleagues = dataInit();Map<String, List<Colleague>> groups = groupBy(colleagues);Gson gson = new Gson();System.out.println(gson.toJson(groups));}
}

输出结果

{"female": [{"name": "Kate","age": 16,"sex": "female"},{"name": "Merry","age": 16,"sex": "female"}],"male": [{"name": "Sandwich","age": 18,"sex": "male"},{"name": "Jack","age": 25,"sex": "male"},{"name": "Michael","age": 30,"sex": "male"},{"name": "Dannel","age": 16,"sex": "male"}]
}

因为Stream操作类型非常多，我们需要记住常用的

• map(): 将流中的元素进行再次加工形成一个新流，流中的每一个元素映射为另外的元素。
• filter(): 返回结果生成新的流中只包含满足筛选条件的数据。
• limit(): 返回指定数量的元素的流。返回的是Stream里前面n个元素。
• skip(): 和limit()相反，将前几个元素跳过再返回一个流，如果流中的元素小于或者等于n， 就会返回一个空的流。
• sorted(): 将流中的元素按照自然排序方式进行排序。
• distinct(): 将流中的元素去重之后输出。
• peek(): 对流中每个元素执行操作，并返回一个新的流，返回的流还是包含原来流中的元素。

5. Stream的底层实现

一个Stream的各个操作是由处理管道组装，并统一完成数据处理的。
我们知道Stream有中间操作和终结操作，那么对于一个写好的Stream处理代码来说，中间操作是通过AbstractPipeline生成一个中间操作Sink链表。
当我们调用终结操作时，会生成一个最终的ReducingSink，通过这个ReducingSink触发之前的中间操作，从最后一个ReducingSink开始，递归产生一个Sink链。

5.1 stream方法

看以下实例

names是ArrayList集合，所以names.stream()方法将会调用集合类基础接口Collection的Stream方法，然后Stream方法会调用到StreamSupport类的Stream方法。

Stream方法中初始化了一个ReferencePipeline的Head内部类对象

5.2 filter方法

先看看filter怎么用的

/*** @author 公众号：IT三明治* @date 2022/1/22*/
public class ColleagueStream {public static List<Colleague> dataInit() {return Arrays.asList(new Colleague("Sandwich", 18, "male"),new Colleague("Jack", 25, "male"),new Colleague("Kate", 16, "female"),new Colleague("Merry", 16, "female"),new Colleague("Michael", 30, "male"),new Colleague("Dannel", 16, "male"));}public static void main(String[] args) {List<Colleague> colleagues = dataInit();List<Colleague> colleaguesAgeOver20 = getColleaguesAgeOver20(colleagues);Gson gson = new Gson();System.out.println(gson.toJson(colleaguesAgeOver20));}/*** Filter the colleagues whose age over 20* @param colleagues all collections* @return Colleague list whose age over 20*/private static List<Colleague> getColleaguesAgeOver20(List<Colleague> colleagues) {return colleagues.stream().filter(colleague -> colleague.getAge()>20).collect(Collectors.toList());}
}

这个方法是无状态的中间操作，所以执行filter时，并没有进行任何操作，而是分别创建了一个Stage来标识用户的每一次操作。
通常情况下Stream的操作需要一个回调函数，所以Stage是由数据来源，操作，回调函数组成的三无组来表示。
看看它的无状态操作源码是怎么写的

5.3 max方法

看如下实例

看max实现

看reduce实现

看evaluate实现，这里有串行和并行的两种实现

我们选串行的接着看下去


由上可知，当Sink链表生成完后，才开始通过spliterator迭代集合，执行Sink链表中的具体操作。（ps: 链表在源码中用得真的非常多，兄弟们一定要注意掌握了）
java8中的Spliterator的forEachRemaining会迭代集合，每迭代一次，都会执行一次filter操作，如果filter操作通过，就会触发map操作，然后将结果放入到临时数组object中，再执行下一次迭代。完成中间操作后，就会执行终结操作max.

前面我们分析max方法源码的时候发现它其实是可以并行evaluate的，我加个断点试试看它默认用的是哪个

可以看到它默认只会用串行的。
如果我想用并行，该怎么做呢？
我先改造一下代码

再断点看看

可见我们在前面加parallel()是可以让max方法代码并行执行的。
看到并行我就来了兴趣了，为什么呢？从我对计算机微处理器硬件通信和软件数据处理的经验来看，并行往往比串行有更高的效率（虽然这种更高效也是有前提的，后面我会分析到）
我把stream改成并行，看max方法是不是会跟Stream的并行

再断点执行一次

由此可见，Stream改成并行，可以影响后面的方法也用并行处理。前面那种在中间加parallel()方法的方式适合控制整个流里面精确的位置用并行
为了验证这种猜想我前面用并行，后面我改成串行再试一次


这样一个流的操作过程中并行和串行可以精确控制的。

5.4 并发stream源码实现

前面我们用过parallelStream，它是怎么实现的呢？

接着看源码

在Collection的实现里面就把parallel初始化为true了。

这里的并行指的是，Stream结合了ForkJoin框架，对Stream处理进行了分片，Splititerator中的estinateSize方法会估算出分片的数据量。
通过预估的数据量获取最小处理单元的阈值，如果当前分片大小大于最小处理单元的阈值，就会断续切分集合。每个分片将会生成一个Sink链表，当所有的分片操作完成后，ForkJoin框架将会合并分片任何结果集。

6.Stream的性能

6.1常规迭代

用以上实例比较100个int所用的时间

import org.springframework.util.StopWatch;
import java.util.Arrays;
import java.util.Random;/*** @author 公众号：IT三明治* @date 2022/1/22*/
public class Comparator {public static void main(String[] args) {StopWatch stopWatch = new StopWatch("效率比拼");stopWatch.start("Init test data");int size = 100;int[] intArray = new int[size];Random random = new Random();for (int i = 0; i < intArray.length; i++) {intArray[i] = random.nextInt();}stopWatch.stop();int result;stopWatch.start("Iterator比较" + size + "个int最小值花费的时间");result = iteratorForMinInt(intArray);stopWatch.stop();System.out.println("Iterator比较" + size + "个int最小值: "+result);stopWatch.start("串行stream比较" + size + "个int最小值花费的时间");result = serialStreamForMinInt(intArray);stopWatch.stop();System.out.println("串行stream比较" + size + "个int最小值: " + result);stopWatch.start("并行stream比较" + size + "个int最小值花费的时间");result = parallelStreamForMinInt(intArray);stopWatch.stop();System.out.println("并行stream比较" + size + "个int最小值: " +result);System.out.println(stopWatch.prettyPrint());}public static int iteratorForMinInt(int[] arr) {int min = Integer.MAX_VALUE;for (int i : arr) {if (i < min) {min = i;}}return min;}public static int serialStreamForMinInt(int[] arr) {return Arrays.stream(arr).min().orElse(Integer.MIN_VALUE);}public static int parallelStreamForMinInt(int[] arr) {return Arrays.stream(arr).parallel().min().orElse(Integer.MIN_VALUE);}
}

打印结果如下

在同样保证数据准确的前提下Interator方式居然是最快的，stream并行又比串行快不少
为什么会这样呢？

1.常规迭代代码简单，起越简单的代码执行效率越高。
2.Stream串行迭代，使用了复杂的设计，导致执行速度偏低。所以性能最差。
3.Stream并行迭代，使用了Fork-Join线程池，可以并行执行，所以比串行Stream迭代快很多，但是比常规要慢一点（因为高复杂度增加了开销）

6.2大数据迭代

接下来我们把数组加大到1亿个试试

可以看到，数据量大的时候，并行的效率最高，不过由于初始化数组太大，初始化已经占据大部分时间
我注释掉初始化数组的时间，再看一下结果

parallelStream都是使用Fork-Join线程池，而线程池线程数仅为cpu的核心数
调整并发线程池数量的系统变量

再看一下结果

由此可见，调整线程池数可以提高并发效率。
为什么会这样？

1.Stream并行迭代使用了Fork-Join线程池，而线程池线程数为cpu的核心数（如我的电脑是8核16线程，在这里核心数算16），大数据场景下，能够利用多线程机制，效率比Stream串行迭代快，同时多线程机制切换带来的开销相对来说还不算多，所以比常规迭代还要快;
2. 常规代码简单，越简单的代码执行效率越高;
3.Stream串行迭代，使用了复杂的设计，导致执行效率偏低，又没有并行的优势，所以性能最低。

我把线程池数据降到2再测试

可以看到Stream并行迭代的效率下降了。
Fork-Join线程池，在大数据场景下，虽然利用多线程机制，但是线程池数太小，会出现多个请求争着执行任务，对于请求的任务来说，他们是交替被执行的。线程池太少效率会降低。
我再把线程池提高到200试试

可以看到线程池太多也影响效率。
线程太多，线程的上下文切换成本过高，导致了效率反而降低了。

7.如何合理使用Stream?

在循环迭代次数较少的情况下，常规的迭代性能反而更好；而在大数据迭代中，parallelStream在合理的线程数上有一定的优势。
由于所有使用并行流paralleStream的地方都是使用同一个Fork-Join线程池，而线程池线程数仅为cpu的核心数，如果对底层不太熟悉的话请不要乱用并行流parallelStream， 特别是服务器核心数比较少的情况下。
大企业的服务器核心数非常多，更适合使用并行Stream。