java流式计算Stream详解

java流式计算Stream详解

Stream 是什么

​ Stream是Java 8新增的重要特性, 它提供函数式编程支持并允许以管道方式操作集合. 流操作会遍历数据源, 使用管道式操作处理数据后生成结果集合, 这个过程通常不会对数据源造成影响。

​ 同时stream不是一种数据结构，它只是某种数据源的一个视图，数据源可以是一个数组，Java容器或I/O channel等。在Stream中的操作每一次都会产生新的流，内部不会像普通集合操作一样立刻获取值，而是惰性取值，只有等到用户真正需要结果的时候才会执行。

​ Stream代表数据流，流中的数据元素的数量可能是有限的，也可能是无限的。

流和集合的区别

关于流和其它集合具体的区别，可以参照下面的列表：

• 不存储数据。流是基于数据源的对象，它本身不存储数据元素，而是通过管道将数据源的元素传递给操作。

• 函数式编程。流的操作不会修改数据源，例如filter不会将数据源中的数据删除。

• 延迟操作。流的很多操作如filter,map等中间操作是延迟执行的，只有到终点操作才会将操作顺序执行。

• 可以解绑。对于无限数量的流，有些操作是可以在有限的时间完成的，比如limit(n) 或 findFirst()，这些操作可是实现”短路”(Short-circuiting)，访问到有限的元素后就可以返回。

• 纯消费。流的元素只能访问一次，类似Iterator，操作没有回头路，如果你想从头重新访问流的元素，对不起，你得重新生成一个新的流。

为什么需要 Stream API

Stream 作为 Java 8 的一大亮点，它与 java.io 包里的 InputStream 和 OutputStream 是完全不同的概念。

​ 流的操作是以管道的方式串起来的。流管道包含一个数据源，接着包含零到N个中间操作，最后以一个终点操作结束。

集合讲的是数据，流讲的是计算

​ Java 8 中的 Stream 是对集合（Collection）对象功能的增强，它专注于对集合对象进行各种非常便利、高效的聚合操作（aggregate operation），或者大批量数据操作 (bulk data operation)。Stream API 借助于同样新出现的 Lambda 表达式，极大的提高编程效率和程序可读性。

​ 同时它提供串行和并行两种模式进行汇聚操作，并发模式能够充分利用多核处理器的优势，使用 fork/join 并行方式来拆分任务和加速处理过程。通常编写并行代码很难而且容易出错, 但使用 Stream API 无需编写一行多线程的代码，就可以很方便地写出高性能的并发程序。所以说，Java 8 中首次出现的 java.util.stream 是一个函数式语言+多核时代综合影响的产物。

Stream深入理解

简单说，对 Stream 的使用就是实现一个 filter-map-reduce 过程，产生一个最终结果，或者导致一个副作用（side effect）。（副作用不是return，而是流执行过程中对入参的修改，或者文件的输出，或者println()）

​ Stream 不是集合元素，它不是数据结构并不保存数据，它是有关算法和计算的，它更像一个高级版本的 Iterator。

​ 对于 Stream，用户只要给出需要对其包含的元素执行什么操作，比如 “过滤掉长度大于 10 的字符串”、“获取每个字符串的首字母”等，Stream 会隐式地在内部进行遍历，做出相应的数据转换。

​ Stream 就如同一个迭代器（Iterator），单向，不可往复，数据只能遍历一次，遍历过一次后即用尽了，就好比流水从面前流过，一去不复返。

​ Stream 的另外一大特点是，数据源本身可以是无限的（即无限流）。

流操作概述

流的操作类型分为两种：

中间操作

​ 一个流可以后面跟随零个或多个 intermediate 操作。其目的主要是打开流，做出某种程度的数据映射/过滤，然后返回一个新的流，交给下一个操作使用。这类操作都是惰性化的（lazy），就是说，仅仅调用到这类方法，并没有真正开始流的遍历。

map (mapToInt, flatMap 等)、 filter、 distinct、 sorted、 peek、 limit、 skip、 parallel、 sequential、 unordered

终止操作

​ 一个流只能有一个 terminal 操作，当这个操作执行后，流就被使用“光”了，无法再被操作。所以这必定是流的最后一个操作。Terminal 操作的执行，才会真正开始流的遍历，并且会生成一个结果，或者一个 side effect。

forEach、 forEachOrdered、 toArray、 reduce、 collect、 min、 max、 count、 anyMatch、 allMatch、 noneMatch、 findFirst、 findAny、 iterator

