Stream流编程——Webflux响应式编程利器

Posted 2021-02-05 algerfan

Stream流编程

1. 是什么，不是什么

是一个高级的迭代器，不是一个数据结构、不是一个集合、不会存放数据、关注的是怎么把数据高效处理

2. 创建/中间操作/终止操作

1） 创建

代码演示

        List<String> list = new ArrayList<>();

        // 从集合创建
        list.stream();
        list.parallelStream();

        // 从数组创建
        Arrays.stream(new int[] { 2, 3, 5 });

        // 创建数字流
        IntStream.of(1, 2, 3);
        IntStream.rangeClosed(1, 10);

        // 使用random创建一个无限流
        new Random().ints().limit(10);
        Random random = new Random();

        // 自己产生流
        Stream.generate(() -> random.nextInt()).limit(20);

2） 中间操作

        String str = "my name is AlgerFan";

        System.out.println("--------------filter------------");
        // 把每个单词的长度调用出来
        Stream.of(str.split(" ")).filter(s -> s.length() > 2)
                .map(String::length).forEach(System.out::println);

        System.out.println("--------------flatMap------------");
        // flatMap A->B属性(是个集合), 最终得到所有的A元素里面的所有B属性集合
        // intStream/longStream 并不是Stream的子类, 所以要进行装箱 boxed
        Stream.of(str.split(" ")).flatMap(s -> s.chars().boxed())
                .forEach(i -> System.out.println((char) i.intValue()));

        System.out.println("--------------peek------------");
        // peek 用于debug. 是个中间操作,和 forEach 是终止操作
        Stream.of(str.split(" ")).peek(System.out::println)
                .forEach(System.out::println);

        System.out.println("--------------limit------------");
        // limit 使用, 主要用于无限流
        new Random().ints().filter(i -> i > 100 && i < 1000).limit(5)
                .forEach(System.out::println);

3） 终止操作

        String str = "my name is AlgerFan";

        System.out.println("-------并行流parallel--------");
        // 使用并行流
        str.chars().parallel().forEach(i -> System.out.print((char) i));
        System.out.println();
        // 使用 forEachOrdered 保证顺序
        str.chars().parallel().forEachOrdered(i -> System.out.print((char) i));
        System.out.println();

        System.out.println("-------collect收集到list--------");
        // 收集到list
        List<String> list = Stream.of(str.split(" "))
                .collect(Collectors.toList());
        System.out.println(list);

        System.out.println("-------使用 reduce 拼接字符串--------");
        // 使用 reduce 拼接字符串
        Optional<String> letters = Stream.of(str.split(" "))
                .reduce((s1, s2) -> s1 + "|" + s2);
        System.out.println(letters.orElse(""));

        System.out.println("-------带初始化值的reduce--------");
        // 带初始化值的reduce
        String reduce = Stream.of(str.split(" ")).reduce("",
                (s1, s2) -> s1 + "|" + s2);
        System.out.println(reduce);

        System.out.println("-------计算所有单词总长度--------");
        // 计算所有单词总长度
        Integer length = Stream.of(str.split(" ")).map(s -> s.length())
                .reduce(0, (s1, s2) -> s1 + s2);
        System.out.println(length);

        System.out.println("-------max 的使用--------");
        // max 的使用
        Optional<String> max = Stream.of(str.split(" "))
                .max((s1, s2) -> s1.length() - s2.length());
        System.out.println(max.get());

        System.out.println("-------使用 findFirst 短路操作--------");
        // 使用 findFirst 短路操作
        OptionalInt findFirst = new Random().ints().findFirst();
        System.out.println(findFirst.getAsInt());

3. 并行流

以上已经接触了parallel()并行流，能够多线程的处理数据

4. 收集器

