• 如果您觉得本站非常有看点,那么赶紧使用Ctrl+D 收藏吧

JDK 8 新特性之函数式编程 → Stream API

开发技术 开发技术 3周前 (08-30) 19次浏览

开心一刻

  今天和朋友们去K歌,看着这群年轻人一个个唱的贼嗨,不禁感慨道:年轻真好啊!

  想到自己年轻的时候,那也是拿着麦克风不放的人

  现在的我没那激情了,只喜欢坐在角落里,默默的听着他们唱,就连旁边的妹子都劝我说:大哥别摸了,唱首歌吧

JDK 8 新特性之函数式编程 → Stream API

Stream 初体验

  很多时候,我们往往会选择在数据库层面进行数据的过滤、汇聚,这就导致我们对 JDK8 的 Stream 应用的特别少,对它也就特别陌生了

  但有时候,我们可以将原始数据加载到内存,在内存中进行数据的过滤和汇聚,这样可以减少数据库操作,提高查询效率(非绝对,数据量不大或走索引的情况下,数据库查询也是很快的)

  假设我们在内存中进行数据的过滤、汇聚,在 JDK8 之前(或不用 JDK8 的 Stream),我们会如何处理? 多次 for 循环结合 if ,并创建多个集合来存放中间结果,最后对中间结果进行汇聚,代码量会非常大;如果想牛逼一点,用多线程来处理,那就更复杂了(线程池、并发等问题)。Stream 就解决了这些痛点,如果你的 JDK 版本是 8(或更高),你还在用 for 循环进行数据的过滤和汇聚,那就有点这味了

JDK 8 新特性之函数式编程 → Stream API

  那 Stream 到底是何方神圣,让楼主如此推崇,我们往下看(再不讲重点,楼主怕是要收刀片了!)

  先闻其声

    我们先来看看她妈是怎么介绍她的: A sequence of elements supporting sequential and parallel aggregate operations. 

    我们能从中获取到两个信息:

      1、Stream 是元素的集合(有点类似 Iterator)

      2、对原 Stream 支持顺序或并行的汇聚操作

    这她妈的介绍还是比较抽象,我们需要从 Stream 自身下手,慢慢去了解她

    常见的 Stream 接口继承关系如下

JDK 8 新特性之函数式编程 → Stream API

    IntStream, LongStream, DoubleStream 对应的是三种基本类型(int, long, double,不是包装类型),Stream 对应所有剩余类型

    为什么不是这样

JDK 8 新特性之函数式编程 → Stream API

    或者取消掉 IntStream, LongStream, DoubleStream,由 Stream 对应所有类型 ?

    我们知道基本类型与包装类型之前的装箱与拆箱是有性能消耗的,频繁的转换会有比较严重的性能损耗,所以为不同数据类型设置不同stream接口,可以提高性能,也可以增加特定接口

  一睹芳容

    上面说了那么多,却始终未一睹 Stream 的芳容,心里着急呀!我们先来瞟一眼

List<Integer> nums = Arrays.asList(1, 8, 0, 5, 3, 2);
// 统计大于 3 的元素个数
long count = nums.stream().filter(e -> e > 3).count();

    是不是很美?千万不要以为 Stream 就这?这还只是她的一条腿,她浑身上下都是宝

JDK 8 新特性之函数式编程 → Stream API

    通过上面的简单示例,我们可以剖析出 Stream 的通用语法

JDK 8 新特性之函数式编程 → Stream API

    也就是说使用 Stream 基本分三步:创建 Steam、转换Stream、汇聚,下面我们就从这三步详细介绍 Stream

创建 Stream

  Stream 的创建方式有很多,我们只讲最常用的两种

  基于数组: Stream<String> arrayStream = Arrays.stream(new String[]{“123”, “abc”, “A”, “张三”}); 

  基于 Collection: Stream<String> collectionStream = Arrays.asList(“123”, “abc”, “A”, “张三”).stream(); 

  把数组变成 Stream 使用 Arrays.stream() 方法;对于 Collection(List、Set、Queue 等),直接调用 stream() 方法就可以获得 Stream

转换 Stream

  转换 Stream 的目的是对原 Stream 做出某种程度的数据映射/过滤,然后返回一个新的流,交给下一个操作使用,对原 Stream 是没有任何影响的。这类操作都是惰性化的(lazy),就是说,仅仅调用到这类方法,并没有真正开始流的遍历

  由于获取的是一个新的流,而不是我们需要的最终结果,所以 转换 Stream 这个操作有个官方的称呼: Intermediate ,即中间操作

  具体的转换操作有很多,我们挑一些常用的来说明一下

  distinct

    对 Stream 中的元素进行去重操作(去重逻辑依赖元素的 equals 方法),新生成的 Stream 中没有重复的元素

