Java8的Stream流 _ JavaCoreII

原创

honlyx

发布于 2023-02-20 10:03:32

9210

发布于 2023-02-20 10:03:32

文章被收录于专栏：书笔记

Java SE8的流库

流提供了一种让我们可以在比集合更高的概念级别上指定计算的数据视图。

使用流，我们可以说明想要完成什么任务，而不是说明如何去实现它。我们讲操作的具体调度留给具体实现去解决。

从迭代到流的操作

迭代遍历元素，并在每个元素上执行某项操作。

// 常规方式
String contents = new String(Files.readAllBytes(
    Paths.get("alice.txt")), StandardCharaset.UTF_8);
List<String> words = Arrays.asList(contents.split("PL+"));
long count = 0;
for(String w : words){
    if(w.length() > 12) count++;
}

// 使用流
long count = words.stream().filter(w -> w.length > 12).count();

流和集合的差异：

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

操作流的典型流程：

创建一个流
指定将初始流转换为其他流的中间操作，可能包含多个步骤
应用终止操作，从而产生结果。这个操作会强制执行之前的惰性操作。从此之后，这个流就再也不能用了。

流的创建

Collection：使用接口的stream方法将任何集合转换为一个流。

数组：可以使用静态的Stream.of方法

还可以使用Array.stream(array, from, to)可以从数组中位于from（包括）和to（不包括）的元素中创建一个流。

Stream接口有两个创建无限流的静态方法：

generate方法会接受一个不包含任何引元的函数（从技术上讲，是一个Supper<T>接口的对象）。无论何时，只要需要一个流类型的值，该函数就会被调用产生一个这样的值。

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

iterate方法，可以生成无限序列（0，1，2...）。它会接受一个“种子”值，以及一个函数（从技术上讲，是一个UnaryOperation<T>），并且会反复地将函数应用到之前的结果上。

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

注意：JavaAPI中有大量方法可以产生流。例如Pattern类有一个splitAsStream方法，他会按照某一个正则表达式来分割一个CharSequence对象。可以使用下面的语句来将一个字符串分割为一个个的单词 Stream<String> words = Pattern.complie("\PL+").splitAsStream(conotents); 静态的Files.lines方法会返回一个包含了文件中所有行的Stream： try(Stream<String> lines = Files.lines(path)){ Process lines }

filter、map和flatMap方法

流的转换会产生一个新的流，它的元素派生自另一个流中的元素。

filter：转换成产生一个流，它的元素与某种条件相匹配。

filter的引元是一个Predicate<T>，即从T到boolean的函数。

map：按照某中方式来转换流中的值。我们可以使用带有方法引用的map，也可以使用lambda表达式。

使用map时，会有一个函数应用到每个元素上，并且其结果是包含了应用函数后所产生的所有结果的流。

flatMap方法：可以摊平包含流的流。例如[..."y","o","u","r","b","o","a","t",...]转换为..."y","o","u","r","b","o","a","t"...

注意：在流之外的类中你也会发现flatMap方法，因为它是计算机科学中的一种通用概念。 假设有一个泛型G，以及将其某种类型T转换为G<U>的函数f和将类型U转换为G<V>的函数g。 然后，我们可以通过使用flatMap来组合它们，即首先应用f，然后应用g。 这是单子论的关键概念。

抽取子流和连接流

stream.limit(n)会返回一个新的流，它在n个元素之后结果（如果原来的流更短，那么就会在流结束时结束）。

stream.skip(n)：它会丢弃前n个元素。

stream.concat(stream1,stream2...)：将多个流连接起来。第一个流不应该是无限的，否则第二个流永远都不会得到处理的机会。

其他的流转换

distinct：返回一个流，它的元素是从原有流中产生的，即原来的元素按照同样的顺序剔除重复元素后产生的。

流的排序：有多种sorted方法的变体可用。

操作Comparable元素的流
接受一个Comparator

sorted方法会产生一个新的流，它的元素是原有流中按照顺序排列的元素。

