[大数据]Stream

Stream

Stream 流

流是Java SE 8引入的特性，通常处理集合中的元素通过迭代，并实现元素的处理逻辑，而流通过做什么的方式来处理集合中的元素，不用关注怎么做。

1. 流的操作尽可能的惰性执行，如果需要其结果是才会执行到。
2. 流操作不修改原来的数据，会产生的新的流。
3. 操作的流的简单典型流程：创建流(.stream())->初始流转化为其他流(.filter())->终止操作(.count())

  //统计集合中长度超过5的元素
  List<String> strings = Arrays.asList(new String[]{"12345", "123456", "1"});
  strings.stream().filter(w -> w.length() > 5).count();

流的创建

1. 通过Collection接口的stream方法。

  List<String> strings = Arrays.asList(new String[]{"12345", "123456", "1"});
  strings.stream().filter(w -> w.length() > 5).count();

2. Stream.of() 传入可变参数。

  Stream<String> str = Stream.of("hh","mm");

3. Arrays.stream(array, from, to)提取数组中from到to的元素（包含from不包含to）生成流。

  Arrays.stream(new String[]{"12345", "123456", "1"},0, 3);

4. Stream.empty() 创建一个空的流。

  Stream<Integer> stream = Stream.empty();

5. Stream.generate(Math::random) 创建常量或无限元素流。

  Stream.generate(() -> "hh");
  Stream.generate(Math::random);

6. Stream.iterate(feed, f(feed)) feed种子，通过f(feed)运算下一个值

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

filter、map和flatMap

1. filter 引元为Predicate，产生boolean的结果，返回符合条件的流元素
2. map 返回元素进行函数操作以后的结果，原集合中有多个数组，分别处理完成后可能为[[1,2],[3,4]]
3. flatMap将返回的多个列表进行整合成一个列表，将上述结构合并为[1,2,3,4]

子流和连接流

1. 子流

  //截取前哪位
  stream.limit(long n);
  //跳过前n位
  stream.skip(long n);

