Java8函數式編程探祕
引子##
將行爲做爲數據傳遞###
怎樣在一行代碼裏同時計算一個列表的和、最大值、最小值、平均值、元素個數、奇偶分組、指數、排序呢？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
統計類聚合器
統計類聚合器一般生成單個值，主要包括 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
看 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
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對象)
除了 Stream 自己自帶的生成Stream 的方法，數組和容器及StreamSupport都有轉換爲流的方法。好比 Arrays.stream , [List|Set|Collection].[stream|parallelStream] , StreamSupport.[int|long|double|]stream；
流的類型主要有：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，並展現了函數式編程在實際應用中所帶來的諸多益處。函數式編程是一把大鋒若鈍的奇劍。基於函數接口編程，將函數做爲數據自由傳遞，結合泛型推導能力，可編寫出精練、通用、易測的代碼，使代碼表達能力得到飛通常的提高。
- 1. java8的函數式編程
- 2. java8函數式編程
- 3. Java8函數式編程
- 4. Java8的函數式編程
- 5. java8 函數式編程一
- 6. java8函數式編程（一）
- 7. JAVA8函數式編程
- 8. java8 函數式編程
- 9. 深刻探尋JAVA8 part1：函數式編程與Lambda表達式
- 10. Java8-Lambda表達式（函數式編程）
- 更多相關文章...
- • Rust 併發編程 - RUST 教程
- • 探索Redis事務回滾 - Redis教程
- • Flink 數據傳輸及反壓詳解
- • 委託模式
-
每一个你不满意的现在，都有一个你没有努力的曾经。
- 1. Duang!超快Wi-Fi來襲
- 2. 機器學習-補充03 神經網絡之**函數(Activation Function)
- 3. git上開源maven項目部署 多module maven項目（多module maven+redis+tomcat+mysql）後臺部署流程學習記錄
- 4. ecliple-tomcat部署maven項目方式之一
- 5. eclipse新導入的項目經常可以看到「XX cannot be resolved to a type」的報錯信息
- 6. Spark RDD的依賴於DAG的工作原理
- 7. VMware安裝CentOS-8教程詳解
- 8. YDOOK：Java 項目 Spring 項目導入基本四大 jar 包 導入依賴，怎樣在 IDEA 的項目結構中導入 jar 包 導入依賴
- 9. 簡單方法使得putty（windows10上）可以免密登錄樹莓派
- 10. idea怎麼用本地maven
- 1. java8的函數式編程
- 2. java8函數式編程
- 3. Java8函數式編程
- 4. Java8的函數式編程
- 5. java8 函數式編程一
- 6. java8函數式編程（一）
- 7. JAVA8函數式編程
- 8. java8 函數式編程
- 9. 深刻探尋JAVA8 part1：函數式編程與Lambda表達式
- 10. Java8-Lambda表達式（函數式編程）