Java8函数式编程探秘

时间 2021-01-24

标签 html java 算法编程数组并发 app 框架 less ide 栏目 Java 繁體版

原文原文链接

引子##

将行为做为数据传递###

怎样在一行代码里同时计算一个列表的和、最大值、最小值、平均值、元素个数、奇偶分组、指数、排序呢？html

答案是思惟反转！将行为做为数据传递。文艺青年的代码以下所示：java

public class FunctionUtil {

   public static <T,R> List<R> multiGetResult(List<Function<List<T>, R>> functions, List<T> list) {
     return functions.stream().map(f -> f.apply(list)).collect(Collectors.toList());
   }

   public static void main(String[] args) {
     System.out.println(multiGetResult(
         Arrays.asList(
             list -> list.stream().collect(Collectors.summarizingInt(x->x)),
             list -> list.stream().filter(x -> x < 50).sorted().collect(Collectors.toList()),
             list -> list.stream().collect(Collectors.groupingBy(x->(x%2==0? "even": "odd"))),
             list -> list.stream().sorted().collect(Collectors.toList()),
             list -> list.stream().sorted().map(Math::sqrt).collect(Collectors.toMap(x->x, y->Math.pow(2,y)))),
         Arrays.asList(64,49,25,16,9,4,1,81,36)));
   }
}

呃，有点卖弄小聪明。不过要是能将行为做为数据自由传递和施加于数据集产生结果，那么其代码表达能力将如庄子之言，恣意潇洒而无所极限。算法

行为就是数据。编程

三要素###

函数编程的最直接的表现，莫过于将函数做为数据自由传递，结合泛型推导能力，使代码表达能力得到飞通常的提高。那么，Java8是怎么支持函数编程的呢？主要有三个核心概念：数组

函数接口(Function)
流(Stream)
聚合器(Collector)

三者的关联是：流（Stream）经过函数接口（Function）进行过滤和转换，最后经过聚合器（Collector）对流中的元素进行聚合操做，获得最终结果。

并发

函数接口##

关于函数接口，须要记住的就是两件事：app

函数接口是行为的抽象；
函数接口是数据转换器。

最直接的支持就是 java.util.Function 包。定义了四个最基础的函数接口：框架

Supplier<T>: 数据提供器，能够提供 T 类型对象；无参的构造器，提供了 get 方法；
Function<T,R>: 数据转换器，接收一个 T 类型的对象，返回一个 R类型的对象；单参数单返回值的行为接口；提供了 apply, compose, andThen, identity 方法；
Consumer<T>: 数据消费器，接收一个 T类型的对象，无返回值，一般用于根据T对象作些处理；单参数无返回值的行为接口；提供了 accept, andThen 方法；
Predicate<T>: 条件测试器，接收一个 T 类型的对象，返回布尔值，一般用于传递条件函数；单参数布尔值的条件性接口。提供了 test (条件测试) , and-or- negate(与或非) 方法。

其中, compose, andThen, and, or, negate 用来组合函数接口而获得更强大的函数接口。less

其它的函数接口都是经过这四个扩展而来。ide

在参数个数上扩展：好比接收双参数的，有 Bi 前缀，好比 BiConsumer<T,U>, BiFunction<T,U,R> ;
在类型上扩展：好比接收原子类型参数的，有 [Int|Double|Long][Function|Consumer|Supplier|Predicate]
特殊经常使用的变形：好比 BinaryOperator ，是同类型的双参数 BiFunction<T,T,T> ，二元操做符； UnaryOperator 是 Function<T,T> 一元操做符。

那么，这些函数接口能够接收哪些值呢？

类/对象的静态方法引用、实例方法引用。引用符号为双冒号 ::
类的构造器引用，好比 Class::new
lambda表达式

在博文“使用函数接口和枚举实现配置式编程(Java与Scala实现)”, “精练代码：一次Java函数式编程的重构之旅” 给出了基本的例子。后面还有更多例子。重在练习和尝试。

聚合器

每个流式计算的末尾总有一个相似 collect(Collectors.toList()) 的方法调用。 Collectors.toList() 会返回一个聚合器 Collector 。

聚合器 Collector 的功能是将指定的数据流根据指定的能力聚合成最终结果。 聚合器是多个函数接口能力的组合，体现了函数编程的精要。 固然，聚合器实现也会相对复杂一点，要细细揣摩。

Reduce

在深刻聚合器的内部实现以前，了解下 Reduce 是合适的。 Reduce 是一个推导过程, 其算法以下：

STEP1: 初始化结果 R = init ；

STEP2: 给定一个值集 S。每次从 S 中取出一个值 v，经过二元操做符 op 施加到 R 和 v ，产生一个新值赋给 R = BinaryOperator(R, v)；重复 STEP2，直到 S 中没有值可取为止。

以下代码所示：S = list , op = biFunc ，R = result。

public static <E,T> T reduce(List<E> list, BiFunction<E,T,T> biFunc, Supplier<T> init) {
    T result = init.get();
    for (E e: list) {
      result = biFunc.apply(e, result);
    }
    return result;
  }