peek：会产生另一个流，它的元素与原来流中的元素相同，但是在每次获得一个元素时，都会调用一个函。对于调试很方便。

简单约简

从流数据中获取答案。

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

count：返回流中元素的数量

max/min：返回最大值和最小值。方法返回的是一个Optional<T>的值。要么其中包装了答案，要么表示没有任何值。

Optional<String> largest = words.max(String::compareToIgnoreCase);
System.out.println("largest: " + largest.orElse(""));

findFirst：返回的是非空集合中的第一个值。一般和filter结合使用

Optional<String> stratWithQ = words.filter(w -> w.startWith("Q")).findFist();

findAny：表示任意的匹配都可以。这个方法在并行处理流时会很有效。

Optional<String> stratWithQ = 
    words.parallel().filter(w -> w.startWith("Q")).findAny();

anyMatch：是否存在匹配。这个方法接受一个断言引元，因此不需要使用filter

boolean aWordStartWithQ = words.parallel().anyMatch(s -> s.startWith("Q"));

allMatch和noneMatch方法：分别会在所有元素和没有任何元素匹配断言的情况下返回true。

Optional类型

Optional<T> 对象是一种包装器对象，要么包装了类型T的对象，要么没有包装任何对象。

对于对一种情况，我们称为值为存在的。

Optional<T>类型被当作一种更安全的方式，用来替代类型T的引用，这种引用要么引用某个对象，要么为null。

如何使用Optional值

关键：它在值不存在的情况下会产生一个可替代物，而只有在值存在的情况下才会使用这个值。

例如：

// 如果值不存在时使用空字符串替换
String result = optionalString.orElse("");
// 如果值不存在时使用默认值
String result = optionalString.orElseGet(
    () -> Locale.getDefault().getDisplayName());
// 或者在没有任何值时抛出异常
String result = optionalString.orElseThrow(IllegalStateException::new);

上面的都是在不存在任何值的情况下产生相应的替代物。

另一条使用可选值的策略是在其存在的情况下才消费该值。 ifPresent方法会接受一个函数。

如果该可选值存在，那么它会被传递给该函数。否则，不会发生任何事情。

optioanlValue.ifPresent(v -> Process V);

如果该值存在的情况下想要将其添加到某个集中

optionalValue.ifPresent(v -> results.add(v));
// 或者
optionalValue.ifPresent(results::add);

调用ifPresent时，该函数不会返回任何值。

要处理函数的结果，应该使用map：

Optional<Boolean> added = optionalValue.map(results::add);

added有三种值之一：optionalValue存在的情况下包装在optional中的true或false，以及在optionalValue不存在的情况下的空optional。

注意：这个map方法与stream接口的map方法类似。可以直接将可选值想象成尺寸为0或1的流。结果的尺寸也是0或1，并且在最后一种情况中，会应用到函数。

不适合使用Optional值的方式

get方法会在Optional值存在的情况下获得其中包装的元素或者在不存在的情况下抛出一个NoSuchElementException对象。因此

Optional<T> optionalValue = ...;
optionalValue.get().someMethod();
// 上面的操作并不比下面的安全
T value = ...;
value.someMethod();

isPresent方法会报告某个Optional<T>对象是否具有一个值。

if(optionalValue.isPresent()) optionalValue.get().someMethod();
// 上面的操作并不比下面更容易操作
if(value != null) value.someMethod();

创建Optional值

Optional.of(result)和Optional.empty();

ofNullable方法用来作为可能出现null值和可选值之间的桥梁。Optional.ofNullable(obj)会在obj不为null的情况下返回Optional.of(obj)，否则会返回Optional.empty();

用flatMap来构建Optional值的函数

假设你有一个可以产生Optional<T>对象的方法f，并且目标类型T具有一个可以产生Optional<U>对象的方法g。如果它们都是普通的方法，那么你可以通过调用s.f().g()将它们组合起来。但是这种组合没办法工作，因为s.f()的类型为Optional<T>，而不是T，需要调用

Optional<U> result = s.f().flatMap(T::g);

