java函数式编程

Posted 2022-05-27 可持续化发展

reduce 阶段的重构还差一小步就差不多结束了。我们要在最后调用toString 方法，将整个步骤串成一个方法链。这很简单，只需要排列好reduce 代码，准备好将其转换为Collector API 就行了（如例5-23 所示）。

例5-23　使用reduce 操作，将工作代理给StringCombiner 对象

String result =

    artists.stream()

           .map(Artist::getName)

           .reduce(new StringCombiner(", ", "[", "]"),

                   StringCombiner::add,

                   StringCombiner::merge)

           .toString();

现在的代码看起来已经差不多完美了，但是在程序中还是不能重用。因此，我们想将reduce 操作重构为一个收集器，在程序中的任何地方都能使用。不妨将这个收集器叫作StringCollector，让我们重构代码使用这个新的收集器，如例5-24 所示。

例5-24　使用定制的收集器StringCollector 收集字符串

String result =

    artists.stream()

           .map(Artist::getName)

           .collect(new StringCollector(", ", "[", "]"));

既然已经将所有对字符串的连接操作代理给了定制的收集器，应用程序就不需要关心StringCollector 对象的任何内部细节，它和框架中其他Collector 对象用起来是一样的。

先来实现Collector 接口（例5-25），由于Collector 接口支持泛型，因此先得确定一些具体的类型：

1、待收集元素的类型，这里是 String；

2、累加器的类型 StringCombiner；

3、最终结果的类型，这里依然是String。

例5-25　定义字符串收集器

public class StringCollector implements Collector<String, StringCombiner, String>

一个收集器由四部分组成。首先是一个Supplier，这是一个工厂方法，用来创建容器，在这个例子中，就是StringCombiner。和reduce 操作中的第一个参数类似，它是后续操作的初值（如例5-26 所示）。

例5-26　Supplier 是创建容器的工厂

public Supplier<StringCombiner> supplier()

    return () -> new StringCombiner(delim, prefix, suffix);

由于收集器可以并行收集，我们要展示的收集操作在两个容器上（比如StringCombiners）并行进行。收集器的每一个组件都是函数，因此我们使用箭头表示，流中的值用圆圈表示，最终生成的值用椭圆表示。收集操作一开始，Supplier 先创建出新的容器（如图5-3）。

收集器的accumulator 的作用和reduce 操作的第二个参数一样，它结合之前操作的结果和当前值，生成并返回新的值。这一逻辑已经在StringCombiners 的add 方法中得以实现，直接引用就好了（如例5-27 所示）。

例5-27　accumulator 是一个函数，它将当前元素叠加到收集器

public BiConsumer<StringCombiner, String> accumulator()

    return StringCombiner::add;

这里的accumulator 用来将流中的值叠加入容器中（如图5-4 所示）。

combine 方法很像reduce 操作的第三个方法。如果有两个容器，我们需要将其合并。同样，在前面的重构中我们已经实现了该功能，直接使用StringCombiner.merge 方法就行了（例5-28）。

例5-28　combiner 合并两个容器

public BinaryOperator<StringCombiner> combiner()

    return StringCombiner::merge;

在收集阶段，容器被combiner 方法成对合并进一个容器，直到最后只剩一个容器为止（如图5-5 所示）。

在使用收集器之前，重构的最后一步将toString 方法内联到方法链的末端，这就将StringCombiners 转换成了我们想要的字符串（如图5-6 所示）。

收集器的finisher 方法作用相同。我们已经将流中的值叠加入一个可变容器中，但这还不是我们想要的最终结果。这里调用了finisher 方法，以便进行转换。在我们想创建字符串等不可变的值时特别有用，这里容器是可变的。

为了实现finisher 方法，只需将该操作代理给已经实现的toString 方法即可（例5-29）。

例5-29　finisher 方法返回收集操作的最终结果

public Function<StringCombiner, String> finisher()

    return StringCombiner::toString;