短路操作

还有一种操作被称为 short-circuiting（短路操作）

• 对于一个 intermediate 操作，如果它接受的是一个无限流，它可以返回一个有限的新 Stream。

• 对于一个 terminal 操作，如果它接受的是一个无限流，但能在有限的时间计算出结果。

anyMatch、 allMatch、 noneMatch、 findFirst、 findAny、 limit

所有操作函数

中间操作

无状态

• unordered()
• filter()
• map()
• mapToInt()
• mapToLong()
• mapToDouble
• flatMap()
• flatMapToInt()
• flatMapToLong()
• flatMapToDouble()
• peek()

有状态

• distinct()
• sorted()
• limit()
• skip()

结束操作

非短路操作

• forEach()
• ‘forEachOrdered()
• toArray()
• reduce()
• collect()
• max()
• min()
• count()

短路操作

• anyMatch()
• allMatch()
• noneMatch()
• findFirst()
• findAny()

基本类型Stream

IntStream、LongStream、DoubleStream，java特别为这三种基本数值型提供了对应的 Stream。

Java 8 中还没有提供其它数值型 Stream，因为这将导致扩增的内容较多。而常规的数值型聚合运算可以通过上面三种 Stream 进行。

数值流的构造

1
2
3

IntStream.of(new int[]{1, 2, 3}).forEach(System.out::println);
IntStream.range(1, 3).forEach(System.out::println);
IntStream.rangeClosed(1, 3).forEach(System.out::println);

​ range()，需要传入开始节点和结束节点两个参数，返回的是一个有序的LongStream。包含开始节点和结束节点两个参数之间所有的参数，间隔为1.

​ rangeClosed的功能和range类似。差别就是rangeClosed包含最后的结束节点，range不包含。

流的特性

并行 Parallelism

所有的流操作都可以串行执行或者并行执行。

​ 除非显示地创建并行流，否则Java库中创建的都是串行流。 Collection.stream()为集合创建串行流而Collection.parallelStream()为集合创建并行流。IntStream.range(int, int)创建的是串行流。通过parallel()方法可以将串行流转换成并行流,sequential()方法将流转换成串行流。

​ 除非方法的Javadoc中指明了方法在并行执行的时候结果是不确定(比如findAny、forEach)，否则串行和并行执行的结果应该是一样的。

不能干扰

流可以从非线程安全的集合中创建，当流的管道执行的时候，非concurrent数据源不应该被改变。

下面的代码会抛出java.util.ConcurrentModificationException异常：

1
2
3

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

在设置中间操作的时候，可以更改数据源，只有在执行终点操作的时候，才有可能出现并发问题(抛出异常，或者不期望的结果)，比如下面的代码不会抛出异常：

1
2
3
4

List l = new ArrayList(Arrays.asList("one", "two"));
Stream sl = l.stream();
l.add("three");
sl.forEach(System.out::println);

对于concurrent数据源，不会有这样的问题，比如下面的代码很正常：

1
2
3

List l = new CopyOnWriteArrayList<>(Arrays.asList("one", "two"));
Stream sl = l.stream();
sl.forEach(s -> l.add("three"));

​ 虽然我们上面例子是在终点操作中对非并发数据源进行修改，但是非并发数据源也可能在其它线程中修改，同样会有并发问题。

无状态

大部分流的操作的参数都是函数式接口，可以使用Lambda表达式实现。它们用来描述用户的行为，称之为行为参数(behavioral parameters)。

如果这些行为参数有状态，则流的操作的结果可能是不确定的，比如下面的代码:

1
2
3
4
5
6
7
8
9
10
11
12
13
14

List<String> l = new ArrayList(Arrays.asList("one", "two", ……));
class State {
    boolean s;
}
final State state = new State();
Stream<String> sl = l.stream().map(e -> {
    if (state.s)
        return "OK";
    else {
        state.s = true;
        return e;
	}
});
sl.forEach(System.out::println);

上面的代码在并行执行时多次的执行结果可能是不同的。这是因为这个lambda表达式是有状态的。

副作用

​ 流水线上所有操作都执行后，用户所需要的结果（如果有）在哪里？首先要说明的是不是所有的Stream结束操作都需要返回结果，有些操作只是为了使用其副作用(Side-effects)，比如使用Stream.forEach()方法将结果打印出来就是常见的使用副作用的场景

