Java 8 实战 P2 Functional-style data processing

内容导读

互联网集市收集整理的这篇技术教程文章主要介绍了Java 8 实战 P2 Functional-style data processing，小编现在分享给大家，供广大互联网技能从业者学习和参考。文章包含20087字，纯文字阅读大概需要29分钟。

内容图文

Chapter 4. Introducing streams

4.1 流是什么

1.声明性，可复合，可并行

List<String> lowCaloricDishesName =
                menu.stream()//利用多核架构用parallelStream()
                    .filter(d -> d.getCalories() < 400)
                    .sorted(comparing(Dish::getCalories))
                    .map(Dish::getName)
                    .collect(toList());

2.流简介
从支持数据处理操作的源生成的元素序列
元素序列：可以访问特定元素类型的一组有序值。但流的目的在于表达计算
源：从有序集合生成流时会保留原有的顺序
数据处理操作：支持类似于数据库的操作。顺序执行，也可并行执行。

4.2 流与集合

集合与流之间的差异就在于什么时候进行计算。比如前者需要全部计算完在显示（在一个时间点上全体存在），后者元素则是按需计算的，算到哪显示到哪（数据分布在一段时间里）。
1.一个stream只能遍历一次(和迭代器类似)

List<String> title = Arrays.asList("Java8", "In", "Action");
Stream<String> s = title.stream();
s.forEach(System.out::println);
s.forEach(System.out::println);//报错

集合可多次（不过其实第二次就相当于第二个迭代器了）
2.外部迭代与内部迭代