List<Integer> nums = Arrays.asList(1, 2, 1, 3, 2, 5);
Stream<Integer> distinctStream = nums.stream().distinct();

JDK 8 新特性之函数式编程 → Stream API

  filter

    对 Stream 中的每个元素使用给定的过滤条件进行过滤操作,新生成的 Stream 只包含符合条件的元素

List<Integer> nums = Arrays.asList(1, 2, 1, 3, 2, 5);
Stream<Integer> filterStream = nums.stream().filter(e -> e >= 2);

JDK 8 新特性之函数式编程 → Stream API

  map

    对 Stream 中的每个元素按给定的转换规则进行转换操作,新生成的 Stream 只包含转换生成的元素

List<Integer> nums = Arrays.asList(1, 2, 1, 3, 2, 5);
Stream<String> mapStream = nums.stream().map(e -> e * e + "");

JDK 8 新特性之函数式编程 → Stream API

    JDK1.8 还提供了三个专门针对基本数据类型的 map 变种方法:mapToInt,mapToLong 和 mapToDouble。这三个方法也比较好理解,就是把原始 Stream 转换成一个新的 Stream,这个新生成的 Stream 中的元素都是对应的基本类型。之所以会有这三个变种方法,是考虑到自动装箱/拆箱的额外消耗

List<Integer> nums = Arrays.asList(1, 2, 1, 3, 2, 5);
IntStream intStream = nums.stream().mapToInt(e -> e * 2);
LongStream longStream = nums.stream().mapToLong(e -> 3L * e);
DoubleStream doubleStream = nums.stream().mapToDouble(e -> 3.0 * e);

  flatMap

    与 map 类似

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

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

    不同的是 flatMap 中每个元素转换得到的是 Stream 对象,然后会把子 Stream 中的元素都放到新的 Stream 中

List<List<String>> groupList = Arrays.asList(Arrays.asList("q","w","e"), Arrays.asList("a", "s", "d"), Arrays.asList("z","x", "c"));
Stream<String> superStarStream = groupList.stream().flatMap(group -> group.stream().map(e -> e + 1));

JDK 8 新特性之函数式编程 → Stream API

    简单点理解就是:把几个小的集合中的元素经过处理后合并到一个大的集合中

    类似的,JDK1.8 也提供了三个专门针对基本数据类型的 flatMap 变种方法:flatMapToInt,flatMapToLong 和 flatMapToDouble

  limit

    拷贝原 Stream 中的前 N 个元素到新的 Stream 中,如果原 Stream 中包含的元素个数小于 N,那就获取其所有的元素

List<Integer> nums = Arrays.asList(1, 2, 1, 3, 2, 5);
Stream<Integer> limitStream = nums.stream().limit(4);

JDK 8 新特性之函数式编程 → Stream API

  skip

    拷贝原 Stream 除了前 N 个元素后剩下的所有元素到新 Stream,如果原 Stream 中包含的元素个数小于 N,那么返回空 Stream

List<Integer> nums = Arrays.asList(1, 2, 1, 3, 2, 5);
Stream<Integer> skipStream = nums.stream().skip(4);

JDK 8 新特性之函数式编程 → Stream API

  sorted

    对原 Stream 进行排序操作,得到一个新的、有序的 Stream

    排序函数有两个,一个是用自然顺序排序,一个是使用自定义比较器排序

Stream<T> sorted();
Stream<T> sorted(Comparator<? super T> comparator);

    使用起来非常简单,如下所示

List<Integer> nums = Arrays.asList(1, 2, 1, 3, 2, 5);
// 自然排序,默认升序排序
Stream<Integer> sortedStream = nums.stream().sorted();
// 自定义排序
Stream<Integer> sortedCompareStream = nums.stream().sorted((a, b) -> b.compareTo(a));

  peek

Stream<T> peek(Consumer<? super T> action);

    生成一个包含原 Stream 所有元素的新 Stream,同时会提供一个消费函数(Consumer 实例),新 Stream 每个元素被消费的时候都会执行给定的消费函数

    与 map 很像,但不会影响新 Stream 中的元素(还是原 Stream 中的元素),可以做一些输出,外部处理等辅助操作

    这个在实际项目中用的不多,知道是怎么回事就好