Collector

来看看 Collector 的主要定义：

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

Collector 与 Reduce 有不少类似之处：有一个初始值提供器 init = supplier ; 有一个累积操做器 accumulator = op ；有一个合并器 combiner ；有一个终值转换器 finisher 。比 Reduce 多出了两样东西： combiner 和 finisher 。

理解 Collector 定义要注意的是，泛型参数在方法参数中的顺序。 A 是值提供器的类型，是累积操做的左参数，是合并操做的类型，也是中间结果的类型； T 是从某个 Stream 中取出的值的类型；R 是终值的类型。显然 A 是一个承前启后的核心类型。

看函数式代码时，每每容易被各类泛型参数弄得很糊涂。但函数式编程加上泛型，才能使代码的表达能力突破类型限制，提高到很是灵活的程度。

聚合器实现

Collectors 里提供了多种 Collector 的实现。 Collector 大体能够划分为四类：列表类、统计类、映射类、自定义。

列表类聚合器

列表类 Collector 一般将 Stream of Collection 中的元素生成 Collection、List 或 Set 。来看 toList 的实现：

public static <T>
    Collector<T, ?, List<T>> toList() {
        return new CollectorImpl<>((Supplier<List<T>>) ArrayList::new, List::add,
                                   (left, right) -> { left.addAll(right); return left; },
                                   CH_ID);
    }

仍是比较容易看懂的：

A 类型是 List ；
值提供器 supplier = ArrayList::new ，建立一个空的结果列表；
值累加器 accumulator = (list,e) -> list.add(e) ; 将取出的值加入到结果列表中；
合并器 combiner = (left, right) -> { left.addAll(right); return left; }
终值转换器 finisher = castingIdentity() = i -> (R) i (默认给出的)

它的返回值 Collector<T, ?, List > 的含义是：List 是结果列表；T 是每次从值集中取出的类型 T ；中间运算结果 ? 是不肯定类型的。

统计类聚合器

统计类聚合器一般生成单个值，主要包括 minBy,maxBy, counting, summing, averaging 等，基于 reducing 来实现。

来看看 reducing 的库实现。记住 Collectors.reducing 的 A 类型是 OptionalBox ，实际上就是上面的 T result 的封装。present 用来处理首值赋值的问题。

public static <T> Collector<T, ?, Optional<T>>
    reducing(BinaryOperator<T> op) {
        class OptionalBox implements Consumer<T> {
            T value = null;
            boolean present = false;

            @Override
            public void accept(T t) {
                if (present) {
                    value = op.apply(value, t);
                }
                else {
                    value = t;
                    present = true;
                }
            }
        }

        return new CollectorImpl<T, OptionalBox, Optional<T>>(
                OptionalBox::new, OptionalBox::accept,
                (a, b) -> { if (b.present) a.accept(b.value); return a; },
                a -> Optional.ofNullable(a.value), CH_NOID);
    }

映射类聚合器

映射类聚合器，一般是将一个 Stream 聚合成 Map<K, U> 或者 Map<K, List > 。

看 toMap 的实现：这里提供了重载方法。

简单形式是只有 keyMapper, valueMapper 两个转换函数，最终的 Map<K,U> = [K=keyMapper.apply(T), U=valueMapper.apply(T)] ；初始值提供器默认 mapSupplier = HashMap::new。

彻底形式是提供了 Collector 的四要素。彻底形式的含义是：

STEP1: 先用简单形式的 keyMapper, valueMapper 两个转换函数，将指定流转换成 first = Map<K,U>;

STEP2: 合并 first 与 mapSupplier 。合并的方法是，对于每个 key 对应的 firstValue = first[key], supplierValue = mapSupplier[key] ， finalValue = mergeFunction(supplierValue,firstValue)

public static <T, K, U>
    Collector<T, ?, Map<K,U>> toMap(Function<? super T, ? extends K> keyMapper,
                                    Function<? super T, ? extends U> valueMapper) {
        return toMap(keyMapper, valueMapper, throwingMerger(), HashMap::new);
}

public static <T, K, U, M extends Map<K, U>>
    Collector<T, ?, M> toMap(Function<? super T, ? extends K> keyMapper,
                                Function<? super T, ? extends U> valueMapper,
                                BinaryOperator<U> mergeFunction,
                                Supplier<M> mapSupplier) {
        BiConsumer<M, T> accumulator
                = (map, element) -> map.merge(keyMapper.apply(element),
                                              valueMapper.apply(element), mergeFunction);
        return new CollectorImpl<>(mapSupplier, accumulator, mapMerger(mergeFunction), CH_ID);
    }

写个示例来理解下。先建立一个 Person 列表，而后经过 toMap 的简单形式能够建立 Map<name, age>；使用 toMap 的彻底形式时，只须要多提供一个已有的 anotherPersonAgeMap = Map<name, age> ，而后建立一个 valueMerge = (age1, age2) -> age1 ，当有年龄重合时，用 MapSupplier 的 age 覆盖。

public class CollectorsToMapDemo {

