本節繼續探討Java 8中的函數式數據處理 - Stream API,主要討論各種強大方便的收集器,它們都有什麼用?如何使用?基本實現原理是什麼呢? ...
上節初步介紹了Java 8中的函數式數據處理,對於collect方法,我們只是演示了其最基本的應用,它還有很多強大的功能,比如,可以分組統計彙總,實現類似資料庫查詢語言SQL中的group by功能。
具體都有哪些功能?有什麼用?如何使用?基本原理是什麼?本節進行詳細討論,我們先來進一步理解下collect方法。
理解collect
在上節中,過濾得到90分以上的學生列表,代碼是這樣的:
List<Student> above90List = students.stream() .filter(t->t.getScore()>90) .collect(Collectors.toList());
最後的collect調用看上去很神奇,它到底是怎麼把Stream轉換為List<Student>的呢?先看下collect方法的定義:
<R, A> R collect(Collector<? super T, A, R> collector)
它接受一個收集器collector作為參數,類型是Collector,這是一個介面,它的定義基本是:
public interface Collector<T, A, R> { Supplier<A> supplier(); BiConsumer<A, T> accumulator(); BinaryOperator<A> combiner(); Function<A, R> finisher(); Set<Characteristics> characteristics(); }
在順序流中,collect方法與這些介面方法的交互大概是這樣的:
//首先調用工廠方法supplier創建一個存放處理狀態的容器container,類型為A A container = collector.supplier().get(); //然後對流中的每一個元素t,調用累加器accumulator,參數為累計狀態container和當前元素t for (T t : data) collector.accumulator().accept(container, t); //最後調用finisher對累計狀態container進行可能的調整,類型轉換(A轉換為R),並返回結果 return collector.finisher().apply(container);
combiner只在並行流中有用,用於合併部分結果。characteristics用於標示收集器的特征,Collector介面的調用者可以利用這些特征進行一些優化,Characteristics是一個枚舉,有三個值:CONCURRENT, UNORDERED和IDENTITY_FINISH,它們的含義我們後面通過例子簡要說明,目前可以忽略。
Collectors.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); }
它的實現類是CollectorImpl,這是Collectors內部的一個私有類,實現很簡單,主要就是定義了兩個構造方法,接受函數式參數並賦值給內部變數。對toList來說:
- supplier的實現是ArrayList::new,也就是創建一個ArrayList作為容器
- accumulator的實現是List::add,也就是將碰到的每一個元素加到列表中,
- 第三個參數是combiner,表示合併結果
- 第四個參數CH_ID是一個靜態變數,只有一個特征IDENTITY_FINISH,表示finisher沒有什麼事情可以做,就是把累計狀態container直接返回
也就是說,collect(Collectors.toList())背後的偽代碼如下所示:
List<T> container = new ArrayList<>(); for (T t : data) container.add(t); return container;
與toList類似的容器收集器還有toSet, toCollection, toMap等,我們來看下。
容器收集器
toSet
toSet的使用與toList類似,只是它可以排重,就不舉例了。toList背後的容器是ArrayList,toSet背後的容器是HashSet,其代碼為:
public static <T> Collector<T, ?, Set<T>> toSet() { return new CollectorImpl<>((Supplier<Set<T>>) HashSet::new, Set::add, (left, right) -> { left.addAll(right); return left; }, CH_UNORDERED_ID); }
CH_UNORDERED_ID是一個靜態變數,它的特征有兩個,一個是IDENTITY_FINISH,表示返回結果即為Supplier創建的HashSet,另一個是UNORDERED,表示收集器不會保留順序,這也容易理解,因為背後容器是HashSet。
toCollection
toCollection是一個通用的容器收集器,可以用於任何Collection介面的實現類,它接受一個工廠方法Supplier作為參數,具體代碼為:
public static <T, C extends Collection<T>> Collector<T, ?, C> toCollection(Supplier<C> collectionFactory) { return new CollectorImpl<>(collectionFactory, Collection<T>::add, (r1, r2) -> { r1.addAll(r2); return r1; }, CH_ID); }
比如,如果希望排重但又希望保留出現的順序,可以使用LinkedHashSet,Collector可以這麼創建:
Collectors.toCollection(LinkedHashSet::new)
toMap
toMap將元素流轉換為一個Map,我們知道,Map有鍵和值兩部分,toMap至少需要兩個函數參數,一個將元素轉換為鍵,另一個將元素轉換為值,其基本定義為:
public static <T, K, U> Collector<T, ?, Map<K,U>> toMap( Function<? super T, ? extends K> keyMapper, Function<? super T, ? extends U> valueMapper)
返回結果為Map<K,U>,keyMapper將元素轉換為鍵,valueMapper將元素轉換為值。比如,將學生流轉換為學生名稱和分數的Map,代碼可以為:
Map<String,Double> nameScoreMap = students.stream().collect(
Collectors.toMap(Student::getName, Student::getScore));
這裡,Student::getName是keyMapper,Student::getScore是valueMapper。
實踐中,經常需要將一個對象列表按主鍵轉換為一個Map,以便以後按照主鍵進行快速查找,比如,假定Student的主鍵是id,希望轉換學生流為學生id和學生對象的Map,代碼可以為:
Map<String, Student> byIdMap = students.stream().collect(
Collectors.toMap(Student::getId, t -> t));
t->t是valueMapper,表示值就是元素本身,這個函數用的比較多,介面Function定義了一個靜態函數identity表示它,也就是說,上面的代碼可以替換為:
Map<String, Student> byIdMap = students.stream().collect(
Collectors.toMap(Student::getId, Function.identity()));
上面的toMap假定元素的鍵不能重覆,如果有重覆的,會拋出異常,比如:
Map<String,Integer> strLenMap = Stream.of("abc","hello","abc").collect(
Collectors.toMap(Function.identity(), t->t.length()));
希望得到字元串與其長度的Map,但由於包含重覆字元串"abc",程式會拋出異常。這種情況下,我們希望的是程式忽略後面重覆出現的元素,這時,可以使用另一個toMap函數:
public static <T, K, U> Collector<T, ?, Map<K,U>> toMap( Function<? super T, ? extends K> keyMapper, Function<? super T, ? extends U> valueMapper, BinaryOperator<U> mergeFunction)
相比前面的toMap,它接受一個額外的參數mergeFunction,它用於處理衝突,在收集一個新元素時,如果新元素的鍵已經存在了,系統會將新元素的值與鍵對應的舊值一起傳遞給mergeFunction得到一個值,然後用這個值給鍵賦值。
對於前面字元串長度的例子,新值與舊值其實是一樣的,我們可以用任意一個值,代碼可以為:
Map<String,Integer> strLenMap = Stream.of("abc","hello","abc").collect(
Collectors.toMap(Function.identity(),
t->t.length(), (oldValue,value)->value));
有時,我們可能希望合併新值與舊值,比如一個聯繫人列表,對於相同的聯繫人,我們希望合併電話號碼,mergeFunction可以定義為:
BinaryOperator<String> mergeFunction = (oldPhone,phone)->oldPhone+","+phone;
toMap還有一個更為通用的形式:
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)
相比前面的toMap,多了一個mapSupplier,它是Map的工廠方法,對於前面兩個toMap,其mapSupplier其實是HashMap::new。我們知道,HashMap是沒有任何順序的,如果希望保持元素出現的順序,可以替換為LinkedHashMap,如果希望收集的結果排序,可以使用TreeMap。
toMap主要用於順序流,對於併發流,Collectors有專門的名稱為toConcurrentMap的收集器,它內部使用ConcurrentHashMap,用法類似,具體我們就不討論了。
字元串收集器
除了將元素流收集到容器中,另一個常見的操作是收集為一個字元串。比如,獲取所有的學生名稱,用逗號連接起來,傳統上,代碼看上去像這樣:
StringBuilder sb = new StringBuilder(); for(Student t : students){ if(sb.length()>0){ sb.append(","); } sb.append(t.getName()); } return sb.toString();
針對這種常見的需求,Collectors提供了joining收集器:
public static Collector<CharSequence, ?, String> joining() public static Collector<CharSequence, ?, String> joining(CharSequence delimiter) public static Collector<CharSequence, ?, String> joining( CharSequence delimiter, CharSequence prefix, CharSequence suffix)
第一個就是簡單的把元素連接起來,第二個支持一個分隔符,第三個更為通用,可以給整個結果字元串加個首碼和尾碼。比如:
String result = Stream.of("abc","老馬","hello") .collect(Collectors.joining(",", "[", "]")); System.out.println(result);
輸出為:
[abc,老馬,hello]
joining的內部也利用了StringBuilder,比如,第一個joining函數的代碼為:
public static Collector<CharSequence, ?, String> joining() { return new CollectorImpl<CharSequence, StringBuilder, String>( StringBuilder::new, StringBuilder::append, (r1, r2) -> { r1.append(r2); return r1; }, StringBuilder::toString, CH_NOID); }
supplier是StringBuilder::new,accumulator是StringBuilder::append,finisher是StringBuilder::toString,CH_NOID表示特征集為空。
分組
分組類似於資料庫查詢語言SQL中的group by語句,它將元素流中的每個元素分到一個組,可以針對分組再進行處理和收集,分組的功能比較強大,我們逐步來說明。
為便於舉例,我們先修改下學生類Student,增加一個欄位grade,表示年級,改下構造方法:
public Student(String name, String grade, double score) { this.name = name; this.grade = grade; this.score = score; }
示例學生列表students改為:
static List<Student> students = Arrays.asList(new Student[] { new Student("zhangsan", "1", 91d), new Student("lisi", "2", 89d), new Student("wangwu", "1", 50d), new Student("zhaoliu", "2", 78d), new Student("sunqi", "1", 59d)});
基本用法
最基本的分組收集器為:
public static <T, K> Collector<T, ?, Map<K, List<T>>> groupingBy(Function<? super T, ? extends K> classifier)
參數是一個類型為Function的分組器classifier,它將類型為T的元素轉換為類型為K的一個值,這個值表示分組值,所有分組值一樣的元素會被歸為同一個組,放到一個列表中,所以返回值類型是Map<K, List<T>>。 比如,將學生流按照年級進行分組,代碼為:
Map<String, List<Student>> groups = students.stream()
.collect(Collectors.groupingBy(Student::getGrade));
學生會分為兩組,第一組鍵為"1",分組學生包括"zhangsan", "wangwu"和"sunqi",第二組鍵為"2",分組學生包括"lisi", "zhaoliu"。
這段代碼基本等同於如下代碼:
Map<String, List<Student>> groups = new HashMap<>(); for (Student t : students) { String key = t.getGrade(); List<Student> container = groups.get(key); if (container == null) { container = new ArrayList<>(); groups.put(key, container); } container.add(t); } System.out.println(groups);
顯然,使用groupingBy要簡潔清晰的多,但它到底是怎麼實現的呢?
基本原理
groupingBy的代碼為:
public static <T, K> Collector<T, ?, Map<K, List<T>>> groupingBy(Function<? super T, ? extends K> classifier) { return groupingBy(classifier, toList()); }
它調用了第二個groupingBy方法,傳遞了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); }
這個方法接受一個下游收集器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)
classifier還是分組器,mapFactory是返回Map的工廠方法,預設是HashMap::new,downstream表示下游收集器,下游收集器負責收集同一個分組內元素的結果。
對最通用的groupingBy函數返回的收集器,其收集元素的基本過程和偽代碼為:
//先創建一個存放結果的Map Map map = mapFactory.get(); for (T t : data) { // 對每一個元素,先分組 K key = classifier.apply(t); // 找存放分組結果的容器,如果沒有,讓下游收集器創建,並放到Map中 A container = map.get(key); if (container == null) { container = downstream.supplier().get(); map.put(key, container); } // 將元素交給下游收集器(即分組收集器)收集 downstream.accumulator().accept(container, t); } // 調用分組收集器的finisher方法,轉換結果 for (Map.Entry entry : map.entrySet()) { entry.setValue(downstream.finisher().apply(entry.getValue())); } return map;
在最基本的groupingBy函數中,下游收集器是toList,但下游收集器還可以是其他收集器,甚至是groupingBy,以構成多級分組,下麵我們來看更多的示例。
分組計數、找最大/最小元素
將元素按一定標準分為多組,然後計算每組的個數,按一定標準找最大或最小元素,這是一個常見的需求,Collectors提供了一些對應的收集器,一般用作下游收集器,比如:
//計數 public static <T> Collector<T, ?, Long> counting() //計算最大值 public static <T> Collector<T, ?, Optional<T>> maxBy(Comparator<? super T> comparator) //計算最小值 public static <T> Collector<T, ?, Optional<T>> minBy(Comparator<? super T> comparator)
還有更為通用的名為reducing的歸約收集器,我們就不介紹了,下麵,看一些例子。
為了便於使用Collectors中的方法,我們將其中的方法靜態導入,即加入如下代碼:
import static java.util.stream.Collectors.*;
統計每個年級的學生個數,代碼可以為:
Map<String, Long> gradeCountMap = students.stream().collect(
groupingBy(Student::getGrade, counting()));
統計一個單詞流中每個單詞的個數,按出現順序排序,代碼示例為:
Map<String, Long> wordCountMap = Stream.of("hello","world","abc","hello").collect( groupingBy(Function.identity(), LinkedHashMap::new, counting()));
獲取每個年級分數最高的一個學生,代碼可以為:
Map<String, Optional<Student>> topStudentMap = students.stream().collect(
groupingBy(Student::getGrade,
maxBy(Comparator.comparing(Student::getScore))));
需要說明的是,這個分組收集結果是Optional<Student>,而不是Student,這是因為maxBy處理的流可能是空流,但對我們的例子,這是不可能的,為了直接得到Student,可以使用Collectors的另一個收集器collectingAndThen,在得到Optional<Student>後調用Optional的get方法,如下所示:
Map<String, Student> topStudentMap = students.stream().collect(
groupingBy(Student::getGrade,
collectingAndThen(
maxBy(Comparator.comparing(Student::getScore)),
Optional::get)));
關於collectingAndThen,我們待會再進一步討論。
分組數值統計
除了基本的分組計數,還經常需要進行一些分組數值統計,比如求學生分數的和、平均分、最高分/最低分等,針對int,long和double類型,Collectors提供了專門的收集器,比如:
//求平均值,int和long也有類似方法 public static <T> Collector<T, ?, Double> averagingDouble(ToDoubleFunction<? super T> mapper) //求和,long和double也有類似方法 public static <T> Collector<T, ?, Integer> summingInt(ToIntFunction<? super T> mapper) //求多種彙總信息,int和double也有類似方法 //LongSummaryStatistics包括個數、最大值、最小值、和、平均值等多種信息 public static <T> Collector<T, ?, LongSummaryStatistics> summarizingLong(ToLongFunction<? super T> mapper)
比如,按年級統計學生分數信息,代碼可以為:
Map<String, DoubleSummaryStatistics> gradeScoreStat =
students.stream().collect(
groupingBy(Student::getGrade,
summarizingDouble(Student::getScore)));
分組內的map
對於每個分組內的元素,我們感興趣的可能不是元素本身,而是它的某部分信息,在上節介紹的Stream API中,Stream有map方法,可以將元素進行轉換,Collectors也為分組元素提供了函數mapping,如下所示:
public static <T, U, A, R> Collector<T, ?, R> mapping(Function<? super T, ? extends U> mapper, Collector<? super U, A, R> downstream)
交給下游收集器downstream的不再是元素本身,而是應用轉換函數mapper之後的結果。比如,對學生按年級分組,得到學生名稱列表,代碼可以為:
Map<String, List<String>> gradeNameMap =
students.stream().collect(
groupingBy(Student::getGrade,
mapping(Student::getName, toList())));
System.out.println(gradeNameMap);
輸出為:
{1=[zhangsan, wangwu, sunqi], 2=[lisi, zhaoliu]}
分組結果處理(filter/sort/skip/limit)
對分組後的元素,我們可以計數,找最大/最小元素,計算一些數值特征,還可以轉換後(map)再收集,那可不可以像上節介紹的Stream API一樣,進行排序(sort)、過濾(filter)、限制返回元素(skip/limit)呢?Collector沒有專門的收集器,但有一個通用的方法:
public static<T,A,R,RR> Collector<T,A,RR> collectingAndThen( Collector<T,A,R> downstream, Function<R,RR> finisher)
這個方法接受一個下游收集器downstream和一個finisher,返回一個收集器,它的主要代碼為:
return new CollectorImpl<>(downstream.supplier(), downstream.accumulator(), downstream.combiner(), downstream.finisher().andThen(finisher), characteristics);
也就是說,它在下游收集器的結果上又調用了finisher。利用這個finisher,我們可以實現多種功能,下麵看一些例子。
收集完再排序,可以定義如下方法:
public static <T> Collector<T, ?, List<T>> collectingAndSort( Collector<T, ?, List<T>> downstream, Comparator<? super T> comparator) { return Collectors.collectingAndThen(downstream, (r) -> { r.sort(comparator); return r; }); }
比如,將學生按年級分組,分組內學生按照分數由高到低進行排序,利用這個方法,代碼可以為:
Map<String, List<Student>> gradeStudentMap =
students.stream().collect(
groupingBy(Student::getGrade,
collectingAndSort(toList(),
Comparator.comparing(Student::getScore).reversed())));
針對這個需求,也可以先對流進行排序,然後再分組。
收集完再過濾,可以定義如下方法:
public static <T> Collector<T, ?, List<T>> collectingAndFilter( Collector<T, ?, List<T>> downstream, Predicate<T> predicate) { return Collectors.collectingAndThen(downstream, (r) -> { return r.stream().filter(predicate).collect(Collectors.toList()); }); }
比如,將學生按年級分組,分組後,每個分組只保留不及格的學生(低於60分),利用這個方法,代碼可以為:
Map<String, List<Student>> gradeStudentMap =
students.stream().collect(
groupingBy(Student::getGrade,
collectingAndFilter(toList(), t->t.getScore()<60)));
針對這個需求,也可以先對流進行過濾,然後再分組。
收集完,只返回特定區間的結果,可以定義如下方法:
public static <T> Collector<T, ?, List<T>> collectingAndSkipLimit( Collector<T, ?, List<T>> downstream, long skip, long limit) { return Collectors.collectingAndThen(downstream, (r) -> { return r.stream().skip(skip).limit(limit).collect(Collectors.toList()); }); }
比如,將學生按年級分組,分組後,每個分組只保留前兩名的學生,代碼可以為:
Map<String, List<Student>> gradeStudentMap =
students.stream()
.sorted(Comparator.comparing(Student::getScore).reversed())
.collect(groupingBy(Student::getGrade,
collectingAndSkipLimit(toList(), 0, 2)));
這次,我們先對學生流進行了排序,然後再進行了分組。
分區
分組的一個特殊情況是分區,就是將流按true/false分為兩個組,Collectors有專門的分區函數:
public static <T> Collector<T, ?, Map<Boolean, List<T>>> partitioningBy(Predicate<? super T> predicate) public static <T, D, A> Collector<T, ?, Map<Boolean, D>> partitioningBy(Predicate<? super T> predicate, Collector<? super T, A, D> downstream)
第一個的下游收集器為toList(),第二個可以指定一個下游收集器。
比如,將學生按照是否及格(大於等於60分)分為兩組,代碼可以為:
Map<Boolean, List<Student>> byPass = students.stream().collect(
partitioningBy(t->t.getScore()>=60));
按是否及格分組後,計算每個分組的平均分,代碼可以為:
Map<Boolean, Double> avgScoreMap = students.stream().collect( partitioningBy(t->t.getScore()>=60, averagingDouble(Student::getScore)));
多級分組
groupingBy和partitioningBy都可以接受一個下游收集器,而下游收集器又可以是分組或分區。
比如,按年級對學生分組,分組後,再按照是否及格對學生進行分區,代碼可以為:
Map<String, Map<Boolean, List<Student>>> multiGroup =
students.stream().collect(
groupingBy(Student::getGrade,
partitioningBy(t->t.getScore()>=60)));
小結
本節主要討論了各種收集器,包括容器收集器、字元串收集器、分組和分區收集器等。
對於分組和分區,它們接受一個下游收集器,對同一個分組或分區內的元素進行進一步收集,下游收集器還可以是分組或分區,以構建多級分組,有一些收集器主要用於分組,比如counting, maxBy, minBy, summarizingDouble等。
mapping和collectingAndThen也都接受一個下游收集器,mapping在把元素交給下游收集器之前先進行轉換,而collectingAndThen對下游收集器的結果進行轉換,組合利用它們,可以構造更為靈活強大的收集器。
至此,關於Java 8中的函數式數據處理Stream API,我們就介紹完了,Stream API提供了集合數據處理的常用函數,利用它們,可以簡潔地實現大部分常見需求,大大減少代碼,提高可讀性。
對於併發編程,Java 8也提供了一個新的類CompletableFuture,類似於Stream API對集合數據的流水線式操作,使用CompletableFuture,可以實現對多個非同步任務進行流水線式操作,它具體是什麼呢?
(與其他章節一樣,本節所有代碼位於 https://github.com/swiftma/program-logic,位於包shuo.laoma.java8.c93下)
----------------
未完待續,查看最新文章,敬請關註微信公眾號“老馬說編程”(掃描下方二維碼),從入門到高級,深入淺出,老馬和你一起探索Java編程及電腦技術的本質。用心原創,保留所有版權。