​ 有副作用的行为参数是被不鼓励使用的，事实上，除了打印之外其他场景都应避免使用副作用。也许你会觉得在Stream.forEach()里进行元素收集是个不错的选择，就像下面代码中那样，但遗憾的是这样使用的正确性和效率都无法保证，因为Stream可能会并行执行。大多数使用副作用的地方都可以使用归约操作更安全和有效的完成。

很多有副作用的行为参数可以被转换成无副作用的实现

1
2
3
4
5
6
7
8

ArrayList<String> list = Lists.newArrayList();
for (int i = 0;i<1000;i++) {
    list.add(i+"");
}
ArrayList<String> list2 = Lists.newArrayList();
// 副作用代码
list.parallelStream().forEach(s -> list2.add(s));
System.out.println(list2);

​ 上面的代码结果为，明显在多线程状态下对ArrayList操作发生了错误，同时如果不指定list2的大小，list在扩容时还可能会报ArrayIndexOutOfBoundsException下标越界异常。

可以改成以下无副作用的代码，或者改用并发集合类CopyOnWriteArrayList

1
2
3
4
5
6

ArrayList<String> list = Lists.newArrayList();
for (int i = 0;i<1000;i++) {
    list.add(i+"");
}
List<String> list2 = list.parallelStream().collect(Collectors.toList());
System.out.println(list2);

排序

某些流的返回的元素是有确定顺序的，我们称之为 encounter order。这个顺序是流提供它的元素的顺序，比如数组的encounter order是它的元素的排序顺序，List是它的迭代顺序(iteration order)，对于HashSet,它本身就没有encounter order。

​ 一个流是否是encounter order主要依赖数据源和它的中间操作，比如数据源List和Array上创建的流是有序的(ordered)，但是在HashSet创建的流不是有序的。

​ sorted()方法可以将流转换成encounter order的，unordered可以将流转换成encounter order的。

​ 注意，这个方法并不是对元素进行排序或者打散，而是返回一个是否encounter order的流。

​ 除此之外，一个操作可能会影响流的有序,比如map方法，它会用不同的值甚至类型替换流中的元素，所以输入元素的有序性已经变得没有意义了，但是对于filter方法来说，它只是丢弃掉一些值而已，输入元素的有序性还是保障的。

​ 对于串行流，流有序与否不会影响其性能，只是会影响确定性(determinism)，无序流在多次执行的时候结果可能是不一样的。

​ 对于并行流，去掉有序这个约束可能会提供性能，比如distinct、groupingBy这些聚合操作。

结合性

一个操作或者函数op满足结合性意味着它满足下面的条件：

1

(a op b) op c == a op (b op c)

对于并发流来说，如果操作满足结合性，我们就可以并行计算：

1

a op b op c op d == (a op b) op (c op d)

比如min、max以及字符串连接都是满足结合性的。

函数对象

​ 使用Stream进行函数式编程时经常需要将操作作为参数传入流方法中, 函数对象即将方法或lambda表达式作为对象。

1

List<String> strArray =  Arrays.asList(stringArrays).stream().filter(x>x.contains("Tomas")).collect(Collectors.toList());

上述示例中filter的参数x>x.contains("Tomas")即为一个lambda表达式.

流的创建

可以通过多种方式创建流：

由集合创建流

Java8 中的 Collection 接口被扩展，提供两个获取流的方法 :

• Stream stream() : 返回一个顺序流
• Stream parallelStream() : 返回一个并行流

1
2

Stream<Integer> stream1 = Arrays.asList(1,2,3,4).stream();
Stream<Integer> stream2 = Arrays.asList(1,2,3,4).parallelStream();

java.util.stream.Stream是一个interface, 各种管道中间操作的返回值都是它的实现类, 这允许我们方便地进行参数传递。

由数组创建流

Java8 中的 Arrays 的静态方法 stream() 可以获取数组流 :static Stream stream(T[] array) : 返回一个流重载形式，能够处理对应基本类型的数组

Arrays也提供了创建流的静态方法stream():

1

Arrays.stream(new int[]{1,2,3})

由值创建流

可以使用静态方法 Stream.of(), 通过显示值创建一个流，它可以接收任意数量的参数

public static Stream of(T… values) : 返回一个流

Stream的静态方法of()也可以用来创建流:

1

Stream<String> stream3 =   Stream.of(new String[]{"1","2","3","4"});

由方法创建流

可以使用静态方法 Stream.iterate() 和 Stream.generate(), 创建无限流