汇聚

  汇聚操作接受一个 Stream 为输入,反复使用某个汇聚操作,把 Stream 中的元素合并成一个汇总的结果,汇总结果可能是某个值,也可能是一个集合

  汇聚操作能够得到我们需要的最终结果,相当于一个终止操作,所以也有另一个称呼: Terminal ,即结束操作

  一个流只能有一个 terminal 操作,当这个操作执行后,流就被使用“光”了,无法再被操作。所以这必定是流的最后一个操作。Terminal 操作的执行,才会真正开始流的遍历

  JDK1.8 提供了很多常用的汇聚操作,我们一起来看看

  foreach

    这个类似我们平时的 for 循环,遍历 Stream 中的元素,执行指定的操作

List<Integer> nums = Arrays.asList(1, 2, 1, 3, 2, 5);
nums.stream().forEach(num -> System.out.println(num));

  max min count

    作用就是字面意思

List<Integer> nums = Arrays.asList(1, 2, 1, 3, 2, 5);
// 求最大值
Integer max = nums.stream().max(Comparator.naturalOrder()).get();
// 求最小值
Integer min = nums.stream().min(Comparator.naturalOrder()).get();
// 求元素个数
long count = nums.stream().count();
System.out.println("max = " + max);
System.out.println("min = " + min);
System.out.println("count = " + count);

  findFirst

    返回一个 Optional,它包含了 Stream 中的第一个元素,若 Stream 是空的,则返回一个空的 Optional

List<Integer> nums = Arrays.asList(1, 2, 1, 3, 2, 5);
Integer firstNum = nums.stream().findFirst().get();

  findAny

    返回一个 Optional,它包含了 Stream 中的任意一个元素,若 Stream 是空的,则返回一个空的 Optional

List<Integer> nums = Arrays.asList(1, 2, 1, 3, 2, 5);
Integer anyNum = nums.stream().findAny().get();

    在串行的流中,findAny 和 findFirst返回的,都是第一个对象;而在并行的流中,findAny 返回的是最快处理完的那个线程的数据,所以说,在并行操作中,对数据没有顺序上的要求,那么 findAny 的效率会比 findFirst 要快的,但是没有 findFirst 稳定

  anyMatch

    Stream 中是否有任意一个元素满足判断条件,有则返回 true

List<Integer> nums = Arrays.asList(1, 2, 1, 3, 2, 5);
boolean matchResult = nums.stream().anyMatch(num -> num > 2);

  allMatch

    Stream 中所有元素都满足判断条件则返回 true

List<Integer> nums = Arrays.asList(1, 2, 1, 3, 2, 5);
boolean matchResult = nums.stream().allMatch(num -> num > 2);

  noneMatch

    与 allMatch 相反,Stream 中所有元素都不满足判断条件,则返回 true

List<Integer> nums = Arrays.asList(1, 2, 1, 3, 2, 5);
boolean matchResult = nums.stream().noneMatch(num -> num > 7);

  reduce

    reduce 的主要作用是把 Stream 元素组合起来

    它提供一个起始值(种子),然后依照运算规则(BinaryOperator),和 Stream 中的第一个、第二个、第 n 个元素组合,生成一个我们需要的值

    JDK 提供了三种 reduce

T reduce(T identity, BinaryOperator<T> accumulator);

Optional<T> reduce(BinaryOperator<T> accumulator);

<U> U reduce(U identity,
                 BiFunction<U, ? super T, U> accumulator,
                 BinaryOperator<U> combiner);

    参数不同,其返回值类型是有所不同的,但其语义、作用还是一样的

    max()、min()其实都是特殊的 reduce,只是因为它们比较常用,所以就简化书写专门设计出了它们

@Override
public final Optional<P_OUT> max(Comparator<? super P_OUT> comparator) {
    return reduce(BinaryOperator.maxBy(comparator));
}

@Override
public final Optional<P_OUT> min(Comparator<? super P_OUT> comparator) {
    return reduce(BinaryOperator.minBy(comparator));

}

    reduce 在实际项目中用的不多,又非常灵活,我们就简单看几个示例

JDK 8 新特性之函数式编程 → Stream APIJDK 8 新特性之函数式编程 → Stream API

        List<Integer> nums = Arrays.asList(1, 2, 1, 3, 2, 5);

        // 求和,相当于sum(); 有起始值
        Integer sum1 = nums.stream().reduce(0, Integer::sum);
        Integer sum2 = nums.stream().reduce(0, (a,b) -> a + b);
        // 求和,相当于sum(); 无起始值
        Integer sum3 = nums.stream().reduce(Integer::sum).get();
        System.out.println("sum = " + sum1 + ", sum2 = " + sum2 + ", sum3 = " + sum3);

        // 求最大值,相当于max()
        Integer max = nums.stream().reduce(Integer.MIN_VALUE, Integer::max);
        System.out.println("max = " + max);
        // 求最小值,相当于min()
        Integer min = nums.stream().reduce(Integer.MAX_VALUE, Integer::min);
        System.out.println("min = " + min);