  public static void main(String[]args) {
    List<Person> persons = Arrays.asList(new Person("qin", 32), new Person("ni", 24));
    Map<String, Integer> personAgeMap = persons.stream().collect(Collectors.toMap(
        Person::getName, Person::getAge
    ));
    System.out.println("personAgeMap: " + personAgeMap);

    List<Person> anotherPersons = Arrays.asList(new Person("su", 24), new Person("ni", 25));
    Map<String, Integer> anotherPersonAgeMap = anotherPersons.stream().collect(Collectors.toMap(
        Person::getName, Person::getAge
    ));

    Map<String,Integer> merged = persons.stream().collect(Collectors.toMap(
        Person::getName, Person::getAge, (age1, age2) -> age1, () -> anotherPersonAgeMap
        ));
    System.out.println("merged: " + merged);
  }

}

@AllArgsConstructor
@Data
class Person {
  private String name;
  private Integer age;
}

输出结果：
personAgeMap: {qin=32, ni=24}
merged: {su=24, qin=32, ni=25}

分析 toMap 获得的启发是：从简单形式着手，更容易理解其原理。复杂形式，每每是在某一方面对简单形式进行了通常化而获得的。

再来看 groupingby 的实现。所涉及的泛型更加眼花缭乱，居然有 T,K,D,A,M 这么多类型！

理一理：

T 是 Stream 中的数据的类型；
K 是 classifier.apply(T) 获得的类型，生成的结果 Map 的 key 的类型；
D 是生成的结果 Map 的 value 的类型；
M 是结果 Map 的类型；
A 是中间结果类型，无限制。

从第二个实现看起，会更容易理解一点。首先，classifier 函数用来生成 key ，接着 downstream 应用于 Stream 生成 value 。好比，downstream = toList() ， value = List ; downstream = toMap , value = Map<K,U>。

public static <T, K> Collector<T, ?, Map<K, List<T>>>
    groupingBy(Function<? super T, ? extends K> classifier) {
        return groupingBy(classifier, toList());
    }

public static <T, K, A, D>
    Collector<T, ?, Map<K, D>> groupingBy(Function<? super T, ? extends K> classifier,
                                          Collector<? super T, A, D> downstream) {
        return groupingBy(classifier, HashMap::new, downstream);
    }

public static <T, K, D, A, M extends Map<K, D>>
    Collector<T, ?, M> groupingBy(Function<? super T, ? extends K> classifier,
                                  Supplier<M> mapFactory,
                                  Collector<? super T, A, D> downstream) {
        Supplier<A> downstreamSupplier = downstream.supplier();
        BiConsumer<A, ? super T> downstreamAccumulator = downstream.accumulator();
        BiConsumer<Map<K, A>, T> accumulator = (m, t) -> {
            K key = Objects.requireNonNull(classifier.apply(t), "element cannot be mapped to a null key");
            A container = m.computeIfAbsent(key, k -> downstreamSupplier.get());
            downstreamAccumulator.accept(container, t);
        };
        BinaryOperator<Map<K, A>> merger = Collectors.<K, A, Map<K, A>>mapMerger(downstream.combiner());
        @SuppressWarnings("unchecked")
        Supplier<Map<K, A>> mangledFactory = (Supplier<Map<K, A>>) mapFactory;

        if (downstream.characteristics().contains(Collector.Characteristics.IDENTITY_FINISH)) {
            return new CollectorImpl<>(mangledFactory, accumulator, merger, CH_ID);
        }
        else {
            // code...
            return new CollectorImpl<>(mangledFactory, accumulator, merger, finisher, CH_NOID);
        }
    }

接下来，会用一个实际例子来讲明其用法。

恭喜你！能坚持读到这里，已是一种胜利。

实际例子

一种经常使用场景，是根据一个对象里的多个字段进行分组。好比，假设一个企业有多个部门(department)，每一个部门都有各类职务(position)的员工(Employee)。如今，要统计每一个部门下的每种职务的员工姓名。其结果形式是： Map<department, Map<position, List<name>>>> groupedEmployees.

实现代码以下所示。

public class CollectorsGroupingbyDemo {

  public static void main(String[]args) {

    List<Employee> employList = Arrays.asList(
        new Employee("su", "mid", "engine"),
        new Employee("lan", "mid", "prod"),
        new Employee("qin", "data", "engine"),
        new Employee("yu", "mid", "engine"),
        new Employee("ming", "data", "engine")
    );


    // Map[department, Map[position, List[name]]]
    Map<String, Map<String, List<String>>> groupedEmployees =
        employList.stream().collect(
            Collectors.groupingBy(Employee::getDepartment,
                Collectors.groupingBy(Employee::getPosition, new EmployNameListCollector())
        ));

    System.out.println("groupedEmployees: " + groupedEmployees);

  }

}

class EmployNameListCollector implements Collector<Employee,List<String>,List<String>> {

  @Override
  public Supplier<List<String>> supplier() {
    return () -> new ArrayList<>();
  }

  @Override
  public BiConsumer<List<String>, Employee> accumulator() {
    return (list, e) -> list.add(e.getName());
  }