迭代流 : public static Stream iterate(final T seed, final UnaryOperator f)

1
2

//初值为1的无限等比数列
Stream.iterate(1, n -> n * 2);

生成流 : public static Stream generate(Supplier s)

1
2

//无限随机数流
Stream.generate(Math::random)

使用IntStream、LongStream、DoubleStream的static方法创建有限流

1
2
3

IntStream.of(new int[]{1, 2, 3});
IntStream.range(1, 3);
IntStream.rangeClosed(1, 3);

使用随机数类的ints()方法创建无限数值流

1
2

Random random = new Random();
IntStream ints = random.ints();

从文件中获得流

使用BufferedReader的lines方法从文件中获得行的流

1
2

BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(new FileInputStream("file.txt")));
Stream<String> lines = bufferedReader.lines();

Files类的操作路径的方法，如list、find、walk等。

其他类提供的创建流

一些类也提供了创建流的方法:

BitSet数值流

1

IntStream stream = new BitSet().stream();

Pattern 将字符串分隔成流

1
2
3

Pattern pattern = compile(",");
Stream<String> stringStream = pattern.splitAsStream("a,b,c,d");
stringStream.forEach(System.out::println);

JarFile 读取jar文件流

1

Stream<JarEntry> stream = new JarFile("").stream();

更底层的使用StreamSupport，它提供了将Spliterator转换成流的方法，目前还不了解

中间操作

​ 流操作是惰性执行的, 中间操作会返回一个新的流对象, 当执行终点操作时才会真正进行计算，下面介绍流的中间操作，除非传入的操作函数有副作用, 函数本身不会对数据源进行任何修改。

​ 这个Scala集合的转换操作不同，Scala集合转换操作会生成一个新的中间集合，显而易见Java的这种设计会减少中间对象的生成。

distinct 唯一

distinct保证数据源中的重复元素在结果中只出现一次, 它使用equals()方法判断两个元素是否相等.

1
2

Stream<String> stream3 = Stream.of(new String[]{"1", "2", "3", "4", "1", "2", "3", "4"});
System.out.println(stream3.distinct().collect(Collectors.toList()));

filter 过滤

filter根据传入的断言函数对所有元素进行检查, 只有使断言函数返回真的元素才会出现在结果中. filter不会对数据源进行修改.

1
2
3

Stream<String> stream3 = Stream.of(new String[]{"1", "2", "3", "4", "6", "7", "8", "9"});
List<String> stringList = stream3.filter(x-> Integer.parseInt(x)%2==0).collect(Collectors.toList());
System.out.println(stringList);

map 映射

map方法根据传入的mapper函数对元素进行一对一映射, 即数据源中的每一个元素都会在结果中被替换(映射)为mapper函数的返回值，也可以根据处理返回不同的数据类型。

1
2
3

Stream<String> stream3 = Stream.of(new String[]{"1", "2", "3", "4", "6", "7", "8", "9"});
List<Integer> integerList = stream3.map(x -> Integer.parseInt(x)).collect(Collectors.toList());
System.out.println(integerList);

flatmap 映射汇总

flatmap方法混合了map + flattern的功能，同时扩展flatMapToDouble、flatMapToInt、flatMapToLong提供了转换成特定流的方法。它将映射后的流的元素全部放入到一个新的流中。它的方法定义如下：

1

<R> Stream<R> flatMap(Function<? super T,? extends Stream<? extends R>> mapper)

​ flatmap适用于多对多或者一对多的映射关系，mapper函数会将每一个元素转换成一个流对象，而flatMap方法返回的一个流包含所有mapper转换后的元素。

下面举个例子来详细说明：

给定一个列表{“aaa”,”bbb”,”ddd”,”eee”,”ccc”}。需要在控制台直接输出aaabbbdddeeeccc字样采用map来做

1
2
3
4
5
6
7
8
9
10

 List<String> list = Arrays.asList("aaa", "bbb", "ddd", "eee", "ccc");
//这里采用了两次forEach循环进行输出，显然不太优雅
list.stream().map(x -> {
    List<Character> characterList = new ArrayList<>();
    char[] chars = x.toCharArray();
    for (char c : chars) {
        characterList.add(c);
    }
    return characterList.stream();
}).forEach(xStream -> xStream.forEach(System.out::print)); //aaabbbdddeeeccc

采用flatMap来做