View Code

    reduce 擅长的是生成一个值,如果想要从 Stream 生成一个集合或者 Map 等复杂的对象该怎么办呢?就需要 collect 出马了

  collect

    collect 是 Stream 接口中最灵活的,也是最强大的;JDK 中提供了两种 collect

// Supplier supplier是一个工厂函数,用来生成一个新的容器;
// BiConsumer accumulator也是一个函数,用来把Stream中的元素添加到结果容器中
// BiConsumer combiner还是一个函数,用来把中间状态的多个结果容器合并成为一个(并发的时候会用到)
<R> R collect(Supplier<R> supplier,
              BiConsumer<R, ? super T> accumulator,
              BiConsumer<R, R> combiner);

<R, A> R collect(Collector<? super T, A, R> collector);

    我们来各看一个案例

List<String> strList = Arrays.asList("123","abc", "1w1");
String concat  = strList.stream().collect(StringBuilder::new, StringBuilder::append, StringBuilder::append).toString();
System.out.println(concat);

List<String> stringList = strList.stream().collect(Collectors.toList());

     实际应用中,基本上用的是第二种,而且用的是 JDK 中已经提供好的 Collector,在 Collectors 中提供了很多常用的 Collector, 如下

JDK 8 新特性之函数式编程 → Stream API

    我们挑一些比较常用的来说明下,有兴趣的可以去通读下

    转集合

      toList、toSet、toMap

JDK 8 新特性之函数式编程 → Stream APIJDK 8 新特性之函数式编程 → Stream API

public class StreamTest {

    public static void main(String[] args) {
        Person[] personArray = {
                new Person("shangsan", 23), new Person("张三", 23),
                new Person("lisi", 24), new Person("李四", 24),
                new Person("wangwu", 20), new Person("王五", 20)};
        //list
        List<Person> personList = Arrays.stream(personArray).collect(Collectors.toList());
        // 转 set
        Set<Person> personSet = Arrays.stream(personArray).collect(Collectors.toSet());
        // 转 map, key为 name, value 为 Person 实例
        Map<String, Person> personMap = Arrays.stream(personArray).collect(Collectors.toMap(Person::getName, person -> person));

        System.out.println(personList);
        System.out.println(personSet);
        System.out.println(personMap);
    }

    static class Person {
        private String name;
        private Integer age;

        Person(){}

        Person(String name, Integer age) {
            this.name = name;
            this.age = age;
        }

        public String getName() {
            return name;
        }

        public void setName(String name) {
            this.name = name;
        }

        public Integer getAge() {
            return age;
        }

        public void setAge(Integer age) {
            this.age = age;
        }

        @Override
        public String toString() {
            return "Person{" +
                    "name='" + name + ''' +
                    ", age=" + age +
                    '}';
        }
    }
}

View Code

      toMap 有两个注意点

        1、底层调用的是 map.merge 方法,该方法遇到 value 为 null 的情况会报 npe

        2、遇到重复的 key 会直接抛 IllegalStateException,因为未指定冲突合并策略,也就是第三个参数BinaryOperator<U> mergeFunction

    分组

JDK 8 新特性之函数式编程 → Stream APIJDK 8 新特性之函数式编程 → Stream API

public class StreamTest {

    public static void main(String[] args) {
        Person[] personArray = {
                new Person("shangsan", 23), new Person("张三", 23),
                new Person("lisi", 24), new Person("李四", 24),
                new Person("wangwu", 20), new Person("王五", 20)};

        // 根据年龄进行分组
        Map<Integer, List<Person>> ageGroup = Arrays.stream(personArray).collect(Collectors.groupingBy(Person::getAge));
        System.out.println(ageGroup);
    }

    static class Person {
        private String name;
        private Integer age;

        Person(){}

        Person(String name, Integer age) {
            this.name = name;
            this.age = age;
        }

        public String getName() {
            return name;
        }

        public void setName(String name) {
            this.name = name;
        }

        public Integer getAge() {
            return age;
        }

        public void setAge(Integer age) {
            this.age = age;
        }

        @Override
        public String toString() {
            return "Person{" +
                    "name='" + name + ''' +
                    ", age=" + age +
                    '}';
        }
    }
}

