关于程序员:Java8中的流处理Stream-API

45次阅读

共计 6499 个字符，预计需要花费 17 分钟才能阅读完成。

1. 是什么

Stream 是 Java8 中解决汇合的要害抽象概念，它能够指定你心愿对
汇合进行的操作，能够执行非常复杂的查找、过滤和映射数据等操作。
应用 Stream API 对汇合数据进行操作，就相似于应用 SQL 执行的数
据库查问。也能够应用 Stream API 来并行执行操作。简而言之，
Stream API 提供了一种高效且易于应用的解决数据的形式。

留神：

  • Stream 本人不会存储元素。
  • Stream 不会扭转源对象。相同，他们会返回一个持有后果的新 Stream。
  • Stream 操作是提早执行的。这意味着他们会等到须要后果的时候才执行。

2. Stream 的应用

三步骤

  • 创立 Stream
    一个数据源（如：汇合、数组），获取一个流
  • 两头操作

一个两头操作链，对数据源的数据进行解决

  • 终止操作(终端操作)

一个终止操作，执行两头操作链，并产生后果

2.1 创立 Stream

  • Collection 提供了两个办法 stream() 与 parallelStream()
List<String> list = new ArrayList<>();
Stream<String> stream = list.stream(); // 获取一个程序流
Stream<String> parallelStream = list.parallelStream(); // 获取一个并行流
  • Java8 中的 Arrays 的静态方法 stream() 能够获取数组流
Integer[] nums = new Integer[10];
Stream<Integer> stream1 = Arrays.stream(nums);

#还有很多重载办法也能够取得流
重载模式，可能解决对应根本类型的数组：public static IntStream stream(int[] array)
 public static LongStream stream(long[] array)
 public static DoubleStream stream(double[] array)
  • 通过 concat 办法让多个流合并
 List<Cart> carts = queryCartList();
        List<Cart> carts2 =  queryCartList();
        Stream<Cart> concat = Stream.concat(carts.stream(), carts2.stream());
    #而后能够接着操作流。
  • 通过 Stream 类中静态方法 of()
Stream<Integer> stream2 = Stream.of(1,2,3,4,5,6);
  • 由函数创立流：创立有限流

能够应用静态方法 Stream.iterate() 和 Stream.generate(), 创立有限流。

// 迭代
Stream<Integer> stream3 = Stream.iterate(0, (x) -> x + 2).limit(10);
stream3.forEach(System.out::println);

// 生成
Stream<Double> stream4 = Stream.generate(Math::random).limit(2);
stream4.forEach(System.out::println);

2.2 两头操作

tips：当没应用显式的应用 Iterator 进行遍历时，称为外部迭代，应用 Strea 外部实现。
多个两头操作能够连接起来造成一个流水线，除非流水线上触发终止操作，否则两头操作不会执行任何的解决！而在终止操作时一次性全副解决，称为“惰性求值”.

2.2.1 筛选与切片
  • filter(Predicate p) —— 接管 Lambda，从流中排除某些元素。
Stream<Employee> stream = emps.stream()
            .filter((e) -> {System.out.println("测试两头操作");
                return e.getAge() <= 35;});
  • limit—— 截断流，使其元素不超过给定数量。
# 相当于短路，当达到数量后就不再进行其余操作
emps.stream()
    .filter((e) -> {System.out.println("短路！"); // &&  ||
        return e.getSalary() >= 5000;}).limit(3)
    .forEach(System.out::println);
  • skip(n) —— 跳过元素，返回一个扔掉了前 n 个元素的流。若流中元素有余 n 个，则返回一个空流。与 limit(n) 互补
emps.parallelStream()
            .filter((e) -> e.getSalary() >= 5000)
            .skip(2)
            .forEach(System.out::println);
  • distinct—— 筛选，通过流所生成元素的 hashCode() 和 equals() 去除反复元素
emps.stream()
        .distinct()
        .forEach(System.out::println);