1
2
3
4
5
6
7
8
9
10

 List<String> list = Arrays.asList("aaa", "bbb", "ddd", "eee", "ccc");
//采用flatMap来做  体会一下flatMap的魅力吧
list.stream().flatMap(x -> {
    List<Character> characterList = new ArrayList<>();
    char[] chars = x.toCharArray();
    for (char c : chars) {
        characterList.add(c);
    }
    return characterList.stream();
}).forEach(System.out::print); //aaabbbdddeeeccc

limit 截断

limit方法指定数量的元素的流。对于串行流，这个方法是有效的，这是因为它只需返回前n个元素即可，但是对于有序的并行流，它可能花费相对较长的时间，如果你不在意有序，可以将有序并行流转换为无序的，可以提高性能。

limit(int n)当流中元素数大于n时丢弃超出的元素, 否则不进行处理, 达到限制流长度的目的.

1
2
3

Stream<Integer> stream3 = Stream.of(3,5,1,4,2,6,8,7);
List<Integer> integerList = stream3.limit(3).collect(Collectors.toList());
System.out.println(integerList);

peek 观察者

生成一个包含原Stream的所有元素的新Stream，同时会提供一个消费函数（Consumer实例），新Stream每个元素被消费的时候都会执行给定的消费函数；这里所说的消费函数有点类似于钩子，每个元素被消费时都会执行这个钩子

​ peek方法会对数据源中所有元素进行给定操作, 但在结果中仍然是数据源中的元素. 通常我们利用操作的副作用, 修改其它数据或进行输入输出.

​ peek接收一个没有返回值的λ表达式，可以做一些输出，外部处理等。map接收一个有返回值的λ表达式，之后Stream的泛型类型将转换为map参数λ表达式返回的类型

1
2
3

Stream<Integer>  stream3 = Stream.of(1,2,3,4,5,6,7,8,9);
List<Integer>  integerList = stream3.peek(x-> System.out.println(x)).collect(Collectors.toList());
System.out.println(integerList);

sorted 排序

sorted()将流中的元素按照自然排序方式进行排序，如果元素没有实现Comparable，则终点操作执行时会抛出java.lang.ClassCastException异常。

​ sorted(Comparator<? super T> comparator)可以指定排序的方式。

​ 对于有序流，排序是稳定的。对于非有序流，不保证排序稳定。

sorted方法用于对数据源进行排序:

1
2
3

Stream<Integer> stream3 = Stream.of(4, 5, 2, 6, 9, 0, 1, 3, 6, 8);
        List<Integer> integerList = stream3.sorted((x, y) -> x - y).collect(Collectors.toList());
System.out.println(integerList);

使用java.util.Comparator是更方便的方法, 默认进行升序排序:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

class Item {
            public Item(int value) {
                this.value = value;
            }
            private int value;
            public int getValue() {
                return value;
            }
            public void setValue(int value) {
                this.value = value;
            }
        }

Stream<Item> stream3 = Stream.of(new Item(4), new Item(3), new Item(6), new Item(9));
List<Item> itemList = stream3.sorted(Comparator.comparingInt(Item::getValue)).collect(Collectors.toList());
itemList.forEach(x -> System.out.print(x.getValue()+","));

使用reversed()方法进行降序排序:

1
2
3

Stream<Item> stream3 = Stream.of(new Item(4), new Item(3), new Item(6), new Item(9));
List<Item> itemList = stream3.sorted(Comparator.comparingInt(Item::getValue).reversed()).collect(Collectors.toList());
itemList.forEach(x -> System.out.print(x.getValue()+","));

skip 跳过

skip(int)返回丢弃了前n个元素的流. 如果流中的元素小于或者等于n，则返回空的流

1
2
3

Stream<Integer> stream3 = Stream.of(3,5,1,4,2,6,8,7);
List<Integer> integerList = stream3.skip(3).collect(Collectors.toList());
System.out.println(integerList);

终点操作

match 断言

1
2
3

public boolean 	allMatch(Predicate<? super T> predicate)
public boolean 	anyMatch(Predicate<? super T> predicate)
public boolean 	noneMatch(Predicate<? super T> predicate)

这一组方法用来检查流中的元素是否满足断言。

• allMatch只有在所有的元素都满足断言时才返回true,否则flase,流为空时总是返回true

• anyMatch只有在任意一个元素满足断言时就返回true,否则flase,

• noneMatch只有在所有的元素都不满足断言时才返回true,否则flase,