如果s.f()的值存在，那么g就可以应用到上面。否则，就会返回一个空Optional<U>。

如果有更多的可以产生Optional值的方法或Lambda表达式，那么就可以重复此过程。可以直接将对flatMap的调用链接起来，从而构建由这些步骤构成的管道，只有所有步骤都成功时，该管道才成功。

例如：

public static Optional<Double> squareRoot(Double x){
    return x < 0 ? Optional.empty() : Optional.of(Math.sqrt(x));
}
// 计算倒数的平方根
Optional<Double> result = inverse(x).flatMap(MyMath::squareRoot);

//可以选择下面的方式
Optional<Double> result =
    Optional.of(-0.4).flatMap(MyMath::inverse).flatMap(MyMath::squareRoot);

收集结果

iterator：产生可以用来访问元素的旧式风格的迭代器

forEach：将某个函数应用到每个元素；在并行流上，forEach方法会以任意的顺序遍历各个元素。如果想要按照流中的顺序来处理它们，可以使用forEachOrdered方法。这个方法会丧失并行处理的部分甚至全部优势。

toArray：获取由流的元素构成的数组。因为无法在运行时创建一个泛型数组，所以表达式stream.toArray()会返回一个Object[]数组。如果想要让数组具有正确的类型，可以将其传递到数组构造器中：String[] result = stream.toArray(String[]::new);

collection：将流中的元素收集到另一个目标中，接受一个Collector接口的实例。Collectors类提供了大量用于生成公共收集器的工厂方法。

List<String> result = stream.collect(Collectors.toList());
//
Set<String> result = stream.collect(Collectors.toSet());
// 控制集的种类
TreeSet<String> result = stream.collect(Collectors.toCollection(TreeSet::new));
// 通过链接操作来收集流中的所有字符串
String result = stream.collect(Collectors.join(", "));

summarizing(Int|Long|Doulbe)：将流的结果约简为总和、平均值、最大值和最小值。该方法会接受一个将流对象为数据的函数，同时，这些方法会产生类型为(Int|Long|Double)SummaryStatistics的结果，同时计算总和、数量、平均值、最小值和最大值。

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

收集到映射表中

Collectors.toMap方法：方法有两个函数引元，用来产生映射表的键和值。

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

通常情况下，值应该是实际的元素，因此第二个元素可以使用Function.identity()

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

如果有多个元素具有相同的键，那么就会产生冲突，收集器将会抛出一个IllegalStateException对象。可以通过提供第3个引元来覆盖这种行为，该函数会针对给定的已有值和新值来解决冲突并确定键对应的值。这个函数应该反映已有键、新值或它们的组合。

构建一个映射表，存储了所有可用Locale中的每种语言，它在默认Locale中的名字为键，而其本地化的名字为值。

Stream<Locale> locales = Stream.of(Locale.getAvailableLocales());
Map<String, String> languageNames = locales.collect(
    Collectors.toMap(
        Local::getDisplayLanguage,
        l -> l.getDisplayLanguage(l),
        (existingVlaue, newValue) -> existingValue
    )
)

我们不关系同一种语言是否可能出现2此，因此我们只记录第一项

假设我们想要了解给定国家的所有语言，这样我们就需要一个Map<String, Set<String>>。

Map<String, Set<String>> countryLanguageSets = locales.collect(
    Collectors.toMap(
        Local::getDisplayCountry,
        l -> Collectors.singleton(l.getDisplayLanguage()),
        (a, b) -> {
            Set<String> union = new HashSet<>(a);
            union.addAll(b);
            return union;
        }
    )
)

如果想要得到TreeMap，那么可以将构造器作为第4个引元提供。

Map<Integer, Person> idToPerson = people.collect(
    Collectors.toMap(
        Person::getId,
        Function.getIdentity(),
        (existingValue, newValue) -> {throw new IllegalStateException()},
        TreeMap::new
    )
)

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

群组和分区

groupingBy方法：将具有相同特征的值群聚成组。

例如：通过国家来群组Locale