View Code

    求和

// 年龄求和 summingInt、summingLong、summingDouble 类似
Integer ageSum = Arrays.stream(personArray).collect(Collectors.summingInt(Person::getAge));
System.out.println("age sum = " + ageSum);

    求平均值

// 求平均值 averagingInt、averagingLong、averagingDouble 类似
Double averageAge = Arrays.stream(personArray).collect(Collectors.averagingInt(Person::getAge));
System.out.println("average age = " + averageAge);

    其他

// 统计人数
Long count = Arrays.stream(personArray).collect(Collectors.counting());
System.out.println("人数 = " + count);

List<Integer> intList = Arrays.asList(1, 2, 3, 1, 5, 2);
// 求和
Integer sum = intList.stream().collect(Collectors.reducing(0, (a, b) -> a + b));
System.out.println("sum = " + sum);

// 字符串拼接
List<String> strList = Arrays.asList("123", "abc", "666");
String str = strList.stream().collect(Collectors.joining(",", "(", ")"));
System.out.println(str);

并行流

  前面讲了那么多,都是基于顺序流(Stream),JDK1.8 也提供了并行流: parallelStream ,使用起来非常简单,通过 parallelStream() 可能创建并行流,流的操作还是和顺序流一样

List<Integer> intList = Arrays.asList(1, 2, 3, 1, 5, 2);
boolean result = intList.parallelStream().anyMatch(e -> e > 5);
System.out.println("result = " + result);

  顾名思义,并行流可以运用多核特性(forkAndJoin)进行并行处理,从而大幅提高效率,既然能提高效率,为什么实际项目中,顺序流用的更多,而并行流用的非常少了,还是有一些原因的

  1、parallelStream 是线程不安全的

    一旦出现并发问题,大家都懂的,非常头疼

  2、parallelStream 适用于 CPU 密集型任务

    如果 CPU 负载已经很大,还用并行流,不但不会提高效率,反而会降低效率

    并行流不适用于 I/O 密集型任务,很可能会造成 I/O 阻塞

  3、并行流无法保证元素顺序,输出结果具有不确定性

    如果我们的业务需要关注元素先后顺序,那么不能用并行流

  4、lambda 的执行并不是瞬间完成的,所有使用 parallel stream 的程序都有可能成为阻塞程序的源头

总结

  Stream 特点

    无存储:Stream 不是数据结构并不保存数据,它是有关算法和计算的,它只是某种数据源的一个视图,数据源可以是一个数组,Java 容器或 I/O channel

    函数式编程:每次转换,原有 Stream 不改变,返回一个新的 Stream 对象,这就允许对其操作可以像链条一样排列;转换过程可以多次

    惰性执行:Stream 上的转换操作(中间操作)并不会立即执行,只有执行汇聚操作(终止操作)时,转换操作才会执行

    一次消费:Stream 只能被使用一次,一旦遍历过就会失效,就像容器的迭代器那样,想要再次遍历必须重新生成

  Stream 优点

    代码简洁且易理解,这个感受是最明显的,用与不用 Stream,代码量与可阅读性相差甚远

    多核友好,如果想多线程处理,只需要调一下 parallel() 方法,仅此而已

  Stream 操作分类

    分两类:中间操作(Intermediate)、结束操作(Terminal)

    中间操作总是会惰式执行,调用中间操作只会生成一个标记了该操作的新 stream,仅此而已

    结束操作会触发实际计算,计算发生时会把所有中间操作积攒的操作以 pipeline 的方式执行,这样可以减少迭代次数;计算完成之后stream就会失效

JDK 8 新特性之函数式编程 → Stream API

  性能问题

    在对于一个 Stream 进行多次转换操作 (Intermediate 操作),每次都对 Stream 的每个元素进行转换,而且是执行多次,这样时间复杂度就是 N(转换次数)个 for 循环里把所有操作都做掉的总和吗?其实不是这样的,转换操作都是 lazy 的,多个转换操作只会在 Terminal 操作的时候融合起来,一次循环完成。我们可以这样简单的理解,Stream 里有个操作函数的集合,每次转换操作就是把转换函数放入这个集合中,在 Terminal 操作的时候循环 Stream 对应的集合,然后对每个元素执行所有的函数

    关于顺序流、并行流、传统 for 的效率问题,大家看看这个:Java Stream API性能测试、for循环与串行化、并行化Stream流性能对比

参考

  Java 8 中的 Streams API 详解

  JDK8函数式编程之Stream API


喜欢 (0)