  @Override
  public BinaryOperator<List<String>> combiner() {
    return (list1, list2) -> { list1.addAll(list2); return list1; };
  }

  @Override
  public Function<List<String>, List<String>> finisher() {
    return i->i;
  }

  @Override
  public Set<Characteristics> characteristics() {
    return Collections.emptySet();
  }
}


@AllArgsConstructor
@Data
class Employee {
  private String name;
  private String department;
  private String position;

}

解读以下：

STEP1: 首先根据 department 分组。使用 groupingby(Employee::getDepartment, positionEmployeeMapCollector) ；须要实现 positionEmployeeMapCollector;

STEP2: 如今获得的是 Stream<Employee> 。根据 position 分组，使用 Collectors.groupingBy(Employee::getPosition, employNameListCollector) ，须要实现 employNameListCollector ；

STEP3：如今获得的是 Stream<Employee> , 要获得 List<String> 。显然，若是要获得 List<Employee> ，只须要使用 Collectors.toList() 便可；可是如今要拿到 List<String>。能够仿照 Collectors.toList() 的实现，自定义一个 EmployNameListCollector 。 EmployNameListCollector 与 Collectors.toList() 的区别仅在于要将 employee.getName() 加到 list 。其它的几乎同样。

经过编写自定义的 Collector ，能够加深对 Collector 的理解。

流##

流（Stream）是Java8对函数式编程的重要支撑。大部分函数式工具都围绕Stream展开。

Stream的接口###

Stream 主要有四类接口：

流到流之间的转换：好比 filter(过滤), map(映射转换), mapTo[Int|Long|Double] (到原子类型流的转换), flatMap(高维结构平铺)，flatMapTo[Int|Long|Double], sorted(排序)，distinct(不重复值)，peek(执行某种操做，流不变，可用于调试)，limit(限制到指定元素数量), skip(跳过若干元素) ；
流到终值的转换：好比 toArray（转为数组），reduce（推导结果），collect（聚合结果），min(最小值), max(最大值), count (元素个数)， anyMatch (任一匹配), allMatch(全部都匹配)， noneMatch(一个都不匹配)， findFirst（选择首元素），findAny(任选一元素) ；
直接遍历： forEach (不保序遍历，好比并行流), forEachOrdered（保序遍历) ；
构造流： empty (构造空流)，of (单个元素的流及多元素顺序流)，iterate (无限长度的有序顺序流)，generate (将数据提供器转换成无限非有序的顺序流)， concat (流的链接)， Builder (用于构造流的Builder对象)

流的类型主要有：Reference(对象流)， IntStream (int元素流), LongStream (long元素流)， Double (double元素流) ，定义在类 StreamShape 中，主要将操做适配于类型系统。

flatMap 的一个例子见以下所示，将一个二维数组转换为一维数组：

List<Integer> nums = Arrays.asList(Arrays.asList(1,2,3), Arrays.asList(1,4,9), Arrays.asList(1,8,27))
                                .stream().flatMap(x -> x.stream()).collect(Collectors.toList());
     System.out.println(nums);

collector实现###

这里咱们仅分析串行是怎么实现的。入口在类 java.util.stream.ReferencePipeline 的 collect 方法：

container = evaluate(ReduceOps.makeRef(collector));
return collector.characteristics().contains(Collector.Characteristics.IDENTITY_FINISH)
          ? (R) container : collector.finisher().apply(container);

这里的关键是 ReduceOps.makeRef(collector)。点进去：

public static <T, I> TerminalOp<T, I>
    makeRef(Collector<? super T, I, ?> collector) {
        Supplier<I> supplier = Objects.requireNonNull(collector).supplier();
        BiConsumer<I, ? super T> accumulator = collector.accumulator();
        BinaryOperator<I> combiner = collector.combiner();
        class ReducingSink extends Box<I>
                implements AccumulatingSink<T, I, ReducingSink> {
            @Override
            public void begin(long size) {
                state = supplier.get();
            }

            @Override
            public void accept(T t) {
                accumulator.accept(state, t);
            }

            @Override
            public void combine(ReducingSink other) {
                state = combiner.apply(state, other.state);
            }
        }
        return new ReduceOp<T, I, ReducingSink>(StreamShape.REFERENCE) {
            @Override
            public ReducingSink makeSink() {
                return new ReducingSink();
            }

            @Override
            public int getOpFlags() {
                return collector.characteristics().contains(Collector.Characteristics.UNORDERED)
                       ? StreamOpFlag.NOT_ORDERED
                       : 0;
            }
        };
    }

private static abstract class Box<U> {
        U state;

        Box() {} // Avoid creation of special accessor

        public U get() {
            return state;
        }
    }

Box 是一个结果值的持有者； ReducingSink 用begin, accept, combine 三个方法定义了要进行的计算；ReducingSink是有状态的流数据消费的计算抽象，阅读Sink接口文档可知。ReduceOps.makeRef(collector) 返回了一个封装了Reduce操做的ReduceOps对象。注意到，这里都是声明要执行的计算，而不涉及计算的实际过程。展现了表达与执行分离的思想。真正的计算过程启动在 ReferencePipeline.evaluate 方法里:

final <R> R evaluate(TerminalOp<E_OUT, R> terminalOp) {
        assert getOutputShape() == terminalOp.inputShape();
        if (linkedOrConsumed)
            throw new IllegalStateException(MSG_STREAM_LINKED);
        linkedOrConsumed = true;

        return isParallel()
               ? terminalOp.evaluateParallel(this, sourceSpliterator(terminalOp.getOpFlags()))
               : terminalOp.evaluateSequential(this, sourceSpliterator(terminalOp.getOpFlags()));
    }

使用 IDE 的 go to implementations 功能，跟进去，能够发现，最终在 AbstractPipeLine 中定义了：

@Override
    final <P_IN> void copyInto(Sink<P_IN> wrappedSink, Spliterator<P_IN> spliterator) {
        Objects.requireNonNull(wrappedSink);

        if (!StreamOpFlag.SHORT_CIRCUIT.isKnown(getStreamAndOpFlags())) {
            wrappedSink.begin(spliterator.getExactSizeIfKnown());
            spliterator.forEachRemaining(wrappedSink);
            wrappedSink.end();
        }
        else {
            copyIntoWithCancel(wrappedSink, spliterator);
        }
    }

Spliterator 用来对流中的元素进行分区和遍历以及施加Sink指定操做，能够用于并发计算。Spliterator的具体实现类定义在 Spliterators 的静态类和静态方法中。其中有：

数组Spliterator:
static final class ArraySpliterator<T> implements Spliterator<T>
static final class IntArraySpliterator implements Spliterator.OfInt
static final class LongArraySpliterator implements Spliterator.OfLong
static final class DoubleArraySpliterator implements Spliterator.OfDouble

迭代Spliterator:
static class IteratorSpliterator<T> implements Spliterator<T>
static final class IntIteratorSpliterator implements Spliterator.OfInt
static final class LongIteratorSpliterator implements Spliterator.OfLong
static final class DoubleIteratorSpliterator implements Spliterator.OfDouble

抽象Spliterator:
public static abstract class AbstractSpliterator<T> implements Spliterator<T>
private static abstract class EmptySpliterator<T, S extends Spliterator<T>, C>
public static abstract class AbstractIntSpliterator implements Spliterator.OfInt
public static abstract class AbstractLongSpliterator implements Spliterator.OfLong
public static abstract class AbstractDoubleSpliterator implements Spliterator.OfDouble

每一个具体类都实现了trySplit，forEachRemaining，tryAdvance，estimateSize，characteristics， getComparator。 trySplit 用于拆分流，提供并发能力；forEachRemaining，tryAdvance 用于遍历和消费流中的数据。下面展现了IteratorSpliterator的forEachRemaining，tryAdvance 两个方法的实现。能够看到，木有特别的地方，就是遍历元素并将指定操做施加于元素。

@Override
        public void forEachRemaining(Consumer<? super T> action) {
            if (action == null) throw new NullPointerException();
            Iterator<? extends T> i;
            if ((i = it) == null) {
                i = it = collection.iterator();
                est = (long)collection.size();
            }
            i.forEachRemaining(action);
        }

        @Override
        public boolean tryAdvance(Consumer<? super T> action) {
            if (action == null) throw new NullPointerException();
            if (it == null) {
                it = collection.iterator();
                est = (long) collection.size();
            }
            if (it.hasNext()) {
                action.accept(it.next());
                return true;
            }
            return false;
        }

总体流程就是这样。回顾一下：

Collector 定义了必要的聚合操做函数；
ReduceOps.makeRef 将 Collector 封装成一个计算对象 ReduceOps ，依赖的 ReducingSink 定义了具体的流数据消费过程；
Spliterator 用于对流中的元素进行分区和遍历以及施加Sink指定的操做。

Pipeline###

那么，Spliterator 又是从哪里来的呢？是经过类 java.util.stream.AbstractPipeline 的方法 sourceSpliterator 拿到的：

private Spliterator<?> sourceSpliterator(int terminalFlags) {
        // Get the source spliterator of the pipeline
        Spliterator<?> spliterator = null;
        if (sourceStage.sourceSpliterator != null) {
            spliterator = sourceStage.sourceSpliterator;
            sourceStage.sourceSpliterator = null;
        }
        else if (sourceStage.sourceSupplier != null) {
            spliterator = (Spliterator<?>) sourceStage.sourceSupplier.get();
            sourceStage.sourceSupplier = null;
        }
        else {
            throw new IllegalStateException(MSG_CONSUMED);
        }
        // code for isParallel
       return spliterator;
}

这里的 sourceStage 是一个 AbstractPipeline。 Pipeline 是实现流式计算的流水线抽象，也是Stream的实现类。能够看到，java.util.stream 定义了四种 pipeline: DoublePipeline, IntPipeline, LongPipeline, ReferencePipeline。能够重点看 ReferencePipeline 的实现。好比 filter, map