Map<String, List<Locale>> countryToLocales = 
    locales.collect(Collectors.groupingBy(Locale::getCountry));

当分类函数是断言函数时，流的元素可以分区为两个列表：该函数返回true的元素和其他的元素。在这种情况下，使用partitioningBy比使用groupingBy要更高效。

Map<Boolean, List<Locale>> enlishAndOtherLocales = locales.collect(
    Collectors.partitioningBy(l -> l.getLanguage().equals("en"))
);
List<Locale> enlishLocales = enlishAndOtherLocales.get(true);

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

下游收集器

groupingBy方法会产生一个映射表，它的每个值都是一个列表。如果想要处理这些列表，需要提供一个“下游收集器”。

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

Java提供了多种可以将群组元素约简为数字的收集器：

counting：会产生收集到元素的个数

Map<String, Long> countryToLocaleCounts = locales.collect(
    Collectors.groupingBy(Locale::getCountry, Collectors.counting())
);

summing(Int|Long|Double)：会接受一个函数引元，将该函数应用到下游元素中，并产生它们的和

Map<String, Integer> stateToCityPopulation= locales.collect( Collectors.groupingBy(City::getState, Collectors.summingInt(City::getPupulation)) );

maxBy和minBy会接受一个比较器，并产生下游元素中的最大值和最小值

Map<String, Optional<City>> stateToLargestCity = cities.collect(
    groupingBy(City::getState, 
               maxBy(Comparator.comparing(City::getPopulation)))
);

mapping：会产生将函数应用到下游结果上的收集器，并将函数值传递给另一个收集器。

Map<String, Optional<String>> stateToLongesCityName = cities.collect(
    groupingBy(City::getState,
              mapping(City::getName, 
                      maxBy(Comparator.comparing(String::length))))
)

将收集器组合起来是一种强大的方式，但是它也可能会导致产生非常复杂的表达式。它们的最佳用法是groupingBy和partitioningBy一起处理“下游的”映射表中的值。否则，应该直接在流上应用诸如map、reduce、count、max或min这样的方法。

约简操作

reduce：是一种用于从流中计算某个值的通用机制，用最简单的形式接受一个二元函数，并从前两个元素开始持续应用它。

如果该函数是求和函数，那么就容易解释这种机制：

List<Integer> values = ...;
Optional<Integer> sum = values.stream().reduce((x, y) -> x + y);

上面的reduce会计算v0 + v1 + v2 + ...,其中vi是流中的元素。如果流为空，那么该返回会返回一个Optional，因为没有任何有效的结果。

注意：上面可以写成reduce(Integer::sum)而不是reduce((x, y) -> x + y);

通常，如果reduce方法有一项约简操作op，那么该约简就会产生v0 op v1 op v2 op ...，其中我们将函数调用op(vi, vi+1)写作vi op vi+1。这项操作是可结合的：即组合元素时使用的顺序不应该成为问题。在数学标记法中，(x op y) op z 必须等于 x op (y op z)。这使得在使用并行流时，可以执行更高效的约简。

可结合操作：求和、乘积、字符串连接、取最大值和最小值、求集的并与交等。

通常，会有一个幺元值e使得 e op x = x，可以使用这个元素作为计算的起点。

0 是加法的幺元。如下，调用第二种reduce方法。

List<Integer> values = ...;
Integer sum = values.stream().reduce(0, (x, y) -> x + y);

如果流为空，则返回幺元值。

假设想要对某些属性求和，例如字符串中的所有字符串的长度。

需要提供一个“累加器”函数(total, word) -> total + word.length()。这个函数会反复调用，产生累计的总和。但是，当计算被并行化时，会有多个这种类型的计算，需要将它们的结果合并。因此，需要提供第二个函数来执行此处理。

int result = words.reduce(
    0,
    (x, y) -> x + y.length,
    (x1,x2) -> x1 + x2
);