//外部
List<String> names = new ArrayList<>();
for(Dish d: menu){
    names.add(d.getName());
    
//内部：Stream帮你把迭代做了，你只需要给出函数表示要做的事情
List<String> names = menu.stream()
                         .map(Dish::getName)
                         .collect(toList());

内部的好处：可以透明地并行处理，或者用更优化的顺序进行处理

4.3 流操作（类似Spark的transform-action）

menu.stream()
    .filter(d -> d.getCalories() > 300)
    .map(Dish::getName)
    .limit(3)
    .collect(toList());

1.intermediate operations

• 上面的代码，由于limit和短路技巧（stay tuned），filter和map只操作三个对象
• 尽管filter和map是两个独立的操作，但它们合并到同一次遍历中

常用中间操作

2.terminal operations
menu.stream().forEach(System.out::println);
常用终端操作

forEach, count, collect
3.使用流
数据源-.stream()-中间操作-终端操作

Chapter 5. Working with streams

5.1 筛选和切片(略)

5.2 映射

//words为["Hello","World"]
List<String> uniqueCharacters =
    words.stream()
         .map(w -> w.split(""))// 得[["H","e","l","l", "o"],["W","o","r","l","d"]]流（stream<String[]>）
         .flatMap(Arrays::stream)//使得使用map(Arrays::stream)时生成的单个流都被合并起来（stream<String>）
         .distinct()
         .collect(Collectors.toList());

map是一变一，flatmap是多变一

//[1, 2, 3]和[3, 4]变为总和能被3整除的数对[(2, 4), (3, 3)]
List<Integer> numbers1 = Arrays.asList(1, 2, 3);
List<Integer> numbers2 = Arrays.asList(3, 4);
List<int[]> pairs =
    numbers1.stream()
            .flatMap(i -> numbers2.stream()
                                  .filter(j -> (i + j) % 3 == 0)
                                  .map(j -> new int[]{i, j})
                    )
            .collect(toList());

5.3 查找和匹配

都运用了短路技巧
anyMatch, allMatch, noneMatch
findAny,findFirst（并行时不合适）

Optional<Dish> dish = //Optional<T>类是一个容器类，代表一个值存在或不存在
    menu.stream()
        .filter(Dish::isVegetarian)
        .findAny();//返回当前流中的任意元素
        .ifPresent(d -> System.out.println(d.getName());

Optional里面几种可以迫使你显式地检查值是否存在或处理值不存在的情形的方法
isPresent(), ifPresent(Consumer<T> block), orElse(T other)

5.4 归约

int sum = numbers.stream().reduce(0, Integer::sum);
Integer::min
并行化
int sum = numbers.parallelStream().reduce(0, Integer::sum);
但要并行执行也要付一定代价：传递给reduce的Lambda不能更改状态（如实例变量），而且要满足结合律。

Tips：流操作的状态
无状态：不需要知道过去流的计算记录，如map或filter
有状态:须知道,如有界的（reduce、 sum、 max）和无界的（sort、distinct）

5.5 付诸实践（摘录）

//1.
.distinct()
.collect(toList());
//上面可改为，当然，返回类型不一样
.collect(toSet())

//2.字符串的合并
.reduce("", (n1, n2) -> n1 + n2);
//用下面的更好
.collect(joining());

//3.找到交易额最小的交易
Optional<Transaction> smallestTransaction =
    transactions.stream()
                .min(comparing(Transaction::getValue));

5.6 数值流

对.reduce(0, Integer::sum);的改进
Streams接口没有定义sum方法，因为一个像menu那样的Stream<Dish>，把各种菜加起来是没有任何意义的

1.原始类型流特化
映射到数值流：mapToInt、 mapToDouble和mapToLong

.mapToInt(Dish::getCalories)
.sum()//特化后就可以用sum、max、 min、 average等方法了

返回的是一个特化流（此处为数值流IntStream），而不是Stream

转换回对象流(把上面未用sum的IntStream转换回来)
Stream<Integer> stream = intStream.boxed();

默认值（针对非sum）
OptionalInt、 OptionalDouble和OptionalLong
在赋值给一个OptionalInt后，可以显示提供默认值,以防没有值
int max = maxCalories.orElse(1);

2.数值范围
range和rangeClosed

IntStream evenNumbers = IntStream.rangeClosed(1, 100)
                                 .filter(n -> n % 2 == 0);//还没计算

3.数值流应用
生成a和b在100内的勾股三元数组，三个数都为整数

Stream<double[]> pythagoreanTriples2 =
    IntStream.rangeClosed(1, 100).boxed()
             .flatMap(//注意flatMap的应用
                a -> IntStream.rangeClosed(a, 100)
                              .mapToObj(
                                 b -> new double[]{a, b, Math.sqrt(a*a + b*b)})
                              .filter(t -> t[2] % 1 == 0));//判断第三个数是否为整数用 (数 % 1 == 0)

注意上面正常map的话要在map前加boxed(),因为map会为流中的每个元素返回一个int数组，IntStream中的map方法只能为流中的每个元素返回另一个int

5.7 构建流

//由值创建流
Stream<String> stream = Stream.of("Java 8 ", "Lambdas ", "In ", "Action");
stream.map(String::toUpperCase).forEach(System.out::println);

//空流
Stream<String> emptyStream = Stream.empty();

//由数组创建流（numbers为int[]）
int sum = Arrays.stream(numbers).sum();

//由文件生成流，并计算不同单词的数量
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){}

//由函数生成流
//iterate
Stream.iterate(0, n -> n + 2)
      .limit(10)
      .forEach(System.out::println);
      
Stream.iterate(new int[]{0, 1},
                t -> new int[]{t[1], t[0]+t[1]})
      .limit(20)
      .forEach(t -> System.out.println("(" + t[0] + "," + t[1] +")"));

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

Chapter 6. Collecting data with streams

Map<Currency, List<Transaction>> transactionsByCurrencies =
        transactions.stream().collect(groupingBy(Transaction::getCurrency));

6.1 收集器简介

收集器定义collect用来生成结果集合的标准。
对流调用collect方法将对流中的元素触发一个归约操作（由Collector来参数化）。
三大功能：将流元素归约和汇总为一个值，元素分组，元素分区

6.2 归约和汇总

数一数菜单里有多少种菜
long howManyDishes = menu.stream().count();

//1.查找流中的最大值和最小值
Comparator<Dish> dishCaloriesComparator =
    Comparator.comparingInt(Dish::getCalories);
Optional<Dish> mostCalorieDish =
    menu.stream()
        .collect(maxBy(dishCaloriesComparator));

//2.汇总（求和，averagingInt，SummaryStatistics）
int totalCalories = menu.stream().collect(summingInt(Dish::getCalories));//summingInt可接受一个把对象映射为求和所需int的函数，并返回一个收集器

IntSummaryStatistics menuStatistics =
        menu.stream().collect(summarizingInt(Dish::getCalories));
//打印结果
IntSummaryStatistics{count=9, sum=4300, min=120,
                     average=477.777778, max=800}

//3.连接字符串
//返回的收集器会把对流中每一个对象应用toString方法得到的所有字符串连接成一个字符串
String shortMenu = menu.stream().map(Dish::getName).collect(joining(", "));
#如果Dish类有一个toString方法
String shortMenu = menu.stream().collect(joining(", "));

//4.广义的归约汇总reducing
//求和需要三个参数：起始值、转换函数、累积函数<T,T,T>
int totalCalories = menu.stream().collect(reducing(
                0, Dish::getCalories, Integer::sum));
//或
menu.stream().map(Dish::getCalories).reduce(Integer::sum).get();
//或（首推，简单，没有箱操作）
menu.stream().mapToInt(Dish::getCalories).sum();

//求最值
Optional<Dish> mostCalorieDish =
    menu.stream().collect(reducing(
        (d1, d2) -> d1.getCalories() > d2.getCalories() ? d1 : d2));
        
//求数量
reducing(0L, e -> 1L, Long::sum)

reduce方法旨在把两个值结合起来生成一个新值，它是一个不可变的归约。一旦以错误的语义使用reduce，就难以并行（影响性能）。
collect适合表达可变容器上的归约和并行操作

6.3 分组groupingBy

groupingBy(f)实际上是groupingBy(f, toList())，其结果是map<key,List<value>>

//有方法转换成所需标准的
Map<Dish.Type, List<Dish>> dishesByType =
                      menu.stream().collect(groupingBy(Dish::getType));

//自定义标准
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;
        } ));