abstract class ReferencePipeline<P_IN, P_OUT>
        extends AbstractPipeline<P_IN, P_OUT, Stream<P_OUT>>
        implements Stream<P_OUT>

@Override
    public final Stream<P_OUT> filter(Predicate<? super P_OUT> predicate) {
        Objects.requireNonNull(predicate);
        return new StatelessOp<P_OUT, P_OUT>(this, StreamShape.REFERENCE,
                                     StreamOpFlag.NOT_SIZED) {
            @Override
            Sink<P_OUT> opWrapSink(int flags, Sink<P_OUT> sink) {
                return new Sink.ChainedReference<P_OUT, P_OUT>(sink) {
                    @Override
                    public void begin(long size) {
                        downstream.begin(-1);
                    }

                    @Override
                    public void accept(P_OUT u) {
                        if (predicate.test(u))
                            downstream.accept(u);
                    }
                };
            }
        };
    }

    @Override
    @SuppressWarnings("unchecked")
    public final <R> Stream<R> map(Function<? super P_OUT, ? extends R> mapper) {
        Objects.requireNonNull(mapper);
        return new StatelessOp<P_OUT, R>(this, StreamShape.REFERENCE,
                                     StreamOpFlag.NOT_SORTED | StreamOpFlag.NOT_DISTINCT) {
            @Override
            Sink<P_OUT> opWrapSink(int flags, Sink<R> sink) {
                return new Sink.ChainedReference<P_OUT, R>(sink) {
                    @Override
                    public void accept(P_OUT u) {
                        downstream.accept(mapper.apply(u));
                    }
                };
            }
        };
    }

套路基本同样，关键点在于 accept 方法。filter 只在知足条件时将值传给下一个 pipeline, 而 map 将计算的值传给下一个 pipeline. StatelessOp 没有什么逻辑，JDK文档解释是：Base class for a stateless intermediate stage of a Stream。相应还有一个 StatefulOp, Head。这些都是 ReferencePipeline ，负责将值在 pipeline 之间传递，交给 Sink 去计算。

static class Head<E_IN, E_OUT> extends ReferencePipeline<E_IN, E_OUT>
abstract static class StatelessOp<E_IN, E_OUT> extends ReferencePipeline<E_IN, E_OUT>
abstract static class StatefulOp<E_IN, E_OUT> extends ReferencePipeline<E_IN, E_OUT>

至此，咱们对整个流计算过程有了更清晰的认识。细节能够再逐步推敲。

函数式编程的益处##

更精练的代码###

函数编程的一大益处，是用更精练的代码表达经常使用数据处理模式。函数接口可以轻易地实现模板方法模式，只要将不肯定的业务逻辑抽象成函数接口，而后传入不一样的lambda表达式便可。博文“精练代码：一次Java函数式编程的重构之旅” 展现了如何使用函数式编程来重构常见代码，萃取更多可复用的代码模式。

这里给出一个列表分组的例子。实际应用经常须要将一个列表 List[T] 转换为一个 Map[K, List[T]] ，其中 K 是经过某个函数来实现的。看下面一段代码：

public static Map<String, List<OneRecord>> buildRecordMap(List<OneRecord> records, List<String> colKeys) {
    Map<String, List<OneRecord>> recordMap = new HashMap<>();
    records.forEach(
        record -> {
          String recordKey = buildRecordKey(record.getFieldValues(), colKeys);
          if (recordMap.get(recordKey) == null) {
            recordMap.put(recordKey, new ArrayList<OneRecord>());
          }
          recordMap.get(recordKey).add(record);
    });
    return recordMap;
  }

可使用 Collectors.groupingby 来简洁地实现：

public static Map<String, List<OneRecord>> buildRecordMapBrief(List<OneRecord> records, List<String> colKeys) {
    return records.stream().collect(Collectors.groupingBy(
        record -> buildRecordKey(record.getFieldValues(), colKeys)
    ));
  }

不少经常使用数据处理算法，均可以使用函数式编程的流式计算简洁表达。

更通用的代码###

使用函数接口，结合泛型，很容易用精练的代码，写出很是通用的工具方法。实际应用中，经常会有这样的需求：有两个对象列表srcList和destList，两个对象类型的某个字段K具备相同的值；须要根据这个相同的值合并对应的两个对象的信息。

这里给出了一个列表合并函数，能够将一个对象列表合并到指定的对象列表中。实现是：先将待合并的列表srcList根据key值函数keyFunc构建起srcMap,而后遍历dest列表的对象R，将待合并的信息srcMap[key]及T经过合并函数mergeFunc生成的新对象R添加到最终结果列表。

public static <K,R> List<R> mergeList(List<R> srcList, List<R> destList ,
                                        Function<R,K> keyFunc,
                                        BinaryOperator<R> mergeFunc) {
    return mergeList(srcList, destList, keyFunc, keyFunc, mergeFunc);
  }

  public static <T,S,K,R> List<R> mergeList(List<S> srcList, List<T> destList ,
                                            Function<S,K> skeyFunc, Function<T,K> dkeyFunc,
                                            BiFunction<S,T,R> mergeFunc) {

    Map<K,S> srcMap = srcList.stream().collect(Collectors.toMap(skeyFunc, s -> s, (k1,k2) -> k1));
    return destList.stream().map(
        dest -> {
          K key = dkeyFunc.apply(dest);
          S src = srcMap.get(key);
          return mergeFunc.apply(src, dest);
        }
    ).collect(Collectors.toList());

  }

