《Java 8 实战》（三）---- 流

一　　流的概念

流是Java API的新成员，它允许以声明性方式处理数据集合（通过查询语句来表达，而不是临时编写一个实现）。

Java 7　从集合中选出低热量菜肴名称：

List<Dish> lowCaloricDishes = new ArrayList<>();
        for (Dish d: menu) {
            if (d.getCalories() < 400) {
                lowCaloricDishes.add(d);
            }
        }
        
        Collections.sort(lowCaloricDishes, new Comparator<Dish>() {
            public int compare(Dish d1, Dish d2) {
                return Integer.compare(d1.getCalories(), d2.getCalories());
            }
        });
        List<String> lowCaloricDishesName = new ArrayList<>();
        for(Dish d: lowCaloricDishes) {
            lowCaloricDishesName.add(d.getName());
        }

Java 8 :

List<String> lowCaloricDishesName2 = menu.stream()
                .filter(d -> d.getCalories() < 400)
                .sorted(comparing(Dish::getCalories))
                .map(Dish::getName)
                .collect(toList());

如果要多核架构并行执行这段代码，只需要把stream()换成parallelStream():

List<String> lowCaloricDishesName2 = menu.parallelStream()
                .filter(d -> d.getCalories() < 400)
                .sorted(comparing(Dish::getCalories))
                .map(Dish::getName)
                .collect(toList());

二　　集合与流的区别

集合与流之间的差异在于什么时候进行计算。

集合是一个内存中的数据结构，它包含数据结构中目前所有的值－－集合中的每个元素都得先算出来才能添加到集合中。可以往集合里加东西或者删东西，但不管什么时候，集合众的每个元素都是放在内存中的，元素都得先算出来才能成为集合的一部分。

流是在概念上固定的数据结构（你不能删除或添加元素），其元素是按需计算的。从另一个角度来讲，流就像一个延迟创建的集合：只有在消费者要求的时候才会计算值。

三　　流的特性

和迭代器类似，流只能遍历一次。遍历完之后，这个流就被消费掉了。

集合的迭代(for-each)叫外部迭代，Stream库用的是内部迭代：它已经把迭代做了，还把得到的流值存在了某个地方，只要给出一个函数说要干什么就可以了。Streams库德内部迭代可以自动选择一种适用于硬件的数据表示和并行实现。但如果使用for-each，就只能自己管理所有的并行问题了。

四　　流操作：

List<String> lowCaloricDishesName2 = menu.stream()
                .filter(d -> d.getCalories() < 400)
                .sorted(comparing(Dish::getCalories))
                .map(Dish::getName)
　　　　　　　　　 .limit(3)
                .collect(toList());

filter/map/limit可以连成一条流水线；collect触发流水线执行并关闭它。

所以Stream的操作可以分为两类：

　　可以连结起来的流操作成为中间操作。返回一个另一个流，除非流水线上触发一个终端操作，否则中间操作不会执行任何处理，它们很懒。这是因为中间操作一般都可以合并起来，在终端操作时一次性全部处理。

　　关闭流的操作成为终端操作。终端操作会从流的流水线生成结果，其结果是任何不是流的值，比如List/Integer,甚至是void.

无限流是没有固定大小的流。

五　　使用流

1.　　筛选和切片：

　　１，用谓词筛选：

filter操作，它接受一个谓词作为参数，并返回一个包括所有符合谓词元素的流。

List<Dish> vegetarianMenu = menu.stream().filter(Dish::isVegetarian).collect(toList());

distinct操作，它会返回一个元素各异（根据流生成元素的hashCode和equals方法实现）的流。

List<Integer> numbers = Arrays.asList(1,2,1,3,3,2,4);
numbers.stream().filter(i->i%2==0).distinct().forEach(System.out::println);

limit方法，会返回一个不超过给定长度的流。所需长度作为参数传递给该方法。

List<Dish> dishes = menu.stream().filter(d -> d.getCalories() > 300).limit(3).collect(toList());

skip(n)方法，返回一个扔掉了前ｎ各元素的流。如果流中元素不足ｎ个，则返回一个空流。limit(n)和skip(n)是互补的。

List<Dish> dishes = menu.stream().filter(d -> d.getCalories() > 300).skip(2).collect(toList());

2　　映射

map方法，它会接受一个函数作为参数，这个函数会应用到每个元素上，并将其映射成一个新的元素。

List<String> dishName = menu.stream().map(Dish.getName).collect(toList());

因为getName返回一个String，所以map方法输出的流的类型就是Stream<String>.

List<String> words = Arrays.asList("Java 8", "Lambdas", "In", "Action");
List<Integer> wordLengths = words.stream().map(String::length).collect(toList());

给定一个单词列表，想要显示每个单词中有几个字母。