示例代码：

        // 测试数据
        List<Student> students = Arrays.asList(
                new Student("小明", 10, Gender.MALE, Grade.ONE),
                new Student("大明", 9, Gender.MALE, Grade.THREE),
                new Student("小白", 8, Gender.FEMALE, Grade.TWO),
                new Student("小黑", 13, Gender.FEMALE, Grade.FOUR),
                new Student("小红", 7, Gender.FEMALE, Grade.THREE),
                new Student("小黄", 13, Gender.MALE, Grade.ONE),
                new Student("小青", 13, Gender.FEMALE, Grade.THREE),
                new Student("小紫", 9, Gender.FEMALE, Grade.TWO),
                new Student("小王", 6, Gender.MALE, Grade.ONE),
                new Student("小李", 6, Gender.MALE, Grade.ONE),
                new Student("小马", 14, Gender.FEMALE, Grade.FOUR),
                new Student("小刘", 13, Gender.MALE, Grade.FOUR));

        // 得到所有学生的年龄列表
        // s -> s.getAge() --> Student::getAge , 不会多生成一个类似 lambda$0这样的函数
        Set<Integer> ages = students.stream().map(Student::getAge)
                .collect(Collectors.toCollection(TreeSet::new));
        System.out.println("所有学生的年龄:" + ages);

        // 统计汇总信息
        IntSummaryStatistics agesSummaryStatistics = students.stream()
                .collect(Collectors.summarizingInt(Student::getAge));
        System.out.println("年龄汇总信息:" + agesSummaryStatistics);

        // 分块
        Map<Boolean, List<Student>> genders = students.stream().collect(
                Collectors.partitioningBy(s -> s.getGender() == Gender.MALE));
        System.out.println("男女学生列表:" + genders);

        // 分组
        Map<Grade, List<Student>> grades = students.stream()
                .collect(Collectors.groupingBy(Student::getGrade));
        System.out.println("学生班级列表:" + grades);

        // 得到所有班级学生的个数
        Map<Grade, Long> gradesCount = students.stream().collect(Collectors
                .groupingBy(Student::getGrade, Collectors.counting()));
        System.out.println("班级学生个数列表:" + gradesCount);

测试结果

所有学生的年龄:[6, 7, 8, 9, 10, 13, 14]
年龄汇总信息:IntSummaryStatistics{count=12, sum=121, min=6, average=10.083333, max=14}
男女学生列表:{false=[[name=小白, age=8, gender=FEMALE, grade=TWO], [name=小黑, age=13, gender=FEMALE, grade=FOUR], [name=小红, age=7, gender=FEMALE, grade=THREE], [name=小青, age=13, gender=FEMALE, grade=THREE], [name=小紫, age=9, gender=FEMALE, grade=TWO], [name=小马, age=14, gender=FEMALE, grade=FOUR]], true=[[name=小明, age=10, gender=MALE, grade=ONE], [name=大明, age=9, gender=MALE, grade=THREE], [name=小黄, age=13, gender=MALE, grade=ONE], [name=小王, age=6, gender=MALE, grade=ONE], [name=小李, age=6, gender=MALE, grade=ONE], [name=小刘, age=13, gender=MALE, grade=FOUR]]}
学生班级列表:{FOUR=[[name=小黑, age=13, gender=FEMALE, grade=FOUR], [name=小马, age=14, gender=FEMALE, grade=FOUR], [name=小刘, age=13, gender=MALE, grade=FOUR]], ONE=[[name=小明, age=10, gender=MALE, grade=ONE], [name=小黄, age=13, gender=MALE, grade=ONE], [name=小王, age=6, gender=MALE, grade=ONE], [name=小李, age=6, gender=MALE, grade=ONE]], THREE=[[name=大明, age=9, gender=MALE, grade=THREE], [name=小红, age=7, gender=FEMALE, grade=THREE], [name=小青, age=13, gender=FEMALE, grade=THREE]], TWO=[[name=小白, age=8, gender=FEMALE, grade=TWO], [name=小紫, age=9, gender=FEMALE, grade=TWO]]}
班级学生个数列表:{FOUR=3, ONE=4, THREE=3, TWO=2}

5. 运行机制

演示一个测试代码

        Random random = new Random();
        // 随机产生数据
        Stream<Integer> stream = Stream.generate(random::nextInt)
                // 产生300个 ( 无限流需要短路操作. )
                .limit(300)
                // 第1个无状态操作，print(s)执行耗时操作5s
                .peek(s -> print("peek: " + s))
                // 第2个无状态操作
                .filter(s -> {
                    print("filter: " + s);
                    return s > 1000000;
                })
                // 有状态操作
                /*.sorted((i1, i2) -> {
                    print("排序: " + i1 + ", " + i2);
                    return i1.compareTo(i2);
                })*/
                // 又一个无状态操作
                .peek(s -> {
                    print("peek2: " + s);
                });

        // 终止操作
        stream.count();

分析以上代码，发现Stream创建了一个256长度的数组

1. 所有操作是链式调用, 一个元素只迭代一次
2. 每一个中间操作返回一个新的流. 流里面有一个属性sourceStage
   指向同一个 地方,就是Head
3. Head->nextStage->nextStage->... -> null
4. 有状态操作会把无状态操作阶段,单独处理
5. 并行环境下, 有状态的中间操作不一定能并行操作.
6. parallel/ sequetial 这2个操作也是中间操作(也是返回stream)
   但是他们不创建流, 他们只修改 Head的并行标志