更可测的代码###

使用函数接口能够方便地隔离外部依赖，使得类和对象的方法更纯粹、更具可测性。博文“使用Java函数接口及lambda表达式隔离和模拟外部依赖更容易滴单测”，“改善代码可测性的若干技巧”集中讨论了如何使用函数接口提高代码的可单测性。

组合的力量###

函数编程的强大威力，在于将函数接口组合起来，构建更强大更具备通用性的实用工具方法。超越类型，超越操做与数据的边界。

前面提到，函数接口就是数据转换器。好比Function<T,R> 就是“将T对象转换成R对象的行为或数据转换器”。对于实际工程应用的普通级函数编程足够了。不过，要玩转函数接口，就要升级下认识。好比 Function<BiFunction<S,Q,R>, Function<T,R>> 该怎么理解呢？这是“一个一元函数g(h(s,q)) ，参数指定的二元函数h(s,q)应用于指定的两个参数S,Q，获得一个一元函数f(t)，这个函数接收一个T对象，返回一个R对象”。以下代码所示：

public static <T,S,Q,R> Function<BiFunction<S,Q,R>, Function<T,R>> op(Function<T,S> funcx, Function<T,Q> funcy) {
    return opFunc -> aT -> opFunc.apply(funcx.apply(aT), funcy.apply(aT));
  }

  System.out.println(op(x-> x.toString().length(), y-> y+",world").apply((x,y) -> x+" " +y).apply("hello"));

实现的是 h(t) = h(funx(t), funy(t)) ，h(x,y) 是一个双参数函数。

“Java函数接口实现函数组合及装饰器模式” 展现了如何使用极少许的代码实现装饰器模式，将简单的函数接口组合成更强大功能的复合函数接口。

来看上面的 public static <T,S,K,R> List<R> mergeList(List<S> srcList, List<T> destList , Function<S,K> skeyFunc, Function<T,K> dkeyFunc,BiFunction<S,T,R> mergeFunc) ，通用性虽好，但是有5个参数，有点丑。怎么改造下呢？看实现，主要包含两步：1. 将待合并列表转化为 srcMap: map<K,S>; 2. 使用指定的函数 dKeyFunc, mergeFunc 做用于destList和srcMap，获得最终结果。能够改写代码以下：

public static <T,S,K,R> List<R> mergeList(List<S> srcList, List<T> destList ,
                                          Function<S,K> skeyFunc, Function<T,K> dkeyFunc,
                                          BiFunction<S,T,R> mergeFunc) {
    return join(destList, mapKey(srcList, skeyFunc)).apply(dkeyFunc, (BiFunction) mergeFunc);

  }

  public static <T,K> Map<K,T> mapKey(List<T> list, Function<T,K> keyFunc) {
    return list.stream().collect(Collectors.toMap(keyFunc, t -> t, (k1,k2) -> k1));
  }

  public static <T,S,K,R> BiFunction<Function<T,K>, BiFunction<S,T,R>, List<R>> join(List<T> destList, Map<K,S> srcMap) {
    return (dkeyFunc,mergeFunc) -> destList.stream().map(
        dest -> {
          K key = dkeyFunc.apply(dest);
          S src = srcMap.get(key);
          return mergeFunc.apply(src, dest);
        }).collect(Collectors.toList());
  }

System.out.println(mergeList(Arrays.asList(1,2), Arrays.asList("an", "a"), s-> s, t-> t.toString().length(), (s,t) -> s+t));

mapKey 是一个通用函数，用于将一个 list 按照指定的 keyFunc 转成一个 Map; join 函数接受一个 list 和待合并的 srcMap, 返回一个二元函数，该函数使用指定的 dkeyFunc 和 mergeFunc 来合并指定数据获得最终的结果列表。这可称之为“延迟指定行为”。如今, mapKey 和 join 都是通用性函数。Amazing ！

完整代码示例##

package zzz.study.function;

import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.function.BiFunction;
import java.util.function.BinaryOperator;
import java.util.function.Function;
import java.util.function.Supplier;
import java.util.stream.Collectors;

/**
 * Created by shuqin on 17/12/3.
 */
public class FunctionUtil {

  public static <T,R> List<R> multiGetResult(List<Function<List<T>, R>> functions, List<T> list) {
    return functions.stream().map(f -> f.apply(list)).collect(Collectors.toList());
  }

  public static <K,R> List<R> mergeList(List<R> srcList, List<R> destList ,
                                        Function<R,K> keyFunc,
                                        BinaryOperator<R> mergeFunc) {
    return mergeList(srcList, destList, keyFunc, keyFunc, mergeFunc);
  }