假设现在需求改为，给定单词列表["Hello", "World"],　想要返回列表["H","e","l","o","W","r","d"].

第一个版本可能是：　

words.stream().map(word -> word.split("")).distinct().collect(toList());

但传给map的方法返回一个String[]类型，所以map返回的流也变成了stream<String[]>, 经过distinct后，是对["H","e","l","l","o"]和["W","o","r","l","d"]进行去重操作，所以结果还是["H","e","l","l","o","W","o","r","l","d"]

Arrays.stream(),可以接受一个数组并产生一个流。第二个版本：

words.stream().map(word -> word.split("")).map(Arrays::stream).distinct().collect(toList());

现在得到的是一个流的列表，先把每个单词转换成一个字母数组，然后把每个数组变成了一个独立的流，所以结果还是不对。

flatMap的作用是各个数组并不是分别映射成一个流，而是映射成流的内容。所以使用flatMap(Arrays::stream)时生成的单个流都被合并起来，即扁平化为一个流。flatMap方法把一个流中的每个值都换成另一个留

3　　查找和匹配

anyMatch: 流中是否有一个元素能匹配给定的谓词。anyMatch返回一个boolean,因此是一个终端操作。

if (menu.stream().anyMatch(Dish::isVegetarian)) {
    System.out.println("The menu is (somewhat) vegetarian friendly!!");
}

allMatch: 流中元素是否都能匹配给定谓词。

boolean isHealthy = menu.stream().allMatch(d -> d.getCalories() < 1000);

noneMatch: 流中是否没有任何元素与给定谓词匹配。

boolean isHealthy = menu.stream().noneMatch(d -> d.getCalories() >= 1000);

这三个操作都用到了短路，它们不用处理整个流就能得到结果。

findAny将返回当前流中的任意元素。

Optianal<Dish> dish = menu.stream().filter(Dish::isVegetarian).findAny();

Optional<T>类是一个容器类，代表一个值存在或不存在。这样就不用返回容易出问题的null了。

findFirst查找第一个元素。

List<Integer> someNumbers = Arrays.asList(1,2,3,4,5);
Optional<Integer> firstSquareDivisibleByThree = someNumbers.stream().map(x - x*x).filter(x -> x%3 == 0).findFirst(); // 9

4　　归约

reduce操作将流中所有元素反复结合起来得到一个值，比如Integer，这样的查询可以被归类为归约操作（将流归约成一个值）。

int sum = numbers.stream().reduce(0, (a, b) -> a + b);

int product = numbers.stream().reduce(1, (a, b) -> a * b);

Lambda反复结合每个元素，直到流被归约成一个值。

求最大值和最小值：

Optional<Integer> max = numbers.stream().reduce(Integer::max);

5　　数值流

Java 8引入了三个原始类型特化流来解决这个问题：IntStream/DoubleStream和LongStream，分别将流中的元素特化为int/long和double，从而避免了暗含的装箱成本。每个接口都有进行常用数值归约的新方法，如sum，max，还有在必要时再把它们转换回对象流的方法。

映射到数值流mapToInt,mapToDouble,mapToLong：

int calories = menu.stream().mapToInt(Dish::getCalories).sum();

转换回对象流boxed：

IntStream intStream = menu.stream().mapToInt(Dish::getCalories);
Stream<Integer> stream = intStream.boxed();

默认值OptionalInt

Optional类可以表示值存在或不存在的容器，对于三种原始流特化，分别有Optional原始类型特化版本：OptionalInt/OptionalDouble和OptionalLong

OptionalInt maxCalories = menu.stream().mapToInt(Dish::getCalories).max();

int max = maxCalories.orElse(1); // 如果没有最大值的话，显式提供一个默认最大值。

数值范围range&rangeClosed:

range&rangeClosed用于生成数值范围，加入要生成１到１００之间的所有数字：

IntStream evenNumbers = IntStream.rangeClosed(1, 100).filter(n -> n%2 ==0);

range&rangeClosed都接受两个参数，第一个是起始值，第二个是结束值，但range不包含结束值，rangeClosed包含结束值。

７　　构建流

由值创建流Stream.of:

Stream<String> stream = Stream.of("Java 8", "Lambda", In", "Action");
stream.map(String::toUpperCase).forEach(System.out::println);

Stream.empty得到一个空流：

Stream<String> emptyStream = Stream.empty();

由数组创建流Arrays.stream:

int[] numbers = {2, 3, 5, 7, 11, 13};
int sum = Arrays.stream(numbers).sum(); // 41

由文件生成流：

long uniqueWords = 0;
try(Stream<String> lines = Files.lines(Paths.get("data.txt"), Charset.defaultCharset())) {
    uniqueWords = lines.flatMap(line -> Arrays.stream(line.split(" ")))
　　　　　　　　　　.distinct()
　　　　　　　　　　.count();
} catch(IOException e) {
}

由函数生成流：

Stream.iterate

Stream.iterate(0, n->n+2).limit(10).forEach(System.out::println);

生成0,2,4,6,8,10,12,14,16,18

一般来说，在需要依次生成一系列值得时候应该使用iterate。

斐波纳契元组序列：

Stream.iterate(new int[]{0,1}, t -> new int[] {t[1], t[0] + t[1]})

Stream.generate

Stream.generate(Math::random).limit(5).forEach(System.out::println);

六　　用流收集数据

归约和汇总

在需要将流项目重组成集合时，一般会使用收集器（Stream.collect()方法的参数）。Collectors类中有很多静态工厂方法，这些方法会生成收集器，用在Stream.collect()方法中。

1， 查找流中的最大最小值：

可以使用两个收集器：Collectors.maxBy和Collectors.minBy。这两个收集器接受一个Comparator参数来比较流中的元素。

比如选择菜单中热量最高的菜：

Comparator<Dish> dishCaloriesComparator = Comparator.comparingInt(Dish::getCalories);

Optional<Dish> mostCalorieDish = menu.stream().collect(maxBy(dishCaloriesComparator));

2, 汇总

Collectors类专门为汇总提供了一个工厂方法：Collectors.summingInt. 它可以接受一个把对象映射为求和所需int的函数，并返回一个收集器；该收集器在传递给普通的collect方法后即执行我们需要的汇总操作。

比如求出菜单列表的总热量：

int totalCalories = menu.stream().collect(summingInt(Dish:;getCalories));

Collectors.averageInt/averageLong/averageDouble可以计算数值的平均值：

double avgCalories = menu.stream().collect(averagingInt(Dish::getCalories));

Collectors.summarizing操作可以一次操作就得到集合的最大值，最小值，平均值，总和。

IntSummaryStatistics menuStatistics = menu.stream().collect(summarizing(Dish::getCalories));

这个收集器会把所有这些信息收集到一个叫做IntSummaryStatistics类里，它提供了方便的取值方法来访问结果。打印menuStatistics得到以下输出：

IntSummaryStatistics{count=9, sum=4300, min=120, average=477.777778, max=800}

3，连接字符串

Collectors.joining工厂方法返回的收集器会把对流中每个对象应用toString方法得到的所有字符串连接成一个字符串。joining在内部使用了StringBuilder来把生成的字符串逐个追加起来。

把菜单中所有菜肴的名称连接起来：

String shortMenu = menu.stream().map(Dish::getName).collect(joining());

如果Dish类有一个toString方法来返回菜肴的名称，那就无需用提取每道菜名称的函数来对原流做映射就能得到相同的结果：

String shortMenu = menu.stream().collect(joining());

joining方法有一个重载版本可以接受元素之间的分界符,可以以该分界符分隔字符串：

String shortMenu = menu.stream().map(Dish::getName).collect(joining(", "));

分组

Collectors.groupingBy方法可以根据一个或多个属性对集合中的项目进行分组。

Map<Dish.Type, List<Dish>> dishesByType = menu.stream().collect(groupingBy(Dish::getType));

给groupingBy方法传递一个分类函数，这个函数可以把流中的元素分成不同的组。分组结果是一个map，分组函数返回的值作为键，流中所有具有这个分类值得项目的列表作为对应的映射值。但分类函数不一定像方法引用那样可用，因为分类函数根据需求改变，不一定把这个操作写成一个方法，所以就不能用方法引用。但可以把分类函数写成Lambda表达式： 

public enum CaloricLevel {DIET, NORMAL, FAT}

Map<CaloricLevel, List<Dish>> dishesByCaloricLevel = menu.stream().collect(groupingBy(dish -> {
    if (dish.getCalories() <= 400) return CaloricLevel.DIET;
    else if (dish.getCalories() <= 700) return CaloricLevel.NORMAL;
    else return CaloricLevel.FAT;
));

多级分组

要实现多极分组，可以使用一个由双参数版本的Collectors.groupingBy工厂方法创建的收集器，它除了普通的分类函数之外，还可以接受Collector类型的第二个参数。那么要进行二级分组的话，可以把一个内层groupingBy传递给外层groupingBy，并定义一个为流中项目分类的二级标准：

Map《Dish.Type, Map<CaloricLevel, List<Dish>>> dishesByTypeCaloricLevel = menu.stream().collect(groupingBy(Dish::getType, groupingBy(dish -> {
    if (dish.getCalories() <= 400) return CaloricLevel.DIET;
    else if (dish.getCalories() <= 700) return CaloricLevel.NORMAL;
    else return CaloricLevel.FAT;
})))

结果：

{MEAT={DIET=[chicken], NORMAL=[beef], FAT=[pork]},
FISH={DIET=[prawns], NORMAL=[salmon]},
OTHER={DIET=[rice, seasonal fruit], NORMAL=[french fries, pizza]}}

 按子组收集数据

传递给第一个groupingBy的第二个收集器可以是任何类型，而不一定是另一个groupingBy。

比如要数菜单中每类菜的个数，可以传递counting收集器作为groupingBy收集器的第二个参数：　

Map<Dish.Type, Long> typesCount = menu.stream().collect(groupingBy(Dish::getType, counting()));

普通的单参数groupingBy(f)实际上是groupingBy(f, toList())的简便写法。

把收集器的结果转换为另一个类型：

Collectors.collectingAndThen可以把收集器返回的结果转换为另一种类型

Map<Dish.Type, Dish> mostCaloriecByType = menu.stream().collect(groupingBy(Dish::getType, collectingAndThen(maxBy(comparingInt(Dish::getCalories)),Optional::get)));

分区

分区是分组的特殊情况：由一个谓词作为分类函数，它称为分区函数。分区函数返回一个布尔值，所以最多可以分为两组--true或false各一组。partitioningBy只接受谓词作参数。

Map<Boolen, List<Dish>> partitionedMenu = menu.stream().collect(partitioningBy(Dish::isVegetarian));
List<Dish> vegetarianDishes = partitionedMenu.get(true);

结果：

{false=[pork, beef, chicken, prawns, salmon],
true=[french fries, rice, season fruit, pizza]}

七　　并行数据处理与性能

并行流

并行流就是一个把内容分成多个数据块，并用不同的线程分别处理每个数据块的流。这样就可以自动把给定操作的工作分配给多核处理器的所有核，让所有核都参与工作。

例如，返回从１到给定参数Ｎ之间所有数字的和。

public static long sequentialSum(long n) {
    return Stream.iterate(1L, i -> i+1)
                    .limit(n)
                    .reduce(0, Long::sum);
}

把流转换成并行流，从而让前面的函数归约过程并行运行

public static long parallelSum(long n) {
    return Stream.iterate(1L, i -> i+1)
                        .limit(n)
                        .parallel()
                        .reduce(0L, Long::sum);
}

Stream在内部分成了几块，可以对不同的块独立并行进行归纳操作，最后，同一个归纳操作会将各个子流的部分归纳结果合并起来，得到整个原始流的归纳结果。

并行流内部使用了默认的ForkJoinPool，它默认的线程数量就是处理器的数量。

上例的并行流性能并没有顺序流性能好，原因是：

　　１，iterate生成的装箱的对象，必须拆箱成数字才能求和。

　　２，很难把iterate分成多个独立块来并行执行。因为每次应用这个函数都要依赖前一次应用的结果。

所以并行编程有时候并没有提升性能，反而为顺序处理增加了开销。为了解决上述问题，可以使用LongStream.rangeClosed, 相比iterate，它有两个优点：

　　１，LongStream.rangeClosed直接产生原始类型的long数字，没有装箱拆箱的开销。

　　２，LongStream.rangeClosed会生成数字范围，很容易拆分为独立的小块。

对新版本应用并行流：

public static long parallelRangedSum(long n) {
    return LongStream.rangeClosed(1, n)
                            .parallel()
                            .reduce(0, Long::sum);
}

这个并行流性能得到很大提升。所以使用正确的数据结构然后使其并行工作能够保证最佳的性能。

并行化并不是没有代价的。并行化过程本身需要对流做递归划分，把每个子流的归纳操作分配到不同的线程，然后把这些操作的结果合并成一个值。但在多个内核之间移动数据的代价也很大。

使用并行流的建议：

　　１，如果不确定并行流和顺序流的性能，那就测试一下。

　　２，留意装箱。自动装箱和拆箱操作会大大降低性能。

　　３，有些操作本身在并行流上的性能就比顺序流差，特别是limit和findFirst等依赖于元素顺序的操作，在并行流上执行的代价非常大。

　　４，还要考虑流的操作流水线的总计算成本。设N是要处理的元素的总数，Q是一个元素通过流水线的大致处理成本，则N*Q就是这个对成本的一个粗略的定性估计。Q值较高就意味着使用并行流时性能好的可能性比较大。

　　５，对于较小的数据量，选择并行流几乎从来不是一个好的决定。

　　６，考虑流背后的数据结构是否易于分解。ArrayList的拆分效率比LinkedList高得多，因为前者用不着遍历就可以平均拆分，后者则必须遍历。

　　７，流自身的特点，以及流水线中中间操作修改流的方式，都可能改变分解过程的性能。

　　８，还要考虑中断操作中合并步骤地代价是大是小。