//1.多级分组
//二级groupingBy情况下为`map<key,map<key,List<value>>>`
Map<Dish.Type, Map<CaloricLevel, List<Dish>>> dishesByTypeCaloricLevel =
menu.stream().collect(
      groupingBy(Dish::getType, 上一段代码的groupingBy)
);
        

Map<Dish.Type, Long> typesCount = menu.stream().collect(


//2.按子组收集数据                    
//求数量
Map<Dish.Type, Long> typesCount = menu.stream().collect(
                    groupingBy(Dish::getType, counting()));

//求最值
Map<Dish.Type, Dish> mostCaloricByType =
    menu.stream()
        .collect(groupingBy(Dish::getType,
                 collectingAndThen(
                    maxBy(comparingInt(Dish::getCalories)),
                 Optional::get)));//去掉maxBy产生的optional

//与mapping配合
//和之前两个groupingBy相比，要加上第二个参数toSet()，也可以是toCollection(HashSet::new) ；结果为map<key, Set<value>>，即只有CaloricLevel，没有菜名
Map<Dish.Type, Set<CaloricLevel>> caloricLevelsByType =
menu.stream().collect(
groupingBy(Dish::getType, mapping(上面根据calories分组的Lambda, toSet() )));

6.4 分区partitioningBy

分组的特殊情况：由一个谓词（返回一个布尔值函数，只能两组）作为分类（分区）函数

Map<Boolean, List<Dish>> partitionedMenu =
            menu.stream().collect(partitioningBy(Dish::isVegetarian));

List<Dish> vegetarianDishes = partitionedMenu.get(true);

1.分区的优势
和groupingBy一样可以多级分

Map<Boolean, Dish> mostCaloricPartitionedByVegetarian =
menu.stream().collect(
    partitioningBy(Dish::isVegetarian,
        collectingAndThen(maxBy(comparingInt(Dish::getCalories)),
                          Optional::get)));
//结果
{false=pork, true=pizza}

2.将数字按质数和非质数分区

//分区函数
public boolean isPrime(int candidate) {
    int candidateRoot = (int) Math.sqrt((double) candidate);
    return IntStream.rangeClosed(2, candidateRoot)
                    .noneMatch(i -> candidate % i == 0);//全为false是返回true，即不能被（2，candidateRoot）里面任何一个数整除
}

//生成质数和非质数区
public Map<Boolean, List<Integer>> partitionPrimes(int n) {
    return IntStream.rangeClosed(2, n).boxed()
                    .collect(partitioningBy(candidate -> isPrime(candidate)));
}

Collectors类的静态工厂方法表

6.5 收集器接口

public interface Collector<T, A, R> {
    Supplier<A> supplier();
    BiConsumer<A, T> accumulator();
    Function<A, R> finisher();
    BinaryOperator<A> combiner();
    Set<Characteristics> characteristics();
}

对于实现一个ToListCollector<T>类，将Stream<T>中的所有元素收集到一个List<T>里，它的代码如下

public class ToListCollector<T> implements Collector<T, List<T>, List<T>> {
    
    @Override
    //1.supplier方法必须返回一个无参数函数，在调用时它会创建一个空的累加器实例，供数据收集过程使用。
    public Supplier<List<T>> supplier() {
        return ArrayList::new;
    }
    
    @Override
    //2.accumulator方法会返回执行归约操作的函数
    //两个参数：保存归约结果的累加器（已收集了流中的前n-1个项目），还有第n个元素本身
    public BiConsumer<List<T>, T> accumulator() {
        return List::add;
    }
    
    @Override
    //3.在遍历完流后， finisher方法将累加器对象转换为最终结果
    public Function<List<T>, List<T>> finisher() {
        return Function.indentity();//无需进行转换
    }
    
    @Override
    //4.combiner方法会返回一个供归约操作使用的函数，它定义了并行处理时如何合并
    public BinaryOperator<List<T>> combiner() {
        return (list1, list2) -> {
            list1.addAll(list2);
            return list1;
        };
    }
   
    @Override
    //5.characteristics会返回一个不可变的Characteristics集合，它定义了收集器的行为:UNORDERED,CONCURRENT(用了UNORDERED或者无序数据源才用),IDENTITY_FINISH（暗示将累加器A不加检查地转换为结果R是可行的）
    public Set<Characteristics> characteristics() {
        return Collections.unmodifiableSet(EnumSet.of(
            IDENTITY_FINISH, CONCURRENT));
    }
}

实现时List<Dish> dishes = menuStream.collect(new ToListCollector<Dish>());
标准下List<Dish> dishes = menuStream.collect(toList());

6.6 开发你自己的收集器以获得更好的性能

优化上面提到的质数和非质数分区
1.仅用质数做除数

//takeWhile的方法，给定一个排序列表和一个谓词，它会返回元素满足谓词的最长前缀。
//即用于产生小于candidateRoot的质数列表
public static <A> List<A> takeWhile(List<A> list, Predicate<A> p) {
    int i = 0;
    for (A item : list) {
        if (!p.test(item)) {
            return list.subList(0, i);
        }
        i++;
    }
    return list;
}

//根据上面的表对candidate进行判断（是否为质数）
public static boolean isPrime(List<Integer> primes, int candidate){
    int candidateRoot = (int) Math.sqrt((double) candidate);
    return takeWhile(primes, i -> i <= candidateRoot)
                    .stream()
                    .noneMatch(p -> candidate % p == 0);
}

自定义collector

//确定类的签名，这里收集Integer流，累 加 器 和 结 果 类 型如下
public class PrimeNumbersCollector 
        implements Collector<Integer,
                             Map<Boolean, List<Integer>>,
                             Map<Boolean, List<Integer>>> {

    @Override
    //不但创建了用作累加器的Map，还为true和false两个键下面初始化了对应的空列表
    public Supplier<Map<Boolean, List<Integer>>> supplier() {
        return () -> new HashMap<Boolean, List<Integer>>() {{
                                put(true, new ArrayList<Integer>());
                                put(false, new ArrayList<Integer>());
                                }};
    }
    
    @Override
    //现在在任何一次迭代中，都可以访问收集过程的部分结果，也就是包含迄今找到的质数的累加器
    public BiConsumer<Map<Boolean, List<Integer>>, Integer> accumulator() {
        return (Map<Boolean, List<Integer>> acc, Integer candidate) -> {
                acc.get(isPrime(acc.get(true), candidate))//判断后得true_list或false_list
                   .add(candidate);//符合的添加到相应的List
                };
    }
    
    @Override
    //实际上这个收集器是不能并行使用的，因为该算法本身是顺序的。这意味着永远都不会调用combiner方法
    public BinaryOperator<Map<Boolean, List<Integer>>> combiner() {
        return (Map<Boolean, List<Integer>> map1,
                Map<Boolean, List<Integer>> map2) -> {
                    map1.get(true).addAll(map2.get(true));
                    map1.get(false).addAll(map2.get(false));
                    return map1;
                };
    }
    @Override
    //用不着进一步转换
    public Function<Map<Boolean, List<Integer>>,
                    Map<Boolean, List<Integer>>> finisher() {
                        return Function.identity();
    }
    @Override
    public Set<Characteristics> characteristics() {
        return Collections.unmodifiableSet(EnumSet.of(IDENTITY_FINISH));
    }
}

使用

public Map<Boolean, List<Integer>>
                    partitionPrimesWithCustomCollector(int n) {
    return IntStream.rangeClosed(2, n).boxed()
                    .collect(new PrimeNumbersCollector());
}

Chapter 7. Parallel data processing and performance

7.1 并行流

parallelStream和sequential
配置并行流使用的线程池:
并行流内部使用了默认的ForkJoinPool，它默认的线程数量就是你的处理器数量，这个值是由Runtime.getRuntime().availableProcessors()得到的。
可通过System.setProperty("java.util.concurrent.ForkJoinPool.common.parallelism","12");改，但不建议
1.测量流性能
启示：
iterate生成的是装箱的对象，必须拆箱成数字才能求和；
很难把iterate分成多个独立块来并行执行（你必须意识到某些流操作比其他操作更容易并行化）

而采用rangeClosed：原始类型的long数字，没有装箱拆箱的开销；会生成数字范围，很容易拆分为独立的小块

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

除了数据产生方式，数据结构之外，我们还需保证在内核中并行执行工作的时间比在内核之间传输数据的时间长。
2.正确使用并行流

//一个错误示范
public static long sideEffectSum(long n) {
    Accumulator accumulator = new Accumulator();
    LongStream.rangeClosed(1, n).forEach(accumulator::add);
    return accumulator.total;
}
public class Accumulator {
    public long total = 0;
    public void add(long value) { total += value; }
}

public static long sideEffectParallelSum(long n) {
    Accumulator accumulator = new Accumulator();
    LongStream.rangeClosed(1, n).parallel().forEach(accumulator::add);//这里foreach里的accumulator被不同线程抢用
    return accumulator.total;
}

共享可变状态会影响并行流以及并行计算，所以要避免
3.高效使用并行流
任何关于什么时候该用并行流的定量建议都是不可能也毫无意义的——机器不同

• 有疑问，测量
• 留意装箱
• 某些操作本身不适合并行，如limit和findFirst等依赖于元素顺序的操作。findFirst可以换findAny，或调用unordered来把有序流变无序。对无序流用limit可能会有改观
• 考虑流的操作流水线的总计算成本。设N是要处理的元素的总数， Q是一个元素通过流水线的大致处理成本，QN为总成本。 Q值较高就意味着使用并行流可能更好。
• 数据量
• 数据结构：如ArrayList的拆分效率比LinkedList高得多。HashSet和TreeSet都比较好。可以自己实现Spliterator来完全掌控分解过程
• 流自身的特点，以及流水线中的中间操作修改流的方式，都可能会改变分解过程的性能。如经过filter后的流的大小是未知的。
• 考虑终端操作中合并步骤的代价

7.2 Fork/Join

1.使用 RecursiveTask
把任务提交到这个池，必须创建RecursiveTask<R>的一个子类，其中R是并行化任务（以及所有子任务）产生的结果类型，或者如果任务不返回结果，则是RecursiveAction类型

public class ForkJoinSumCalculator
             extends java.util.concurrent.RecursiveTask<Long> {
             
    private final long[] numbers;
    private final int start;
    private final int end;
    
    public static final long THRESHOLD = 10_000;
    
    public ForkJoinSumCalculator(long[] numbers) {
        this(numbers, 0, numbers.length);
    }
    
    private ForkJoinSumCalculator(long[] numbers, int start, int end) {
        this.numbers = numbers;
        this.start = start;
        this.end = end;
    }
    
    @Override
    protected Long compute() {
        int length = end - start;
        if (length <= THRESHOLD) {
            return computeSequentially();
        }
        ForkJoinSumCalculator leftTask =
            new ForkJoinSumCalculator(numbers, start, start + length/2);
        leftTask.fork();
        ForkJoinSumCalculator rightTask =
            new ForkJoinSumCalculator(numbers, start + length/2, end);
        Long rightResult = rightTask.compute();
        Long leftResult = leftTask.join();
        return leftResult + rightResult;
    }
    private long computeSequentially() {
        long sum = 0;
        for (int i = start; i < end; i++) {{
            sum += numbers[i];
        }
        return sum;
    }
}

使用

public static long forkJoinSum(long n) {
    long[] numbers = LongStream.rangeClosed(1, n).toArray();//必须先要把整个数字流都放进一个long[]，所以性能会差一点
    ForkJoinTask<Long> task = new ForkJoinSumCalculator(numbers);//ForkJoinTask（RecursiveTask的父类）
    return new ForkJoinPool().invoke(task);
}

在实际应用时，使用多个ForkJoinPool是没有什么意义的。这里创建时用了其默认的无参数构造函数，这意味着想让线程池使用JVM能够使用的所有处理器（Runtime.availableProcessors的返回值，包括超线程生成的虚拟内核）。

2.使用分支/合并框架的最佳做法

• 对一个任务调用join方法会阻塞调用方，直到该任务做出结果。因此，有必要在两个子任务的计算都开始之后再调用它。
• 不应该在RecursiveTask内部使用ForkJoinPool的invoke方法。相反，你应该始终直接调用compute或fork方法，只有顺序代码才应该用invoke来启动并行计算。
• 对子任务调用fork方法可以把它排进ForkJoinPool，但不要左右都用。
• 调试使用分支/合并框架的并行计算比较麻烦
• 一个惯用方法是把输入/输出放在一个子任务里，计算放在另一个里，这样计算就可以和输入/输出同时进行。
• 分支/合并框架需要“预热”或者说要执行几遍才会被JIT编译器优化。
• 编译器内置的优化可能会为顺序版本带来一些优势（例如执行死码分析——删去从未被使用的计算）
• 设定好最小划分标准

3.工作窃取
分出大量的小任务一般来说都是一个好的选择。因为，理想情况下，划分并行任务时，能让所有的CPU内核都同样繁忙。即便实际中子任务花的时间差别还是很大，利用work stealing能解决这一问题。其过程是空闲的线程窃取其他线程的工作。

7.3 Spliterator（略）

虽然在实践中可能用不着自己开发Spliterator，但了解一下它的实现方式会让你对并行流的工作原理有更深入的了解。

public interface Spliterator<T> {
    boolean tryAdvance(Consumer<? super T> action);//类似于普通的Iterator
    Spliterator<T> trySplit();//划分元素给其他Spliterator使用
    long estimateSize();//估计还剩下多少元素要遍历
    int characteristics();
}

内容总结

以上是互联网集市为您收集整理的Java 8 实战 P2 Functional-style data processing全部内容，希望文章能够帮你解决Java 8 实战 P2 Functional-style data processing所遇到的程序开发问题。 如果觉得互联网集市技术教程内容还不错，欢迎将互联网集市网站推荐给程序员好友。

内容备注

版权声明：本文内容由互联网用户自发贡献，该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 gblab@vip.qq.com 举报，一经查实，本站将立刻删除。

内容手机端