  public static <T,S,K,R> List<R> mergeList(List<S> srcList, List<T> destList ,
                                          Function<S,K> skeyFunc, Function<T,K> dkeyFunc,
                                          BiFunction<S,T,R> mergeFunc) {
    return join(destList, mapKey(srcList, skeyFunc)).apply(dkeyFunc, (BiFunction) mergeFunc);

  }

  public static <T,K> Map<K,T> mapKey(List<T> list, Function<T,K> keyFunc) {
    return list.stream().collect(Collectors.toMap(keyFunc, t -> t, (k1,k2) -> k1));
  }

  public static <T,S,K,R> BiFunction<Function<T,K>, BiFunction<S,T,R>, List<R>> join(List<T> destList, Map<K,S> srcMap) {
    return (dkeyFunc,mergeFunc) -> destList.stream().map(
        dest -> {
          K key = dkeyFunc.apply(dest);
          S src = srcMap.get(key);
          return mergeFunc.apply(src, dest);
        }).collect(Collectors.toList());
  }

  /** 对给定的值 x,y 应用指定的二元操做函数 */
  public static <T,S,R> Function<BiFunction<T,S,R>, R> op(T x, S y) {
    return opFunc -> opFunc.apply(x, y);
  }

  /** 将两个函数使用组合成一个函数，这个函数接受一个二元操做函数 */
  public static <T,S,Q,R> Function<BiFunction<S,Q,R>, R> op(Function<T,S> funcx, Function<T,Q> funcy, T x) {
    return opFunc -> opFunc.apply(funcx.apply(x), funcy.apply(x));
  }

  public static <T,S,Q,R> Function<BiFunction<S,Q,R>, Function<T,R>> op(Function<T,S> funcx, Function<T,Q> funcy) {
    return opFunc -> aT -> opFunc.apply(funcx.apply(aT), funcy.apply(aT));
  }

  /** 将两个函数组合成一个叠加函数, compose(f,g) = f(g) */
  public static <T> Function<T, T> compose(Function<T,T> funcx, Function<T,T> funcy) {
    return x -> funcx.apply(funcy.apply(x));
  }

  /** 将若干个函数组合成一个叠加函数, compose(f1,f2,...fn) = f1(f2(...(fn))) */
  public static <T> Function<T, T> compose(Function<T,T>... extraFuncs) {
    if (extraFuncs == null || extraFuncs.length == 0) {
      return x->x;
    }
    return x -> Arrays.stream(extraFuncs).reduce(y->y, FunctionUtil::compose).apply(x);
  }

   public static void main(String[] args) {
     System.out.println(multiGetResult(
         Arrays.asList(
             list -> list.stream().collect(Collectors.summarizingInt(x->x)),
             list -> list.stream().filter(x -> x < 50).sorted().collect(Collectors.toList()),
             list -> list.stream().collect(Collectors.groupingBy(x->(x%2==0? "even": "odd"))),
             list -> list.stream().sorted().collect(Collectors.toList()),
             list -> list.stream().sorted().map(Math::sqrt).collect(Collectors.toMap(x->x, y->Math.pow(2,y)))),
         Arrays.asList(64,49,25,16,9,4,1,81,36)));

     List<Integer> list = Arrays.asList(1,2,3,4,5);
     Supplier<Map<Integer,Integer>> mapSupplier = () -> list.stream().collect(Collectors.toMap(x->x, y-> y * y));

     Map<Integer, Integer> mapValueAdd = list.stream().collect(Collectors.toMap(x->x, y->y, (v1,v2) -> v1+v2, mapSupplier));
     System.out.println(mapValueAdd);

     List<Integer> nums = Arrays.asList(Arrays.asList(1,2,3), Arrays.asList(1,4,9), Arrays.asList(1,8,27))
                                .stream().flatMap(x -> x.stream()).collect(Collectors.toList());
     System.out.println(nums);

     List<Integer> fibo = Arrays.asList(1,2,3,4,5,6,7,8,9,10).stream().collect(new FiboCollector());
     System.out.println(fibo);

     System.out.println(op(new Integer(3), Integer.valueOf(3)).apply((x,y) -> x.equals(y.toString())));

     System.out.println(op(x-> x.length(), y-> y+",world", "hello").apply((x,y) -> x+" " +y));

     System.out.println(op(x-> x, y-> y+",world").apply((x,y) -> x+" " +y).apply("hello"));

     System.out.println(op(x-> x.toString().length(), y-> y+",world").apply((x,y) -> x+" " +y).apply("hello"));

     System.out.println(mergeList(Arrays.asList(1,2), Arrays.asList("an", "a"),
                                  s-> s, t-> t.toString().length(), (s,t) -> s+t));

   }

}

小结##

本文深刻学习了Java8函数式编程框架：Function&Stream&Collector，并展现了函数式编程在实际应用中所带来的诸多益处。函数式编程是一把大锋若钝的奇剑。基于函数接口编程，将函数做为数据自由传递，结合泛型推导能力，可编写出精练、通用、易测的代码，使代码表达能力得到飞通常的提高。

函数式编程分享