1
2
3
4
5
6

System.out.println(Stream.of(1,2,3,4,5).allMatch( i -> i > 0)); //true
System.out.println(Stream.of(1,2,3,4,5).anyMatch( i -> i > 0)); //true
System.out.println(Stream.of(1,2,3,4,5).noneMatch( i -> i > 0)); //false
System.out.println(Stream.<Integer>empty().allMatch( i -> i > 0)); //true
System.out.println(Stream.<Integer>empty().anyMatch( i -> i > 0)); //false
System.out.println(Stream.<Integer>empty().noneMatch( i -> i > 0)); //true

count 计数

count方法返回流中的元素的数量。

1
2

String[] arr = new String[]{"a","b","c","d"};
long count = Arrays.stream(arr).count();

你也可以手动来实现它

1
2

String[] arr = new String[]{"a","b","c","d"};
long count = Arrays.stream(arr).mapToLong(x->1L).sum();

collect 收集

collect(Collector c) 将流转换为其他形式。接收一个 Collector接口的实现，用于给Stream中元素做汇总的方法。辅助类Collectors提供了很多的collector收集器，可以满足我们日常的需求，你也可以创建新的collector实现特定的需求。它是一个值得关注的类，你需要熟悉这些特定的收集器，如聚合类averagingInt、最大最小值maxBy minBy、计数counting、分组groupingBy、字符串连接joining、分区partitioningBy、汇总summarizingInt、化简reducing、转换toXXX等。

收集器

Collectors里常用搜集器介绍：

示例

collect是使用最广泛的终点操作, 也上文中多次出现:

1
2
3

List<String> list = Stream.of("a","b","c","b")
        .distinct()
        .collect(Collectors.toList())

​ toList()将流转换为List实例, 是最常见的用法, java.util.Collectors类中还有求和, 计算均值, 取最值, 字符串连接等多种收集方法。

find 返回

findAny()

返回任意一个元素，如果流为空，返回空的Optional，对于并行流来说，它只需要返回任意一个元素即可，所以性能可能要好于findFirst()，但是有可能多次执行的时候返回的结果不一样。

findFirst()

返回第一个元素，如果流为空，返回空的Optional。

forEach 遍历

forEach遍历流的每一个元素，执行指定的action。它是一个终点操作，和peek方法不同。这个方法不担保按照流的encounter order顺序执行，如果对于有序流按照它的encounter order顺序执行，你可以使用forEachOrdered方法。

forEach方法对流中所有元素执行给定操作, 没有返回值.

1

Stream.of(1,2,3,4,5).forEach(System.out::println);

嵌套遍历(不推荐)

如果要对两个集合进行遍历操作，可以将流嵌套，但是这种遍历的性能跟跟foreach嵌套一样，而且不能进行更复杂的操作，不推荐。

1
2
3
4
5
6
7

ArrayList<String> list = Lists.newArrayList("1", "2");
ArrayList<String> list2 = Lists.newArrayList("一", "二");
list.stream().forEach(str1->{
    list2.stream().forEach(str2->{
        System.out.println(str1+str2);
    });
});

max、min 最大最小值

max返回流中的最大值，

min返回流中的最小值。

1
2
3
4
5
6
7

ArrayList<Integer> list = Lists.newArrayList(3,5,2,1);
Integer max = list.stream().max(new Comparator<Integer>() {
    @Override
    public int compare(Integer o1, Integer o2) {
        return o1 - o2;
    }
}).get();

concat 组合

concat(Stream a, Stream b)用来连接类型一样的两个流。

1
2
3

List<Integer> list1 = Arrays.asList(1,2,3);
List<Integer> list2 = Arrays.asList(4,3,2);
Stream.concat(list1.stream(),list2.stream()).forEach(System.out::print);

toXXX 转换

toArray方法将一个流转换成数组，而如果想转换成其它集合类型，西需要调用collect方法，利用Collectors.toXXX方法进行转换。

toArray()

将流中的元素放入到一个数组中，默认为Object数组

他还有一个重载方法可以返回指定类型的数组

1
2

Object[] objects = Stream.of(1, 2, 3, 4, 5).toArray();
Integer[] integers = Stream.of(1, 2, 3, 4, 5).toArray(Integer[]::new);

reduce 归约

reduce是常用的一个方法，事实上很多操作都是基于它实现的。

方法重载

它有几个重载方法：

PS: BinaryOperator 函数式接口，也即Lambada表达式