2.2.2 映射
  • map——接管 Lambda，将元素转换成其余模式或提取信息。接管一个函数作为参数，该函数会被利用到每个元素上，并将其映射成一个新的元素。
  List<String> strList = Arrays.asList("aaa", "bbb", "ccc", "ddd");        
  Stream<String> stream = strList.stream()
               .map(String::toUpperCase);

  stream.forEach(System.out::println);
  • flatMap——接管一个函数作为参数，将流中的每个值都换成另一个流，而后把所有流连接成一个流。
# 这里给出一个案例，需要是把字符串汇合的每个字符离开变成单个字节而后输入。map 实现的是每个字符串生成流，而后返回流的汇合。而 flatMap 是把所有的流中的元素放在一个流中。通过函数的返回值能够清晰的看进去区别。public void test1(){List<String> strList = Arrays.asList("aaa", "bbb", "ccc", "ddd", "eee");
        Stream<Stream<Character>> stream2 = strList.stream()
//               .map(TestStreamAPI1::filterCharacter);
                .map((e) -> filterCharacter(e));
        stream2.forEach((sm) -> {sm.forEach(System.out::println);
        });
        
        System.out.println("-------------------------------------");
        
        Stream<Character> stream3 = strList.stream()
               .flatMap(TestStreamAPI1::filterCharacter);

        stream3.forEach(System.out::println);
    }

    public static Stream<Character> filterCharacter(String str){List<Character> list = new ArrayList<>();
        
        for (Character ch : str.toCharArray()) {list.add(ch);
        }
        
        return list.stream();}
  • 这两个还有很多子类型:

mapToDouble(ToDoubleFunction f) 接管一个函数作为参数，该函数会被利用到每个元素上，产生一个新的 DoubleStream。
mapToInt(ToIntFunction f)
mapToLong(ToIntFunction f)
flatMap 也有同样的三种办法。这里不再细说。

2.2.3 排序
  • sorted()——天然排序
emps.stream()
        .map(Employee::getName)
        .sorted()
        .forEach(System.out::println);
  • sorted(Comparator com)——定制排序
emps.stream()
        .sorted((x, y) -> {if(x.getAge() == y.getAge()){return x.getName().compareTo(y.getName());
            }else{return Integer.compare(x.getAge(), y.getAge());
            }
        }).forEach(System.out::println);
2.2.4 range 操作
  • range 和 rangeClosed

range 不包含范畴中的前面的数，而 rangeClosed 包含前面的数。

public class RangeTest {public static void main(String[] args) {IntStream.range(0,10).forEach(System.out::print);

        System.out.println("==============");
        IntStream.rangeClosed(0,10).forEach(System.out::print);
    }
}

// 输入：0123456789==============
//012345678910

2.3 终止操作

留神：流进行了终止操作后，不能再次应用
** 终止操作会从流的流水线生成后果。其后果能够是任何不是流的
值，例如：List、Integer，甚至是 void。**

2.3.1 查找与匹配
  • allMatch——查看是否匹配所有元素
       boolean bl = emps.stream()
                .allMatch((e) -> e.getStatus().equals(Status.BUSY));

        System.out.println(bl);
  • anyMatch——查看是否至多匹配一个元素
        boolean bl1 = emps.stream()
                .anyMatch((e) -> e.getStatus().equals(Status.BUSY));

        System.out.println(bl1);
  • noneMatch——查看是否没有匹配的元素
        boolean bl2 = emps.stream()
                .noneMatch((e) -> e.getStatus().equals(Status.BUSY));

        System.out.println(bl2);
  • findFirst——返回第一个元素
 Optional<Employee> op = emps.stream()
               .sorted((e1, e2) -> 
                        Double.compare(e1.getSalary(), e2.getSalary()))
                .findFirst();

        System.out.println(op.get());
  • findAny——返回以后流中的任意元素
        Optional<Employee> op2 = emps.parallelStream()
                .filter((e) -> e.getStatus().equals(Status.FREE))
                .findAny();

        System.out.println(op2.get());
  • count——返回流中元素的总个数
        long count = emps.stream()
                .filter((e) -> e.getStatus().equals(Status.FREE))
                .count();

        System.out.println(count);
  • max——返回流中最大值
 Optional<Double> op = emps.stream()
                .map(Employee::getSalary)
                .max(Double::compare);
  • min——返回流中最小值
 Optional<Employee> op2 = emps.stream()
                .min((e1, e2) ->
                       Double.compare(e1.getSalary(), e2.getSalary()));