2. 连接流 Stream<T> concat(Stream<? extends T> a, Stream<? extend T> b)

  Stream<String> stream =  Stream.concat(Stream.of("a","b"), Stream.of("c","d")）;

约简

1. reduce 通过在流中计算某个值，存在3种形式，提供一种约简操作op,reduce会进行a1 op a2 op…op an的操作，可用用来进行流中的元素数值进行
   运算操作，字符拼接等。

  Integer[] integers = {1,2,3};
  //1 提供一个二元函数进行运算
  Optional<Integer> sum = Stream.of(integers).reduce(Integer::sum);
  System.out.println(sum.get());
  //2 多提供一个幺元值，当流中元素为空是返回幺元值
  Integer sum1 = Stream.of(new Integer[]{}).reduce(0, Integer::sum);
  System.out.println(sum1);
  //3 对象流中对多个属性进行操作，需要提供一个累加变量,当并行流是需提供另个一函数对多个结果进行整合
  int totalValue = plays.parallelStream().
          reduce(0,
                  (total, y) -> total + y.getName().length(),
                  (total1, total2) -> total1 + total2);
  System.out.println(totalValue);

2. 简单约简

  //count() 统计流中元素数量
  list.stream().count();
  //查找第一个匹配的元素，需要配合filter使用
  Optional<String> start = list.stream().filter(i -> i.startsWith("a")).findFirst();
  //查找任意一个以a开头元素，可并行处理
  list.stream().paraller().filter(i -> i.startsWith("a")).findAny();
  //查找流中是否存在匹配，不关注值
  boolean result = list.stream().paraller().anyMatch();
  //allMatch和noneMatch分别对应全部匹配以及全不匹配返回逻辑值

收集流产生的结果（一个结果只能收集一次）

1. 输出流中的结果集，使用iterator()生成迭代器，或通过forEach()方法遍历

  Iterator iterator = Stream.concat(Stream.of("a","b"), Stream.of("c","d")).iterator();
  Stream.concat(Stream.of("a","b"), Stream.of("c","d")).forEach(System.out::println);

2. 转为数组通过流的toArray()，由于泛型不能在运行时生成数组，直接生成的Object[],可以传递类型对象引用到数组构造器生成具体类型数组

    Stream.concat(Stream.of("a","b"), Stream.of("c","d")).toArray(String[]::new);

3. 通Collectors.to**()方法收集流中的元素到集合中

  Stream<String> stream = Stream.empty();
  //收集到常用集合
  List<String> list = stream.collect(Collectors.toList());
  Set<String> set = stream.collect(Collectors.toSet());
  //收集到其他种类集合
  TreeSet treeSet = stream.collect(Collectors.toCollection(TreeSet::new));
  //收集为带分隔符的字符串
   String str = stream.collect(Collectors.joining(","));

4. 收集到映射表 Collectors.toMap()

toMap方法中两个函数引元，生成映射表的键和值，如果出现相同的键会抛出一个IllegalStateException异常，可以通过添加第三个引元函数解决。

  Map<Integer, String> map = Stream.of(play).collect(Collectors.toMap(Play::getId, Play::getName));
  //Function.identity() 
  Map<Integer, Play> map = Stream.of(play).collect(Collectors.toMap(Play::getId, Function.identity()));
  //old和new 只记录第一次的
  Stream<Locale> locales = Stream.of(Locale.getAvailableLocales());
        Map<String, String> localeMap = locales.
                collect(Collectors.toMap(Locale::getDisplayLanguage, l -> l.getDisplayLanguage(l),
                        (o,n) -> o));
  //提供第四个引元构造器可以生成指定的map类型
  Stream<Locale> locales = Stream.of(Locale.getAvailableLocales());
        Map<String, String> localeMap = locales.
                collect(Collectors.toMap(Locale::getDisplayLanguage, l -> l.getDisplayLanguage(l),
                        (o,n) -> o, TreeMap::new));

群组和分区

1. 通过groupingBy对集合进行分组，返回分组以后内容

  Stream<Locale> locales = Stream.of(Locale.getAvailableLocales());
  Map<String, List<Locale>> map = locales.collect(Collectors.groupingBy(Locale::getCountry));
  map.get("CH").stream().forEach(System.out::println);

2. 通过partitioningBy分组，断言返回true或false，比groupingBy高效

  Stream<Locale> locales = Stream.of(Locale.getAvailableLocales());
  Map<Boolean, List<Locale>> map = locales.
                  collect(Collectors.partitioningBy(l -> l.getLanguage().equals("en")));
  map.get(true).stream().forEach(System.out::println);

下游收集器

对收集到结果中的元素进行再处理,需要提供一个下游收集器，groupingBy产生的映射表可以进行，可以提供一个下游收集器再处理，例如将每个值的列表
转为集，IntSummarStatistics可以将每个组的值汇总成统计结果，获取总和、个数、平均值、最大最小值等。

  //将分组的结果转为集
  Stream<Locale> locales = Stream.of(Locale.getAvailableLocales());
  Map<String, Set<Locale>> localeMap = locales
          .collect(Collectors.groupingBy(Locale::getCountry, Collectors.toSet()));
  localeMap.get("CN").stream().forEach(System.out::println);
  
  //对分组的结果进行分别约简counting/summingInt(Double|Long)
  Map<String, Long> count = locales
          .collect(Collectors.groupingBy(Locale::getCountry, Collectors.counting()));
  System.out.println(count.get("CN"));
  
  //maxBy和minBy通过比较器，获取下游元素的最大最小值
  Map<String, Optional<String>> country = locales
                .collect(Collectors.groupingBy(Locale::getDisplayCountry,
                        Collectors.mapping(Locale::getDisplayLanguage, Collectors.maxBy(Comparator.comparing(String::length)))));
  System.out.println(country.get("卢森堡"));
  
  //map获取分组国家元素分别转为集
  Map<String, Set<String>> country = locales
                .collect(Collectors.groupingBy(Locale::getDisplayCountry,
                        Collectors.mapping(Locale::getDisplayLanguage, Collectors.toSet())));
        System.out.println(country.get("卢森堡"));

基本类型流

流对基本类型也提供了支持，具有专门IntStream.of()方法，generate，iterate方法。、LongStream、DoubleStream，short、char、byte、boolean可以用，float可以用DoubleStream。
创建可以直接使用IntStream.of()方法，generate，iterate方法。
对象流和基本流可以相互转化，mapToInt、mapToLong、mapToDouble,可以将对象流转化为基本类型流，boxed()反之。

并行流

并行处理可提高处理效率，需要保证中间的流操作可以任意顺序运行并不影响最终结果，不在意元素的顺序可以通过unorderded方法放弃顺序，这样可以更有效
并行操作，放弃顺序也可以提升limit()效率。

  //创建一个并行流
  Stream<String> parallelString = list.parallelStream()
  //将普通流转为并行流
  Stream<String> parallelString = Stream.of(array).parallel();

并行流需满足条件：

1. 数据已经存在内存中
2. 流可以被高效分成若子部分
3. 流操作不能被阻塞