从最后剩下的容器中得到最终结果。

特征是一组描述收集器的对象，框架可以对其适当优化。characteristics 方法定义了特征。在这里我有必要重申，这些代码只作教学用途，和joining 收集器的内部实现略有出入。

读者也许会认为StringCombiner 看起来非常有用，别担心——你没必要亲自去编写，Java 8 有一个java.util.StringJoiner 类，它的作用和StringCombiner 一样，有类似的API。

做这些练习的主要目的不仅在于展示定制收集器的工作原理，而且还在于帮助读者编写自己的收集器。特别是你有自己特定领域内的类，希望从集合中构建一个操作，而标准的集合类并没有提供这种操作时，就需要定制自己的收集器。

以StringCombiner 为例，收集值的容器和我们想要创建的值（字符串）不一样。如果想要收集的是不可变对象，而不是可变对象，那么这种情况就非常普遍，否则收集操作的每一步都需要创建一个新值。

想要收集的最终结果和容器一样是完全有可能的。事实上，如果收集的最终结果是集合，比如toList 收集器，就属于这种情况。

此时，finisher 方法不需要对容器做任何操作。更正式地说，此时的finisher 方法其实是identity 函数：它返回传入参数的值。如果这样，收集器就展现出IDENTITY_FINISH 的特征，需要使用characteristics 方法声明。

要点回顾

1、方法引用是一种引用方法的轻量级语法，形如： ClassName::methodName。

2、收集器可用来计算流的最终值，是  reduce 方法的模拟。

3、Java 8 提供了收集多种容器类型的方式，同时允许用户自定义收集器。

数据并行化

并行和并发

并发是两个任务共享时间段，并行则是两个任务在同一时间发生，比如运行在多核CPU上。如果一个程序要运行两个任务，并且只有一个CPU 给它们分配了不同的时间片，那么这就是并发，而不是并行。

并行化是指为缩短任务执行时间，将一个任务分解成几部分，然后并行执行。这和顺序执行的任务量是一样的，区别就像用更多的马来拉车，花费的时间自然减少了。实际上，和顺序执行相比，并行化执行任务时，CPU 承载的工作量更大。

数据并行化是指将数据分成块，为每块

数据分配单独的处理单元。还是拿马拉车那个例子打比方，就像从车里取出一些货物，放到另一辆车上，两辆马车都沿着同样的路径到达目的地。

当需要在大量数据上执行同样的操作时，数据并行化很管用。它将问题分解为可在多块数据上求解的形式，然后对每块数据执行运算，最后将各数据块上得到的结果汇总，从而获得最终答案。

拿任务并行化和数据并行化做比较，在任务并行化中，线程不同，工作各异。我们最常遇到的Java  EE 应用容器便是任务并行化的例子之一，每个线程不光可以为不同用户服务，还可以为同一个用户执行不同的任务，比如登录或往购物车添加商品。

并行化流操作

并行化操作流只需改变一个方法调用。如果已经有一个Stream 对象，调用它的parallel 方法就能让其拥有并行操作的能力。如果想从一个集合类创建一个流，调用parallelStream 就能立即获得一个拥有并行能力的流。

例6-1 计算了一组专辑的曲目总长度。它拿到每张专辑的曲目信息，然后得到曲目长度，最后相加得出曲目总长度。

例6-1　串行化计算专辑曲目长度

public int serialArraySum()

    return albums.stream()

                 .flatMap(Album::getTracks)

                 .mapToInt(Track::getLength)

                 .sum();

调用parallelStream 方法即能并行处理，如例6-2 所示，剩余代码都是一样的。

例6-2　并行化计算专辑曲目长度

public int parallelArraySum()

    return albums.parallelStream()

                 .flatMap(Album::getTracks)

                 .mapToInt(Track::getLength)

                 .sum();