reduce思想

reduce是很重要的一种编程思想。这里重点介绍一下。reduce的作用是把stream中的元素给组合起来。至于怎么组合起来：

​ 它需要我们首先提供一个起始种子，然后依照某种运算规则使其与stream的第一个元素发生关系产生一个新的种子，这个新的种子再紧接着与stream的第二个元素发生关系产生又一个新的种子，就这样依次递归执行，最后产生的结果就是reduce的最终产出，这就是reduce的算法最通俗的描述；

​ 所以运用reduce我们可以做sum,min,max,average,所以这些我们称之为针对具体应用场景的reduce，这些常用的reduce，stream api已经为我们封装了对应的方法。

1
2
3
4
5
6
7
8
9
10
11
12
13

//求和 sum
List<Integer> integers = Arrays.asList(1, 2, 3, 4, 5);
// 没有起始值时返回为Optional类型
Optional<Integer> sumOptional = integers.stream().reduce(Integer::sum);
System.out.println(sumOptional.get()); //15

// 可以给一个起始种子值
Integer sumReduce = integers.stream().reduce(0, Integer::sum);
System.out.println(sumReduce); //15

//直接用sum方法
Integer sum = integers.stream().mapToInt(i -> i).sum();
System.out.println(sum); //15

第三个重载

前面两个方法比较简单，重点说说三个参数的reduce(U identity, BiFunction a, BinaryOperator combiner)

三个参数时是最难以理解的。 分析下它的三个参数：

• identity: 一个初始化的值；这个初始化的值其类型是泛型U，与Reduce方法返回的类型一致；注意此时Stream中元素的类型是T，与U可以不一样也可以一样，这样的话操作空间就大了；不管Stream中存储的元素是什么类型，U都可以是任何类型，如U可以是一些基本数据类型的包装类型Integer、Long等；或者是String，又或者是一些集合类型ArrayList等；后面会说到这些用法。
• accumulator: 其类型是BiFunction，输入是U与T两个类型的数据，而返回的是U类型；也就是说返回的类型与输入的第一个参数类型是一样的，而输入的第二个参数类型与Stream中元素类型是一样的
• combiner: 其类型是BinaryOperator，支持的是对U类型的对象进行操作，第三个参数combiner主要是使用在并行计算的场景下；如果Stream是非并行时，它实际上是不生效的。

因此针对这个方法的分析需要分并行与非并行两个场景。

​ 就是因为U和T不一样，所以给了我们更多的发挥。比如设U的类型是ArrayList，那么可以将Stream中所有元素添加到ArrayList中再返回了，如下示例：

1
2
3
4
5
6

ArrayList<String> result = Stream.of("aa", "ab", "c", "ad").reduce(new ArrayList<>(),
                (u, s) -> {
                    u.add(s);
                    return u;
                }, (strings, strings2) -> strings);
System.out.println(result); //[aa, ab, c, ad]

​ 注意由于是非并行的，第三个参数实际上没有什么意义，可以指定r1或者r2为其返回值，甚至可以指定null为返回值。下面看看并行的情况：

​ 当Stream是并行时，第三个参数就有意义了，它会将不同线程计算的结果调用combiner做汇总后返回。注意由于采用了并行计算，前两个参数与非并行时也有了差异！ 看个例子：

1
2
3
4
5

Integer reduce = Stream.of(1, 2, 3).parallel().reduce(
                4,
               (integer, integer2) -> integer + integer2,
               (integer, integer2) -> integer + integer2);
       System.out.println(reduce); //18

输出：18

​ omg，结果竟然是18。显然串行的话结果是10；这个不太好理解，但是我下面写一个等价的方式，可以帮助很好的理解这个结果：

1
2

Optional<Integer> reduce = Stream.of(1, 2, 3).map(n -> n + 4).reduce((s1, s2) -> s1 + s2);
System.out.println(reduce.get()); //18

​ 这种方式有助于理解并行三个参数时的场景，实际上就是第一步使用accumulator进行转换（它的两个输入参数一个是identity, 一个是序列中的每一个元素），由N个元素得到N个结果；第二步是使用combiner对第一步的N个结果做汇总。

reduce能干什么

好了，三个参数的reduce先介绍到这。下面继续看看reduce能为我们做什么？

1
2
3
4
5
6
7
8