2.3.2 规约
  • reduce(T identity, BinaryOperator) ` 能够将流中元素重复联合起来，失去一个值。

返回 T`

  • reduce(BinaryOperator) ` 能够将流中元素重复联合起来，失去一个值。

返回 Optional<T>`

List<Integer> list = Arrays.asList(1,2,3,4,5,6,7,8,9,10);
        #形式一，会把这个 o 当做初始的 x 而后进行操作。Integer sum = list.stream()
            .reduce(0, (x, y) -> x + y);
        
        System.out.println(sum);
        # 形式二
        Optional<Double> op = emps.stream()
            .map(Employee::getSalary)
            .reduce(Double::sum);
        
        System.out.println(op.get());

map 和 reduce 的连贯通常称为 map-reduce 模式

2.3.3 收集
  • collect(Collector c) 将流转换为其余模式。接管一个 Collector 接口的实现，用于给 Stream 中元素做汇总的办法

其中 toMap 的时候能够指定键抵触时候的策略

 Map<Long, CartItemSku> collect = Stream.concat(list.stream(), frontCartItems.stream())
                        .collect(Collectors
                            .toMap(CartItemSku::getSkuId,  val -> val, (oldValue, newValue) -> {oldValue.setQuantity(oldValue.getQuantity() + newValue.getQuantity());
                             return oldValue;
                          }));

Collector 接口中办法的实现决定了如何对流执行收集操作(如收集到 List、Set、Map)。然而 Collectors 类提供了很多静态方法，能够不便地创立常见收集器实例，具体应用如下图：

3. 并行流和串行流

3.1. 简介

并行流就是把一个内容分成多个数据块，并用不同的线程别离解决每个数据块的流。
Java 8 中将并行进行了优化，咱们能够很容易的对数据进行并行操作。Stream API 能够申明性地通过 parallel() 与 sequential() 在并行流与程序流之间进行切换。

3.2. 理解 Fork/Join 框架

Fork/Join 框架：就是在必要的状况下，将一个大工作，进行拆分 (fork) 成若干个小工作（拆到不可再拆时），再将一个个的小工作运算的后果进行 join 汇总.

Fork/Join 框架与传统线程池的区别:

采纳“工作窃取”模式（work-stealing）：
当执行新的工作时它能够将其拆分分成更小的工作执行，并将小工作加到线程队列中，而后再从一个随机线程的队列中偷一个并把它放在本人的队列中。绝对于个别的线程池实现,fork/join 框架的劣势体现在对其中蕴含的工作的解决形式上. 在个别的线程池中, 如果一个线程正在执行的工作因为某些起因无奈持续运行, 那么该线程会处于期待状态. 而在 fork/join 框架实现中, 如果某个子问题因为期待另外一个子问题的实现而无奈持续运行. 那么解决该子问题的线程会被动寻找其余尚未运行的子问题来执行. 这种形式缩小了线程的等待时间, 进步了性能。

3.3. 应用

java8 中的并行流底层应用的就是 Fork/Join 模式。

    long start = System.currentTimeMillis();
        
    Long sum = LongStream.rangeClosed(0L, 10000000000L)
                         .parallel()
                         .sum();
        
    System.out.println(sum);    
    long end = System.currentTimeMillis();
        
    System.out.println("消耗的工夫为:" + (end - start));

4 总结

Stream API 的性能能够说十分弱小，学会是在练习的根底之上，所以勤分割能力晋升。

正文完
程序员
发表至： 程序员
2020-12-20
 0
关于程序员:专栏开通记录一下
关于程序员:Python-异步-同时运行多个协程10
关于程序员:融云-2022-社交泛娱乐出海嘉年华邀你一起超浪
关于程序员:共建开源人才生态2022-开放原子全球开源峰会聚焦-产学研用