注意：有时reduce会显得并不够通用。例如，假设我们想要收集BitSet中的结果。如果收集操作是并行的，那么就不能直接将元素放到单个BitSet中，因为BitSet对象不是线程安全的。因此，我们不能使用reduce，因为每个部分都需要以其自己的空集开始，并且reduce只能让我们提供一个幺元值。此时，应该使用collect，它会接受单个引元：

一个提供者，它会创建目标类型的新实例，例如散列集的构造器
一个累加器，它会将一个元素添加到一个实例上，例如add
一个组合其，它会将两个实例合并成一个，例如addAll

BitSet result = stream.collect(BitSet::new, BitSet::set, BitSet::or);

基本类型流

将每个整数都包装到包装器对象中是很低效的。

Java提供了基本类型流：IntStream、LongStream、DoubleStream;

想要存储short、char、byte和boolean，可以使用IntStream，对于float可以使用DoubleStream。

创建IntStream需要调用IntStream.of和Array.stream方法：

IntStream stream = IntStream.of(1,1,2,3,5);
stream = Arrays.stream(values, from, to);

与对象流一样，可以使用静态的generate和iterate方法。此外，IntStream和LongStream有静态的range和rangeClosed，可以生成步长为1的整数范围：

IntStream zeroToNintyNine = IntStream.range(0, 100);
IntStream zeroToHundred = IntStream.rangeClosed(0, 100);

CharSequence接口拥有codePoints和chars方法，可以生成由字符的Unicode码或由UTF-6编码机制的码元构成的IntStream。

String sentence = "\uD835\uDD46 is the set of octonions.";

IntStream codes = sentence.codePoints();

一个对象流可以用mapToInt、mapToLong和mapToDouble将其转换为基本类型流。

Stream<String> words = ...;
IntStream lengths = words.mapToInt(String::length);

基本类型流转换为对象流，可以使用boxed方法

Stream<Integer> integers = IntStream.range(0, 100).boxed();

基本类型流上的方法和对象流上的方法类似：

toArray方法会返回基本类型数组
产生可选结果的方法会返回一个OptionalInt、OptionalLong和OptionalDouble。这些类与Optional类类似，但是具有getAsInt、getAsLong、getAsDouble而不是get方法。
具有返回总和、平均数、最大值和最小值的sum、average、max和min方法。
summaryStatistics方法会产生一个类型为IntSummaryStatictis、LongSummaryStatistics、DoubleSummaryStatistics的对象，它们可以同时报告流的总和、平均值、最大值和最小值。

注意：Random类具有ints、longs和doubles方法，它们会返回由随机数构成的基本类型流。

并行流

Collection.parallelStream()：从任何集合中获取一个并行流。

parallel方法：可以将任意的顺序流转换为并行流。

只要在终结方法执行时，流处于并行模式，那么所有的中间流操作都将被并行化。
当流操作并行运行时，其目标是让其返回结果与顺序执行时返回的结果相同。重要的是，这些操作可以以任意顺序执行。

传递给并行流操作的函数不应该被阻塞。并行流使用fork-join池来操作流的各个部分。如果多个流操作被阻塞，那么池可能就无法做任何事情了。

默认情况下，从有序集合（数组和列表）、范围、生成器和迭代器产生的流，或者通过调用stream.ordered产生的流，都是有序的。

当放弃排序需求是，有些操作可以被更有效地并行化。通过调用unordered方法，可以明确表示我们对排序不感兴趣。可以提高distinct、limit。

Collectors.groupByConcurrent方法使用了共享的并发映射表。为了从并行流中获益，映射表中值的顺序不会与流中的顺序相同。

不要修改在执行某项流操作后会将元素返回流中的集合（即使这种修改是线程安全的）。记住，流并不会收集它们的数据，数据总是在单独的集合中。如果修改了这样的集合，那么流操作的结果就是未定义的。

并行流正常工作，需要满足大量的条件：

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

原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。

如有侵权，请联系 cloudcommunity@tencent.com 删除。

mapreduce

编程算法

原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。

如有侵权，请联系 cloudcommunity@tencent.com 删除。

mapreduce

编程算法

登录后参与评论

0 条评论

热度