Stream 流的基本概念以及常见应用

流的基本概念

流是一种指定的计算视图。流遵循 “做什么而非怎么做” 的原则,它比循环更易于阅读。可以让你以一种声明的方式处理数据。

例如代码:有一个变量名为 words,它是一个集合,里面一本书中所有的单词,需要找出单词长度大于 12 的单词数量。

循环

1
2
3
4
5
long count = 0;

for (String w : words) {
if (w.length > 12) count++;
}

1
2
3
let count = words.stream()
.filter(w -> w.length > 12)
.count();

典型流程

由上述流示例代码可知,流的典型流程:

  1. 创建一个流:words.stream ()
  2. 指定将初始化流转换为其它流的中间操作(可能包含多个步骤):.filter (w -> w.length > 12)
  3. 应用终止操作,从而产生结果。这个操作会强制执行之前的惰性操作。从此之后,这个流就在也不能使用了:.count ();

流与集合的区别

  1. 流并不存储元素,这些元素可能存储在底层的集合中,或者是按需生成的。
  2. 流的操作不会修改其数据源 例如,filter () 方法不会从新的流中移除元素,而是会生成一个新的流,其中不包含被过滤掉的元素。
  3. 流的操作是尽可能惰性执行的这意味着直至需要其结果时,操作才会执行。

流的常见应用

创建流

流有很多创建方式,举几个例子:

创建任意数量的流

1
Stream<String> word = Stream. of("1231254135".split("1"));

从数组的指定位置创建流

1
Stream<String> song = Array.stream(words, 1, 3);

创建空流

1
Stream<String> silence = Stream.emty();

Function

Supplier<T>

1
Stream<String> echos = Stream.generate(() -> "Echo");
1
Stream<Double> randoms = Stream.generate(Math::random);

UnaryOperation<T>

1
Stream<BigInteger> intergers = Stream.iterate(BigInteger.ZERO, n -> n.add(BigInteger.ONE));

上述代码使用 iterate 方法创建了一个无限序列(0 1 2 3 4 5…)。第一个参数是种子 BigInteger.ZERO,第二个元素是 f (seed),即 1,下一个元素是 f (f (seed)),依次类推。

Pattern

正则分割产生流。

1
Stream<String> words = Pattern.compile("\\PL+") .splitAsStream(contents);

操作流

常用的几个操作流的方法

filter

过滤流中的元素

1
Stream<String> longWords = words.stream().filter(w -> w.length > 12);

map

转换流中元素

1
2
// 流中所有的单词转换为小写
Stream<String> lowerCaseWords = words.stream().map(String::toLowerCase);
1
2
// 流中所有的单词,通过截取字符串,转换为首字母
Stream<Strirng> firstLette = wordds.stream().map(s -> s.substring(O, 1));

flatMap

通过传入参数(Function<? super T , ? extends R> mapper),将流中所有的元素产生的结果连接在一起产生一个新的流。

例如:letters 方法,将单词转为字母流返回。letters (“boat”) 的返回值是:流 [“b”, “o”, “a”, “t”]。

1
2
3
4
5
6
7
public static Stream<String> letters(String s) { 
List<String> result = new ArrayList<>();
for (int i = 0; i < s.length); i++)
result.add(s.substring(i, i +1));
}
return result.stream();
}

使用 words 单词流调用 letters 方法,将会返回一个包含流的流 Stream<Stream<String>>

1
Stream<Stream<String>> result = words.stream().map(w -> letters(w));

上面代码的结果并不是我们想要的,我们想要的是单词流转为字母流,而不是一个流中还包含另一个流。

这时就需要使用到 flatMap 方法,此方法会摊平流中包含的字母流。将流:[[“y”, “o”, “u”, “r”], [“b”, “o”, “a”, “t”]] 摊平为 [“y”, “o”, “u”, “r”,”b”, “o”, “a”, “t”]。

1
Stream<String> result = words.stream().flatMap(w -> letters(w));

limit(n)

丢弃第 n 个位置之后的元素

1
2
// 截止到第 100 个元素,创建流
Stream<Double> randoms = Stream.generate(Math::random).limit(100)

skip(n)

