基礎篇：JAVA.Stream函數，優雅的數據流操做
前言
平時操做集合數據，咱們通常都是for或者iterator去遍歷，不是很好看。java提供了Stream的概念，它可讓咱們把集合數據當作一個個元素在處理，而且提供多線程模式java
- 流的建立
- 流的各類數據操做
- 流的終止操做
- 流的聚合處理
- 併發流和CompletableFuture的配合使用
關注公衆號，一塊兒交流，微信搜一搜: 潛行前行
1 stream的構造方式
stream內置的構造方法
public static<T> Stream<T> iterate(final T seed, final UnaryOperator<T> f) public static <T> Stream<T> concat(Stream<? extends T> a, Stream<? extends T> b) public static<T> Builder<T> builder() public static<T> Stream<T> of(T t) public static<T> Stream<T> empty() public static<T> Stream<T> generate(Supplier<T> s)
Collection聲明的stream函數
default Stream<E> stream()
- Collection聲明瞭stream轉化函數，也就是說，任意Collection子類都存在官方替咱們實現的由Collection轉爲Stream的方法
- 示例，List轉Stream
public static void main(String[] args){
List<String> demo = Arrays.asList("a","b","c");
long count = demo.stream().peek(System.out::println).count();
System.out.println(count);
}
-------result--------
a
b
c
3
2 接口stream對元素的操做方法定義
過濾 filter
Stream<T> filter(Predicate<? super T> predicate)
- Predicate是函數式接口，能夠直接用lambda代替；若是有複雜的過濾邏輯，則用or、and、negate方法組合
- 示例
List<String> demo = Arrays.asList("a", "b", "c");
Predicate<String> f1 = item -> item.equals("a");
Predicate<String> f2 = item -> item.equals("b");
demo.stream().filter(f1.or(f2)).forEach(System.out::println);
-------result--------
a
b
映射轉化 map
<R> Stream<R> map(Function<? super T, ? extends R> mapper) IntStream mapToInt(ToIntFunction<? super T> mapper); LongStream mapToLong(ToLongFunction<? super T> mapper); DoubleStream mapToDouble(ToDoubleFunction<? super T> mapper);
- 示例
static class User{
public User(Integer id){this.id = id; }
Integer id; public Integer getId() { return id; }
}
public static void main(String[] args) {
List<User> demo = Arrays.asList(new User(1), new User(2), new User(3));
// User 轉爲 Integer(id)
demo.stream().map(User::getId).forEach(System.out::println);
}
-------result--------
1
2
3
數據處理 peek
Stream<T> peek(Consumer<? super T> action);
- 與map的區別是其無返回值
- 示例
static class User{
public User(Integer id){this.id = id; }
Integer id;
public Integer getId() { return id; }
public void setId(Integer id) { this.id = id; }
}
public static void main(String[] args) {
List<User> demo = Arrays.asList(new User(1), new User(2), new User(3));
// id平方，User 轉爲 Integer(id)
demo.stream().peek(user -> user.setId(user.id * user.id)).map(User::getId).forEach(System.out::println);
}
-------result--------
1
4
9
映射攆平 flatMap
<R> Stream<R> flatMap(Function<? super T, ? extends Stream<? extends R>> mapper); IntStream flatMapToInt(Function<? super T, ? extends IntStream> mapper); LongStream flatMapToLong(Function<? super T, ? extends LongStream> mapper); DoubleStream flatMapToDouble(Function<? super T, ? extends DoubleStream> mapper);
- flatMap：將元素爲Stream<T>類型的流攆平成一個元素類型爲T的Stream流
- 示例
public static void main(String[] args) {
List<Stream<Integer>> demo = Arrays.asList(Stream.of(5), Stream.of(2), Stream.of(1));
demo.stream().flatMap(Function.identity()).forEach(System.out::println);
}
-------result--------
5
2
1
去重 distinct
Stream<T> distinct();
- 示例
List<Integer> demo = Arrays.asList(1, 1, 2); demo.stream().distinct().forEach(System.out::println); -------result-------- 1 2
排序 sorted
Stream<T> sorted(); Stream<T> sorted(Comparator<? super T> comparator);
- 示例
List<Integer> demo = Arrays.asList(5, 1, 2); //默認升序 demo.stream().sorted().forEach(System.out::println); //降序 Comparator<Integer> comparator = Comparator.<Integer, Integer>comparing(item -> item).reversed(); demo.stream().sorted(comparator).forEach(System.out::println); -------默認升序 result-------- 1 2 5 -------降序 result-------- 5 2 1
個數限制limit和跳過skip
//截取前maxSize個元素 Stream<T> limit(long maxSize); //跳過前n個流 Stream<T> skip(long n);
- 示例
List<Integer> demo = Arrays.asList(1, 2, 3, 4, 5, 6); //跳過前兩個，而後限制截取兩個 demo.stream().skip(2).limit(2).forEach(System.out::println); -------result-------- 3 4
JDK9提供的新操做
- 和filter的區別，takeWhile是取知足條件的元素，直到不知足爲止；dropWhile是丟棄知足條件的元素，直到不知足爲止
default Stream<T> takeWhile(Predicate<? super T> predicate); default Stream<T> dropWhile(Predicate<? super T> predicate);
3 stream的終止操做action
遍歷消費
//遍歷消費 void forEach(Consumer<? super T> action); //順序遍歷消費,和forEach的區別是forEachOrdered在多線程parallelStream執行，其順序也不會亂 void forEachOrdered(Consumer<? super T> action);
- 示例
List<Integer> demo = Arrays.asList(1, 2, 3); demo.parallelStream().forEach(System.out::println); demo.parallelStream().forEachOrdered(System.out::println); -------forEach result-------- 2 3 1 -------forEachOrdered result-------- 1 2 3
獲取數組結果
//流轉成Object數組 Object[] toArray(); //流轉成Ａ[]數組，指定類型A <A> A[] toArray(IntFunction<A[]> generator)
- 示例
List<String> demo = Arrays.asList("1", "2", "3");
//<A> A[] toArray(IntFunction<A[]> generator)
String[] data = demo.stream().toArray(String[]::new);
最大最小值
//獲取最小值 Optional<T> min(Comparator<? super T> comparator) //獲取最大值 Optional<T> max(Comparator<? super T> comparator)
- 示例
List<Integer> demo = Arrays.asList(1, 2, 3); Optional<Integer> min = demo.stream().min(Comparator.comparing(item->item)); Optional<Integer> max = demo.stream().max(Comparator.comparing(item->item)); System.out.println(min.get()+"-"+max.get()); -------result-------- 1-3
查找匹配
//任意一個匹配 boolean anyMatch(Predicate<? super T> predicate) //所有匹配 boolean allMatch(Predicate<? super T> predicate) //不匹配 boolean noneMatch(Predicate<? super T> predicate) //查找第一個 Optional<T> findFirst(); //任意一個 Optional<T> findAny();
歸約合併
//兩兩合併 Optional<T> reduce(BinaryOperator<T> accumulator) //兩兩合併，帶初始值的 T reduce(T identity, BinaryOperator<T> accumulator) //先轉化元素類型再兩兩合併，帶初始值的 <U> U reduce(U identity, BiFunction<U, ? super T, U> accumulator, BinaryOperator<U> combiner)
- 示例
List<Integer> demo = Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8);
//數字轉化爲字符串，而後使用「-」拼接起來
String data = demo.stream().reduce("0", (u, t) -> u + "-" + t, (s1, s2) -> s1 + "-" + s2);
System.out.println(data);
-------result--------
0-1-2-3-4-5-6-7-8
計算元素個數
long count()
- 示例
List<Integer> demo = Arrays.asList(1, 2, 3, 4, 5, 6); System.out.println(demo.stream().count()); -------result-------- 6
對流的聚合處理
/** * supplier:返回結果類型的生產者 * accumulator:元素消費者（處理並加入R） * combiner: 返回結果 R 怎麼組合（多線程執行時，會產生多個返回值R，須要合併） */ <R> R collect(Supplier<R> supplier, BiConsumer<R, ? super T> accumulator, BiConsumer<R, R> combiner); /** * collector通常是由 supplier、accumulator、combiner、finisher、characteristics組合成的聚合類 * Collectors 可提供一些內置的聚合類或者方法 */ <R, A> R collect(Collector<? super T, A, R> collector);
- 示例，看下面
4 Collector(聚合類)的工具類集Collectors
接口Collector和實現類CollectorImpl
//返回值類型的生產者 Supplier<A> supplier(); //流元素消費者 BiConsumer<A, T> accumulator(); //返回值合併器（多個線程操做時，會產生多個返回值，須要合併） BinaryOperator<A> combiner(); //返回值轉化器（最後一步處理，實際返回結果，通常原樣返回） Function<A, R> finisher(); //流的特性 Set<Characteristics> characteristics(); public static<T, A, R> Collector<T, A, R> of(Supplier<A> supplier, BiConsumer<A, T> accumulator, BinaryOperator<A> combiner, Function<A, R> finisher, Characteristics... characteristics)
流聚合轉換成List, Set
//流轉化成List public static <T> Collector<T, ?, List<T>> toList() //流轉化成Set public static <T> Collector<T, ?, Set<T>> toSet()
- 示例
List<Integer> demo = Arrays.asList(1, 2, 3); List<Integer> col = demo.stream().collect(Collectors.toList()); Set<Integer> set = demo.stream().collect(Collectors.toSet());
流聚合轉化成Map
//流轉化成Map
public static <T, K, U> Collector<T, ?, Map<K,U>> toMap(
Function<? super T, ? extends K> keyMapper,
Function<? super T, ? extends U> valueMapper)
/**
* mergeFunction:相同的key,值怎麼合併
*/
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)
/**
* mergeFunction:相同的key,值怎麼合併
* mapSupplier：返回值Map的生產者
*/
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)
- 若是存在相同key的元素，會報錯;或者使用groupBy
- 示例
List<User> demo = Arrays.asList(new User(1), new User(2), new User(3));
Map<Integer,User> map = demo.stream().collect(Collectors.toMap(User::getId,item->item));
System.out.println(map);
-------result-------
{1=TestS$User@7b23ec81, 2=TestS$User@6acbcfc0, 3=TestS$User@5f184fc6}
字符串流聚合拼接
//多個字符串拼接成一個字符串 public static Collector<CharSequence, ?, String> joining(); //多個字符串拼接成一個字符串（指定分隔符） public static Collector<CharSequence, ?, String> joining(CharSequence delimiter)
- 示例
List<String> demo = Arrays.asList("c", "s", "c","w","潛行前行");
String name = demo.stream().collect(Collectors.joining("-"));
System.out.println(name);
-------result-------
c-s-c-w-潛行前行
映射處理再聚合流
- 至關於先map再collect
/** * mapper:映射處理器 * downstream:映射處理後須要再次聚合處理 */ public static <T, U, A, R> Collector<T, ?, R> mapping(Function<? super T, ? extends U> mapper, Collector<? super U, A, R> downstream);
- 示例
List<String> demo = Arrays.asList("1", "2", "3");
List<Integer> data = demo.stream().collect(Collectors.mapping(Integer::valueOf, Collectors.toList()));
System.out.println(data);
-------result-------
[1, 2, 3]
聚合後再轉換結果
/** * downstream:聚合處理 * finisher:結果轉換處理 */ public static<T,A,R,RR> Collector<T,A,RR> collectingAndThen(Collector<T,A,R> downstream, Function<R, RR> finisher);
- 示例
List<Integer> demo = Arrays.asList(1, 2, 3, 4, 5, 6); //聚合成List,最後提取數組的size做爲返回值 Integer size = demo.stream().collect(Collectors.collectingAndThen(Collectors.toList(), List::size)); System.out.println(size); ---------result---------- 6
流分組（Map是HashMap）
/** * classifier 指定T類型某一屬性做爲Key值分組 * 分組後，使用List做爲每一個流的容器 */ public static <T, K> Collector<T, ?, Map<K, List<T>>> groupingBy( Function<? super T, ? extends K> classifier); /** * classifier: 流分組器 * downstream: 每組流的聚合處理器 */ public static <T, K, A, D> Collector<T, ?, Map<K, D>> groupingBy( Function<? super T, ? extends K> classifier， Collector<? super T, A, D> downstream) /** * classifier: 流分組器 * mapFactory: 返回值map的工廠（Map的子類） * 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)
- 示例
public static void main(String[] args) throws Exception {
List<Integer> demo = Stream.iterate(0, item -> item + 1)
.limit(15)
.collect(Collectors.toList());
// 分紅三組，而且每組元素轉化爲String類型
Map<Integer, List<String>> map = demo.stream()
.collect(Collectors.groupingBy(item -> item % 3,
HashMap::new,
Collectors.mapping(String::valueOf, Collectors.toList())));
System.out.println(map);
}
---------result----------
{0=[0, 3, 6, 9, 12], 1=[1, 4, 7, 10, 13], 2=[2, 5, 8, 11, 14]}
流分組(分組使用的Map是ConcurrentHashMap)
/** * classifier: 分組器 ； 分組後，使用List做爲每一個流的容器 */ public static <T, K> Collector<T, ?, ConcurrentMap<K, List<T>>> groupingByConcurrent( Function<? super T, ? extends K> classifier); /** * classifier: 分組器 * downstream: 流的聚合處理器 */ public static <T, K, A, D> Collector<T, ?, ConcurrentMap<K, D>> groupingByConcurrent( Function<? super T, ? extends K> classifier, Collector<? super T, A, D> downstream) /** * classifier: 分組器 * mapFactory: 返回值類型map的生產工廠（ConcurrentMap的子類） * downstream: 流的聚合處理器 */ public static <T, K, A, D, M extends ConcurrentMap<K, D>> Collector<T, ?, M> groupingByConcurrent( Function<? super T, ? extends K> classifier, Supplier<M> mapFactory, Collector<? super T, A, D> downstream);
- 用法和groupingBy同樣
拆分流，一變二（至關於特殊的groupingBy）
public static <T> Collector<T, ?, Map<Boolean, List<T>>> partitioningBy( Predicate<? super T> predicate) /** * predicate: 二分器 * downstream: 流的聚合處理器 */ public static <T, D, A> Collector<T, ?, Map<Boolean, D>> partitioningBy( Predicate<? super T> predicate, Collector<? super T, A, D> downstream)
- 示例
List<Integer> demo = Arrays.asList(1, 2,3,4, 5,6);
// 奇數偶數分組
Map<Boolean, List<Integer>> map = demo.stream()
.collect(Collectors.partitioningBy(item -> item % 2 == 0));
System.out.println(map);
---------result----------
{false=[1, 3, 5], true=[2, 4, 6]}
聚合求平均值
// 返回Double類型 public static <T> Collector<T, ?, Double> averagingDouble(ToDoubleFunction<? super T> mapper) // 返回Long 類型 public static <T> Collector<T, ?, Double> averagingLong(ToLongFunction<? super T> mapper) //返回Int 類型 public static <T> Collector<T, ?, Double> averagingInt(ToIntFunction<? super T> mapper)
- 示例
List<Integer> demo = Arrays.asList(1, 2, 5); Double data = demo.stream().collect(Collectors.averagingInt(Integer::intValue)); System.out.println(data); ---------result---------- 2.6666666666666665
流聚合查找最大最小值
//最小值 public static <T> Collector<T, ?, Optional<T>> minBy(Comparator<? super T> comparator) //最大值 public static <T> Collector<T, ?, Optional<T>> maxBy(Comparator<? super T> comparator)
- 示例
List<Integer> demo = Arrays.asList(1, 2, 5); Optional<Integer> min = demo.stream().collect(Collectors.minBy(Comparator.comparing(item -> item))); Optional<Integer> max = demo.stream().collect(Collectors.maxBy(Comparator.comparing(item -> item))); System.out.println(min.get()+"-"+max.get()); ---------result---------- 1-5
聚合計算統計結果
- 能夠得到元素總個數，元素累計總和，最小值，最大值，平均值
//返回Int 類型 public static <T> Collector<T, ?, IntSummaryStatistics> summarizingInt( ToIntFunction<? super T> mapper) //返回Double 類型 public static <T> Collector<T, ?, DoubleSummaryStatistics> summarizingDouble( ToDoubleFunction<? super T> mapper) //返回Long 類型 public static <T> Collector<T, ?, LongSummaryStatistics> summarizingLong( ToLongFunction<? super T> mapper)
- 示例
List<Integer> demo = Arrays.asList(1, 2, 5);
IntSummaryStatistics data = demo.stream().collect(Collectors.summarizingInt(Integer::intValue));
System.out.println(data);
---------result----------
IntSummaryStatistics{count=3, sum=8, min=1, average=2.666667, max=5}
JDK12提供的新聚合方法
//流分別通過downstream一、downstream2聚合處理，再合併兩聚合結果 public static <T, R1, R2, R> Collector<T, ?, R> teeing( Collector<? super T, ?, R1> downstream1, Collector<? super T, ?, R2> downstream2, BiFunction<? super R1, ? super R2, R> merger)
5 併發paralleStream的使用
- 配合CompletableFuture和線程池的使用
- 示例
public static void main(String[] args) throws Exception{
List<Integer> demo = Stream.iterate(0, item -> item + 1)
.limit(5)
.collect(Collectors.toList());
//示例1
Stopwatch stopwatch = Stopwatch.createStarted(Ticker.systemTicker());
demo.stream().forEach(item -> {
try {
Thread.sleep(500);
System.out.println("示例1-"+Thread.currentThread().getName());
} catch (Exception e) { }
});
System.out.println("示例1-"+stopwatch.stop().elapsed(TimeUnit.MILLISECONDS));
//示例2, 注意須要ForkJoinPool，parallelStream纔會使用executor指定的線程，不然仍是用默認的 ForkJoinPool.commonPool()
ExecutorService executor = new ForkJoinPool(10);
stopwatch.reset(); stopwatch.start();
CompletableFuture.runAsync(() -> demo.parallelStream().forEach(item -> {
try {
Thread.sleep(1000);
System.out.println("示例2-" + Thread.currentThread().getName());
} catch (Exception e) { }
}), executor).join();
System.out.println("示例2-"+stopwatch.stop().elapsed(TimeUnit.MILLISECONDS));
//示例３
stopwatch.reset(); stopwatch.start();
demo.parallelStream().forEach(item -> {
try {
Thread.sleep(1000);
System.out.println("示例3-"+Thread.currentThread().getName());
} catch (Exception e) { }
});
System.out.println("示例3-"+stopwatch.stop().elapsed(TimeUnit.MILLISECONDS));
executor.shutdown();
}
- -------------------result--------------------------
示例1-main 示例1-main 示例1-main 示例1-main 示例1-main 示例1-2501 示例2-ForkJoinPool-1-worker-19 示例2-ForkJoinPool-1-worker-9 示例2-ForkJoinPool-1-worker-5 示例2-ForkJoinPool-1-worker-27 示例2-ForkJoinPool-1-worker-23 示例2-1004 示例3-main 示例3-ForkJoinPool.commonPool-worker-5 示例3-ForkJoinPool.commonPool-worker-7 示例3-ForkJoinPool.commonPool-worker-9 示例3-ForkJoinPool.commonPool-worker-3 示例3-1001
- parallelStream的方法確實會使用多線程去運行，而且能夠指定線程池，不過自定義線程必須是ForkJoinPool類型，不然會默認使ForkJoinPool.commonPool()的線程
相關文章
- 1. MySql基礎篇----DQL數據查詢的基本操做（上）2.6
- 2. MySql基礎篇----DQL數據查詢的基本操做（下）2.8
- 3. python基礎之文件操做，函數
- 4. python函數基礎 與文件操做
- 5. Python基礎——文件操做、函數（0412）
- 6. mysql基礎一：數據庫的操做
- 7. MySQL數據庫表的基礎操做
- 8. MySQL數據庫基礎——基本操做
- 9. python框架之Flask基礎篇（二）-------- 數據庫的操做
- 10. [Android開源]EasySharedPreferences：優雅的進行SharedPreferences數據存儲操做
- 更多相關文章...
- • NoSQL數據庫的優勢有哪些？ - NoSQL教程
- • MySQL的優勢（優點） - MySQL教程
- • Flink 數據傳輸及反壓詳解
- • TiDB 在摩拜單車在線數據業務的應用和實踐
相關標籤/搜索
每日一句
-
每一个你不满意的现在，都有一个你没有努力的曾经。
歡迎關注本站公眾號,獲取更多信息
相關文章