读到这里，大家的第一反应可能是立即将手头代码中的stream 方法替换为parallelStream方法，因为这样做简直太简单了！先别忙，为了将硬件物尽其用，利用好并行化非常重要，但流类库提供的数据并行化只是其中的一种形式。

我们先要问自己一个问题：并行化运行基于流的代码是否比串行化运行更快？这不是一个简单的问题。回到前面的例子，哪种方式花的时间更多取决于串行或并行化运行时的环境。

并行化数组操作

Java  8 还引入了一些针对数组的并行操作，脱离流框架也可以使用Lambda 表达式。像流框架上的操作一样，这些操作也都是针对数据的并行化操作。让我们看看如何使用这些操作解决那些使用流框架难以解决的问题。这些操作都在工具类Arrays 中。

方法名	操　　作
parallelPrefix	任意给定一个函数，计算数组的和
parallelSetAll	使用Lambda 表达式更新数组元素
parallelSort	并行化对数组元素排序

例6-7　使用for 循环初始化数组

public static double[] imperativeInitilize(int size)

    double[] values = new double[size];

    for(int i = 0; i < values.length;i++)

        values[i] = i;

   

    return values;

使用parallelSetAll 方法能轻松地并行化该过程，代码如例6-8 所示。首先提供了一个用于操作的数组，然后传入一个Lambda 表达式，根据数组下标计算元素的值。在该例中，数组下标和元素的值是一样的。使用这些方法有一点要小心：它们改变了传入的数组，而没有创建一个新的数组。

例6-8　使用并行化数组操作初始化数组

public static double[] parallelInitialize(int size)

    double[] values = new double[size];

    Arrays.parallelSetAll(values, i -> i);

    return values;

parallelPrefix 操作擅长对时间序列数据做累加，它会更新一个数组，将每一个元素替换为当前元素和其前驱元素的和，这里的“和”是一个宽泛的概念，它不必是加法，可以是任意一个BinaryOperator。

使用该方法能计算的例子之一是一个简单的滑动平均数。在时间序列上增加一个滑动窗口，计算出窗口中的平均值。如果输入数据为0、1、2、3、4、3.5，滑动窗口的大小为3，则简单滑动平均数为1、2、3、3.5。例6-9 展示了如何计算滑动平均数。

例6-9　计算简单滑动平均数

public static double[] simpleMovingAverage(double[] values, int n)

    double[] sums = Arrays.copyOf(values, values.length); n

    Arrays.parallelPrefix(sums, Double::sum); o

    int start = n - 1;

    return IntStream.range(start, sums.length) p

                    .mapToDouble(i ->

                        double prefix = i == start ? 0 : sums[i - n];

                        return (sums[i] - prefix) / n; q

                    )

                    .toArray(); r

参数n 是时间窗口的大小，我们据此计算滑动平均值。由于要使用的并行操作会改变数组内容，为了不修改原有数据，在➊处复制了一份输入数据。

在➋处执行并行操作，将数组的元素相加。现在sums 变量中保存了求和结果。比如输入0、1、2、3、4、3.5，则计算后的值为0.0、1.0、3.0、6.0、10.0、13.5。

现在有了和，就能计算出时间窗口中的和了，减去窗口起始位置的元素即可，除以n 即得到平均值。可以使用已有的流中的方法计算该值，那就让我们来试试吧！使用Intstream.range得到包含所需元素下标的流。

在➍处使用总和减去窗口起始值，然后再除以n 得到平均值。最后在➎处将流转换为数组。

要点回顾

1、数据并行化是把工作拆分，同时在多核CPU 上执行的方式。

2、如果使用流编写代码，可通过调用 parallel 或者parallelStream 方法实现数据并行化操作。

3、影响性能的五要素是：数据大小、源数据结构、值是否装箱、可用的CPU 核数量，以 及处理每个元素所花的时间。

开发者涨薪指南 48位大咖的思考法则、工作方式、逻辑体系