与 limit 相反,丢弃第 n 个位置之前的元素

1
2
// 跳过第1个元素,创建流
Stream<String> words = Stream.of(content.split("\\PL+")).skip(1)

concat

连接流

1
Stream<String> word = Stream.concat(lettes("Hello"), letters("World"))

distinct

去重

1
Stream<String> uniqueWords = Stream.of("a", "a", "b").distinct()

sort

排序

1
2
// 按单词长度,从短到长升序
Stream<String> longesFirst = words.stream().sorted(Comparator.comparing(String::length))

peek

在每次访问一个元素时,都会调用 peek 方法中的函数,对于调试来说非常方便。

1
2
3
Object[] powers = Stream.iterate(1.O, p -> p * 2)
.peek(e -> System.out.println("Fetching " + e))
.limit(20).toArray();

简单约简 / 终止操作

约简是一种终结操作( terminal operation ),它们会将流约简为可以在程序中使用的非流值。

count

返回流中元素的数量

max

返回流中最大的元素

min

返回流中最小的元素

findFirst

找到流中的第一个元素

findAny

找到流中的任意一个元素

anyMatch

根据指定参数(匹配条件),判断流中是否含有元素符合

allMatch

根据指定参数(匹配条件),判断流中是否所有元素符合

noneMatch

根据指定参数(匹配条件),判断流中是否所有元素都不符合

收集结果

forEach

此方法会将传入的函数,应用于每个元素

1
stream.forEach(System.out::println);

toArray

返回 Object [] 数组

1
String[] result = stream.toArray(String[]::new)

collect

将流中的元素收集到另一个目标

1
stream.collect(Collectors.tolist());

控制获得结果集的类型

1
TreeSet<String> result = stream.collect(Collectors.toCollection(TreeSet::new));

收集流中的字符串

1
String result = stream.collect(Collectors.joining())

元素间加入分隔符收集结果

1
String result = stream.collect(Collectors.joining("、"))

将其它类型对象,转为字符串收集

1
String result = stream.map(Object::toString).collect(Collectors.joining("、"))

如果要将流的结果约简为总和、平均值、最大值或最小值,可以使用 summarizing (Int|Long|Double) 方法中的某一个。

1
2
3
IntSummaryStatistics summary = stream.collect(Collectors.summarizingInt(String::length));
double averageWordLength = summary.getAverage();
double maxWordLength = summary.getMax();

收集至 Map

使用 Collectors.toMap,可以将想要的元素收集至 Map 中

1
Map<Integer, String> idToName = people.collect(Collectors.toMap(Person::getId, Person::getName));
1
Map<Integer, Person> idToPerson = people.collect(Collectors.toMap(Person::getId, Function.identity()));

一个 key,多个 value

通过第三个参数,传入的函数,控制当一个 key,存在多种 value 的情况。

1
2
3
4
5
6
7
8
9
10
11
12
Stream<Locale> locales = Stream.of(Locale.getAvailableLocales());

Map<String, String> languageNames = locales.collect(

Collectors.toMap(

Locale::getDisplayLanguage,

l -> l.getDisplayLanguage(l),

(existingValue, newValue) -> existingValue))

控制指定收集 TreeMap 类型数据

1
2
3
4
5
6
7
8
9
10
11
Map<Integer, Person> idToPerson = people.collect(

Collectors.toMap(

Person::getId,

Function.identity(),

(existingValue, newValue) -> { throw new IllegalStateException(); },

TreeMap::new));

对于每一个 toMap 方法,都有一个等价的可以产生并发映射表的 toConcurrentMap 方法。单个并发映射表可以用于并行集合处理。当使用并行流时,共享的映射表比合并映射表要更高效。 注意,元素不再是按照流中的顺序收集的,但是通常这不会有什么问题

群组与分区

将需要相同值的元素,分成一组。可以使用 groupingBy 方法。

1
2
3
4
5
Stream<Locale> locales = Stream.of(Locale.getAvailableLocales());

Map<String, List<Locale>> countryToLocales = locales.collect(

Collectors.groupingBy(Locale::getCountry));

当要分组的 key 为 boolean 类型时,使用 partitioningBy 更加高效。