//构造字符串流
List<String> strs = Arrays.asList("H", "E", "L", "L", "O");
// reduce
String concatReduce = strs.stream().reduce("", String::concat);
System.out.println(concatReduce); //HELLO
Stream<Integer> integerStream = Stream.of(1, 2, 3, 4, 5);
Integer minReduce = integerStream.reduce(Integer.MAX_VALUE, Integer::min);
System.out.println(minReduce); //1

并发问题

除非显式地创建并行流, 否则默认创建的都是串行流.Collection.stream()为集合创建串行流,而Collection.parallelStream()创建并行流.

stream.parallel()方法可以将串行流转换成并行流,stream.sequential()方法将流转换成串行流.

1
2

Stream<Integer> stream3 = Stream.of(1,2,3,4,5,6,7,8,9);
stream3.forEach(x-> System.out.print(x+","));

输出

1,2,3,4,5,6,7,8,9,

流可以在非线程安全的集合上创建, 流操作不应该对非线程安全的数据源产生任何副作用, 否则将发生java.util.ConcurrentModificationException异常.

1
2

List<String> list = new ArrayList(Arrays.asList("x", "y"));
list.stream().forEach(x-> list.add("z"));

输出

1
2
3
4

Exception in thread "main" java.util.ConcurrentModificationException
	at java.util.ArrayList$ArrayListSpliterator.forEachRemaining(ArrayList.java:1388)
	at java.util.stream.ReferencePipeline$Head.forEach(ReferencePipeline.java:580)
	at com.test.lambda.LambdaTest.main(LambdaTest.java:15)

对于线程安全的容器不会存在这个问题:

1
2
3
4
5

List<String> list = new CopyOnWriteArrayList(Arrays.asList("x", "y"));
list.stream().forEach(x->{
	list.add("z");
	System.out.println(list);
});

输出

[x, y, z]
[x, y, z, z]

当然作者建议Stream操作不要对数据源进行任何修改. 当然, 修改其它数据或者输入输出是允许的:

1
2
3
4
5

Set<String> set = new HashSet<String>();
List<String> list = new CopyOnWriteArrayList(Arrays.asList("x", "y"));
list.stream().forEach(x->{
    set.add(x);
});

理想的管道操作应该是无状态且与访问顺序无关的. 无状态是指操作的结果只与输入有关, 下面即是一个有状态的操作示例:

1
2
3
4
5
6
7
8
9
10

State state = getState();
List<String> list = new ArrayList(Arrays.asList("a", "b"));
list = list.stream().map(s -> {
  if (state.isReady()) {
    return s;
  }
  else {
    return null;
  }
});

无状态的操作保证无论系统状态如何管道的行为不变, 与顺序无关则有利于进行并行计算.

函数式接口

函数式接口会将签名匹配的函数对象(lambda表达式或方法)视作接口的实现。

1
2
3
4
5

@FunctionalInterface
interface Greeter
{
    void hello(String message);
}

函数式接口中有且只有一个非抽象方法。

1

Greeter greeter = message -> System.out.println("Hello " + message);

这在 Java 8 之前通常使用匿名内部类实现的:

1
2
3
4
5
6

Greeter greeter = new Greeter() {
            @Override
            public void hello(String message) {
                System.out.println("Hello " + message);
            }
        };

Java 8 将已有的一些接口实现为函数式接口:

• java.lang.Runnable
• java.util.concurrent.Callable
• java.util.Comparator
• java.lang.reflect.InvocationHandler
• java.io.FileFilter
• java.nio.file.PathMatcher

java.util.function中定义了一些常用的函数式接口:

• Consumer: 接受参数无返回
  • Consumer<T> -> void accept(T t);
  • BiConsumer<T,U> -> void accept(T t, U u);
  • DoubleConsumer -> void accept(double value);
• Supplier: 不接受参数有返回
  • Supplier<T> -> T get();
  • DoubleSupplier -> double getAsDouble();
• Function: 接受参数并返回
  • Function<T, R> -> R apply(T t);
  • BiFunction<T, U, R> -> R apply(T t, U u);
  • DoubleFunction<R> -> R apply(double value);
  • DoubleToIntFunction -> int applyAsInt(double value);
  • BinaryOperator<T> extends BiFunction<T,T,T>
• Predicate: 接受参数返回boolean
  • Predicate<T> -> boolean test(T t);
  • BiPredicate<T, U> -> boolean test(T t, U u);
  • DoublePredicate -> boolean test(double value);

默认构造器可以作为supplier: Supplier<Item> supplier = Item::new;