1
2
3
4
5
Map<Boolean, List<Locale>> englishAndOtherLocales = locales.collect(

Collectors.partitioningBy(l -> l.getLanguage().equals("en")));

List<Locale> englishLocales = englishAndOtherLocales.get(true);

如果调用 groupingByConcurrent 方法,就会在使用并行流时获得一个被并行组装的并行映射表。这与 toConcurrentMap 方法完全类似。

下游收集器

如果要控制分组的 value 时,需要提供一个 “下游收集器(downstream collector)”。例如我们想收集的 value 为 set 类型,而非列表 list。

1
Map<String, Set<Locale>> countryToLocaleSet = locales.collect(Collectors.groupingBy(Locale::getCountry, Collectors.toSet()));

除了可以使用 toSet (),也可以使用 counting、summingInt、maxBy 等约简方法。

mapping

此方法会将传入的函数,应用到下游收集器的结果上。例如:还是上面的程序,我们想收集 Map 类型,其中 key 是字符串,value 是 Set<String> 类型。

1
2
3
4
5
6
7
8
import java.util.stream.Collectors.*

...

Map<String, Set<String>> countryToLanguages = locales.collect(
groupingBy(Locale::getDisplayCountry,
mapping(Locale::getDisplayLanguage,
toSet())));

约简操作

reduce 方法,支持自定义约简函数。

1
2
3
4
List<Integer> values = . . .;

// 计算流中元素的和
Optional<Integer> sum = values.stream().reduce(Integer::sum)

上述代码,如果流中的元素用 V 表示,具体在流中就会执行 V0 + V1 + V2 + … Vi 个元素。如果流为空,就会返回一个 Optional 里面为空的对象。

在实践中,建议通过 toMap (),转为数字流,并使用其自带的求和、最大值、最小值等方法更容易。

基本类型流

在流库中,有针对基本数据类型使用的流类型。IntStream、LongStream、DoubleStream,用来直接存储基本类型值,而无需使用包装器,如果想要存储 short、char、byte、boolean,可以使用 IntStream ,而对于 float ,可以使用 DoubleStream。

转为对象流

使用 boxed 方法

1
2
// 生成0~100范围内的基本类型流,并转为包装对象流
Stream<Integer> integers = IntStream.range(0, 100).boxed();

并行流

并行流就是将一个流的内容分成多个数据块,并用不同的线程分别处理每个不同数据块的流。

默认情况下,从有序集合(数组和列表)、范围、生成器和迭代产生的流,或者通过调用 Stream.sorted 产生的流,都是有序的。它们的结果是按照原来元素的顺序累积的,因此是完全可预知的。如果运行相同的操作两次,将会得到完全相同的结果。

创建并行流

parallel ():产生一个与当前流中元素相的并行流

unordered ():产生一个与当前流中元素相 的无序流

parallel Stream ():用当前集合中的元素产生一个并行流

乱序执行

打印流中的每个元素。由于并行流使用不同线程处理不同数据块,那么线程的执行先后顺序也变的不可知,所以打印的数字乱序。

1
2
3
4
5
6
7
8
List<Integer> numbers = Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9);
System.out.println("-------------stream---------------");
numbers.stream().forEach(out::print);
numbers.stream().forEach(out::print);
System.out.println("");
System.out.println("----------parallelStream----------");
numbers.parallelStream().forEach(out::print);
numbers.parallelStream().forEach(out::print);
1
2
3
4
5
6
-------------stream---------------
123456789
123456789
----------parallelStream----------
657893421
643157289

注意:不要将所有的流都转换为并行流,只有在对已经位于内存中的数据执行大量计算操作时,才应该使用并行流。

为了让并行流正常工作,需要满足大量的条件:

  • 数据应该在内存中 必须等到数据到达是非常低效的。
  • 流应该可以被高效地分成若干个子部分 由数组或平衡二叉树支撑的流都可以工作得很好,但是 Stream.iterate 返回的结果不行。
  • 流操作的工作量应该具有较大的规模。如果总工作负载并不是很大,那么搭建并行计算时所付出的代价就没有什么意义。
  • 流操作不应该被阻塞。