java8之stream

時間 2019-11-12

標籤 java8 java stream 欄目 Java 简体版

原文原文鏈接

lambda表達式是stream的基礎，初學者建議先學習lambda表達式， http://www.cnblogs.com/andywithu/p/7357069.html

1.初識stream

先來一個總綱：

東西就是這麼多啦，stream是java8中加入的一個很是實用的功能，最初看時覺得是io中的流（其實一點關係都沒有），讓咱們先來看一個小例子感覺一下：

@Before
public void init() {
    random = new Random();
    stuList = new ArrayList<Student>() {
        {
            for (int i = 0; i < 100; i++) {
                add(new Student("student" + i, random.nextInt(50) + 50));
            }
        }
    };
}
 
public class Student {
    private String name;
    private Integer score;
    //-----getters and setters-----
}
 
//1列出班上超過85分的學生姓名，並按照分數降序輸出用戶名字
@Test
public void test1() {
    List<String> studentList = stuList.stream()
            .filter(x->x.getScore()>85)
            .sorted(Comparator.comparing(Student::getScore).reversed())
            .map(Student::getName)
            .collect(Collectors.toList());
    System.out.println(studentList);
}

列出班上分數超過85分的學生姓名，並按照分數降序輸出用戶名字，在java8以前咱們須要三個步驟：

1）新建一個List<Student> newList，在for循環中遍歷stuList，將分數超過85分的學生裝入新的集合中

2）對於新的集合newList進行排序操做

3）遍歷打印newList

這三個步驟在java8中只須要兩條語句，若是牢牢須要打印，不須要保存新生產list的話實際上只須要一條，是否是很是方便。

2.stream的特性

咱們首先列出stream的以下三點特性，在以後咱們會對照着詳細說明

1.stream不存儲數據

2.stream不改變源數據

3.stream的延遲執行特性

一般咱們在數組或集合的基礎上建立stream，stream不會專門存儲數據，對stream的操做也不會影響到建立它的數組和集合,對於stream的聚合、消費或收集操做只能進行一次，再次操做會報錯，以下代碼：

@Test
public void test1(){
    Stream<String> stream = Stream.generate(()->"user").limit(20);
    stream.forEach(System.out::println);
    stream.forEach(System.out::println);
}

程序在正常完成一次打印工做後報錯。

stream的操做是延遲執行的，在列出班上超過85分的學生姓名例子中，在collect方法執行以前，filter、sorted、map方法還未執行，只有當collect方法執行時纔會觸發以前轉換操做

看以下代碼：

public boolean filter(Student s) {
    System.out.println("begin compare");
    return s.getScore() > 85;
}
 
@Test
public void test() {
    Stream<Student> stream = Stream.of(stuArr).filter(this::filter);
    System.out.println("split-------------------------------------");
    List<Student> studentList = stream.collect(toList());
}

咱們將filter中的邏輯抽象成方法，在方法中加入打印邏輯，若是stream的轉換操做是延遲執行的，那麼split會先打印，不然後打印，代碼運行結果爲

可見stream的操做是延遲執行的。

TIP：

當咱們操做一個流的時候，並不會修改流底層的集合（即便集合是線程安全的），若是想要修改原有的集合，就沒法定義流操做的輸出。

因爲stream的延遲執行特性，在聚合操做執行前修改數據源是容許的。

List<String> wordList;
 
@Before
public void init() {
    wordList = new ArrayList<String>() {
        {
            add("a");
            add("b");
            add("c");
            add("d");
            add("e");
            add("f");
            add("g");
        }
    };
}
/**
 * 延遲執行特性，在聚合操做以前均可以添加相應元素
 */
@Test
public void test() {
    Stream<String> words = wordList.stream();
    wordList.add("END");
    long n = words.distinct().count();
    System.out.println(n);
}

最後打印的結果是8

以下代碼是錯誤的

/**
 * 延遲執行特性，會產生干擾
 * nullPointException
 */
@Test
public void test2(){
    Stream<String> words1 = wordList.stream();
    words1.forEach(s -> {
        System.out.println("s->"+s);
        if (s.length() < 4) {
            System.out.println("select->"+s);
            wordList.remove(s);
            System.out.println(wordList);
        }
    });
}

結果報空指針異常

3.建立stream

1）經過數組建立

/**
 * 經過數組建立流
 */
@Test
public void testArrayStream(){
    //1.經過Arrays.stream
    //1.1基本類型
    int[] arr = new int[]{1,2,34,5};
    IntStream intStream = Arrays.stream(arr);
    //1.2引用類型
    Student[] studentArr = new Student[]{new Student("s1",29),new Student("s2",27)};
    Stream<Student> studentStream = Arrays.stream(studentArr);
    //2.經過Stream.of
    Stream<Integer> stream1 = Stream.of(1,2,34,5,65);
    //注意生成的是int[]的流
    Stream<int[]> stream2 = Stream.of(arr,arr);
    stream2.forEach(System.out::println);
}

2）經過集合建立流

/**
 * 經過集合建立流
 */
@Test
public void testCollectionStream(){
    List<String> strs = Arrays.asList("11212","dfd","2323","dfhgf");
    //建立普通流
    Stream<String> stream  = strs.stream();
    //建立並行流
    Stream<String> stream1 = strs.parallelStream();
}

3）建立空的流

@Test
public void testEmptyStream(){
    //建立一個空的stream
    Stream<Integer> stream  = Stream.empty();
}

4）建立無限流

@Test
public void testUnlimitStream(){
    //建立無限流，經過limit提取指定大小
    Stream.generate(()->"number"+new Random().nextInt()).limit(100).forEach(System.out::println);
    Stream.generate(()->new Student("name",10)).limit(20).forEach(System.out::println);
}

5）建立規律的無限流

/**
 * 產生規律的數據
 */
@Test
public void testUnlimitStream1(){
    Stream.iterate(0,x->x+1).limit(10).forEach(System.out::println);
    Stream.iterate(0,x->x).limit(10).forEach(System.out::println);
    //Stream.iterate(0,x->x).limit(10).forEach(System.out::println);與以下代碼意思是同樣的
    Stream.iterate(0, UnaryOperator.identity()).limit(10).forEach(System.out::println);
}

4.對stream的操做

1）最常使用

map：轉換流，將一種類型的流轉換爲另一種流

/**
 * map把一種類型的流轉換爲另一種類型的流
 * 將String數組中字母轉換爲大寫
 */
@Test
public void testMap() {
    String[] arr = new String[]{"yes", "YES", "no", "NO"};
    Arrays.stream(arr).map(x -> x.toLowerCase()).forEach(System.out::println);
}

filter：過濾流，過濾流中的元素

@Test
public void testFilter(){
    Integer[] arr = new Integer[]{1,2,3,4,5,6,7,8,9,10};
    Arrays.stream(arr).filter(x->x>3&&x<8).forEach(System.out::println);
}

flapMap：拆解流，將流中每個元素拆解成一個流

/**
 * flapMap：拆解流
 */
@Test
public void testFlapMap1() {
    String[] arr1 = {"a", "b", "c", "d"};
    String[] arr2 = {"e", "f", "c", "d"};
    String[] arr3 = {"h", "j", "c", "d"};
   // Stream.of(arr1, arr2, arr3).flatMap(x -> Arrays.stream(x)).forEach(System.out::println);
    Stream.of(arr1, arr2, arr3).flatMap(Arrays::stream).forEach(System.out::println);
}

sorted：對流進行排序

String[] arr1 = {"abc","a","bc","abcd"};
/**
 * Comparator.comparing是一個鍵提取的功能
 * 如下兩個語句表示相贊成義
 */
@Test
public void testSorted1_(){
    /**
     * 按照字符長度排序
     */
    Arrays.stream(arr1).sorted((x,y)->{
        if (x.length()>y.length())
            return 1;
        else if (x.length()<y.length())
            return -1;
        else
            return 0;
    }).forEach(System.out::println);
    Arrays.stream(arr1).sorted(Comparator.comparing(String::length)).forEach(System.out::println);
}
 
/**
 * 倒序
 * reversed(),java8泛型推導的問題，因此若是comparing裏面是非方法引用的lambda表達式就沒辦法直接使用reversed()
 * Comparator.reverseOrder():也是用於翻轉順序，用於比較對象（Stream裏面的類型必須是可比較的）
 * Comparator. naturalOrder()：返回一個天然排序比較器，用於比較對象（Stream裏面的類型必須是可比較的）
 */
@Test
public void testSorted2_(){
    Arrays.stream(arr1).sorted(Comparator.comparing(String::length).reversed()).forEach(System.out::println);
    Arrays.stream(arr1).sorted(Comparator.reverseOrder()).forEach(System.out::println);
    Arrays.stream(arr1).sorted(Comparator.naturalOrder()).forEach(System.out::println);
}
 
/**
 * thenComparing
 * 先按照首字母排序
 * 以後按照String的長度排序
 */
@Test
public void testSorted3_(){
    Arrays.stream(arr1).sorted(Comparator.comparing(this::com1).thenComparing(String::length)).forEach(System.out::println);
}
public char com1(String x){
    return x.charAt(0);
}

2）提取流和組合流

@Before
    public void init(){
        arr1 = new String[]{"a","b","c","d"};
        arr2 = new String[]{"d","e","f","g"};
        arr3 = new String[]{"i","j","k","l"};
    }
    /**
     * limit，限制從流中得到前n個數據
     */
    @Test
    public void testLimit(){
        Stream.iterate(1,x->x+2).limit(10).forEach(System.out::println);
    }
 
    /**
     * skip，跳過前n個數據
     */
    @Test
    public void testSkip(){
//        Stream.of(arr1).skip(2).limit(2).forEach(System.out::println);
        Stream.iterate(1,x->x+2).skip(1).limit(5).forEach(System.out::println);
    }
 
    /**
     * 能夠把兩個stream合併成一個stream（合併的stream類型必須相同）
     * 只能兩兩合併
     */
    @Test
    public void testConcat(){
        Stream<String> stream1 = Stream.of(arr1);
        Stream<String> stream2 = Stream.of(arr2);
        Stream.concat(stream1,stream2).distinct().forEach(System.out::println);
     }

3）聚合操做

@Before
public void init(){
    arr = new String[]{"b","ab","abc","abcd","abcde"};
}
 
/**
 * max、min
 * 最大最小值
 */
@Test
public void testMaxAndMin(){
    Stream.of(arr).max(Comparator.comparing(String::length)).ifPresent(System.out::println);
    Stream.of(arr).min(Comparator.comparing(String::length)).ifPresent(System.out::println);
}
 
/**
 * count
 * 計算數量
 */
@Test
public void testCount(){
    long count = Stream.of(arr).count();
    System.out.println(count);
}
 
/**
 * findFirst
 * 查找第一個
 */
@Test
public void testFindFirst(){
    String str =  Stream.of(arr).parallel().filter(x->x.length()>3).findFirst().orElse("noghing");
    System.out.println(str);
}
 
/**
 * findAny
 * 找到全部匹配的元素
 * 對並行流十分有效
 * 只要在任何片斷髮現了第一個匹配元素就會結束整個運算
 */
@Test
public void testFindAny(){
    Optional<String> optional = Stream.of(arr).parallel().filter(x->x.length()>3).findAny();
    optional.ifPresent(System.out::println);
}
 
/**
 * anyMatch
 * 是否含有匹配元素
 */
@Test
public void testAnyMatch(){
    Boolean aBoolean = Stream.of(arr).anyMatch(x->x.startsWith("a"));
    System.out.println(aBoolean);
}
 
 
@Test
public void testStream1() {
    Optional<Integer> optional = Stream.of(1,2,3).filter(x->x>1).reduce((x,y)->x+y);
    System.out.println(optional.get());
}

4）Optional類型

一般聚合操做會返回一個Optional類型，Optional表示一個安全的指定結果類型，所謂的安全指的是避免直接調用返回類型的null值而形成空指針異常，調用optional.ifPresent()能夠判斷返回值是否爲空，或者直接調用ifPresent(Consumer<? super T> consumer)在結果部位空時進行消費操做；調用optional.get()獲取返回值。一般的使用方式以下：

@Test
    public void testOptional() {
        List<String> list = new ArrayList<String>() {
            {
                add("user1");
                add("user2");
            }
        };
        Optional<String> opt = Optional.of("andy with u");
        opt.ifPresent(list::add);
        list.forEach(System.out::println);
    }

使用Optional能夠在沒有值時指定一個返回值，例如

@Test
public void testOptional2() {
    Integer[] arr = new Integer[]{4,5,6,7,8,9};
    Integer result = Stream.of(arr).filter(x->x>9).max(Comparator.naturalOrder()).orElse(-1);
    System.out.println(result);
    Integer result1 = Stream.of(arr).filter(x->x>9).max(Comparator.naturalOrder()).orElseGet(()->-1);
    System.out.println(result1);
    Integer result2 = Stream.of(arr).filter(x->x>9).max(Comparator.naturalOrder()).orElseThrow(RuntimeException::new);
    System.out.println(result2);
}

Optional的建立

採用Optional.empty()建立一個空的Optional，使用Optional.of()建立指定值的Optional。一樣也能夠調用Optional對象的map方法進行Optional的轉換，調用flatMap方法進行Optional的迭代

@Test
public void testStream1() {
    Optional<Student> studentOptional = Optional.of(new Student("user1",21));
    Optional<String> optionalStr = studentOptional.map(Student::getName);
    System.out.println(optionalStr.get());
}
 
public static Optional<Double> inverse(Double x) {
    return x == 0 ? Optional.empty() : Optional.of(1 / x);
}
 
public static Optional<Double> squareRoot(Double x) {
    return x < 0 ? Optional.empty() : Optional.of(Math.sqrt(x));
}
 
/**
 * Optional的迭代
 */
@Test
public void testStream2() {
    double x = 4d;
    Optional<Double> result1 = inverse(x).flatMap(StreamTest7::squareRoot);
    result1.ifPresent(System.out::println);
    Optional<Double> result2 = Optional.of(4.0).flatMap(StreamTest7::inverse).flatMap(StreamTest7::squareRoot);
    result2.ifPresent(System.out::println);
}

5）收集結果

Student[] students;
@Before
public void init(){
    students = new Student[100];
    for (int i=0;i<30;i++){
        Student student = new Student("user",i);
        students[i] = student;
    }
    for (int i=30;i<60;i++){
        Student student = new Student("user"+i,i);
        students[i] = student;
    }
    for (int i=60;i<100;i++){
        Student student = new Student("user"+i,i);
        students[i] = student;
    }
 
}
@Test
public void testCollect1(){
    /**
     * 生成List
     */
    List<Student> list = Arrays.stream(students).collect(toList());
    list.forEach((x)-> System.out.println(x));
    /**
     * 生成Set
     */
    Set<Student> set = Arrays.stream(students).collect(toSet());
    set.forEach((x)-> System.out.println(x));
    /**
     * 若是包含相同的key，則須要提供第三個參數，不然報錯
     */
    Map<String,Integer> map = Arrays.stream(students).collect(toMap(Student::getName,Student::getScore,(s,a)->s+a));
    map.forEach((x,y)-> System.out.println(x+"->"+y));
}
 
/**
 * 生成數組
 */
@Test
public void testCollect2(){
    Student[] s = Arrays.stream(students).toArray(Student[]::new);
    for (int i=0;i<s.length;i++)
        System.out.println(s[i]);
}
 
/**
 * 指定生成的類型
 */
@Test
public void testCollect3(){
    HashSet<Student> s = Arrays.stream(students).collect(toCollection(HashSet::new));
    s.forEach(System.out::println);
}
 
/**
 * 統計
 */
@Test
public void testCollect4(){
    IntSummaryStatistics summaryStatistics = Arrays.stream(students).collect(Collectors.summarizingInt(Student::getScore));
    System.out.println("getAverage->"+summaryStatistics.getAverage());
    System.out.println("getMax->"+summaryStatistics.getMax());
    System.out.println("getMin->"+summaryStatistics.getMin());
    System.out.println("getCount->"+summaryStatistics.getCount());
    System.out.println("getSum->"+summaryStatistics.getSum());
}

6）分組和分片

分組和分片的意義是，將collect的結果集展現位Map<key,val>的形式，一般的用法以下：

Student[] students;
@Before
public void init(){
    students = new Student[100];
    for (int i=0;i<30;i++){
        Student student = new Student("user1",i);
        students[i] = student;
    }
    for (int i=30;i<60;i++){
        Student student = new Student("user2",i);
        students[i] = student;
    }
    for (int i=60;i<100;i++){
        Student student = new Student("user3",i);
        students[i] = student;
    }
 
}
@Test
public void testGroupBy1(){
    Map<String,List<Student>> map = Arrays.stream(students).collect(groupingBy(Student::getName));
    map.forEach((x,y)-> System.out.println(x+"->"+y));
}
 
/**
 * 若是隻有兩類，使用partitioningBy會比groupingBy更有效率
 */
@Test
public void testPartitioningBy(){
    Map<Boolean,List<Student>> map = Arrays.stream(students).collect(partitioningBy(x->x.getScore()>50));
    map.forEach((x,y)-> System.out.println(x+"->"+y));
}
 
/**
 * downstream指定類型
 */
@Test
public void testGroupBy2(){
    Map<String,Set<Student>> map = Arrays.stream(students).collect(groupingBy(Student::getName,toSet()));
    map.forEach((x,y)-> System.out.println(x+"->"+y));
}
 
/**
 * downstream 聚合操做
 */
@Test
public void testGroupBy3(){
    /**
     * counting
     */
    Map<String,Long> map1 = Arrays.stream(students).collect(groupingBy(Student::getName,counting()));
    map1.forEach((x,y)-> System.out.println(x+"->"+y));
    /**
     * summingInt
     */
    Map<String,Integer> map2 = Arrays.stream(students).collect(groupingBy(Student::getName,summingInt(Student::getScore)));
    map2.forEach((x,y)-> System.out.println(x+"->"+y));
    /**
     * maxBy
     */
    Map<String,Optional<Student>> map3 = Arrays.stream(students).collect(groupingBy(Student::getName,maxBy(Comparator.comparing(Student::getScore))));
    map3.forEach((x,y)-> System.out.println(x+"->"+y));
    /**
     * mapping
     */
    Map<String,Set<Integer>> map4 = Arrays.stream(students).collect(groupingBy(Student::getName,mapping(Student::getScore,toSet())));
    map4.forEach((x,y)-> System.out.println(x+"->"+y));
}

5.原始類型流

在數據量比較大的狀況下，將基本數據類型（int,double...）包裝成相應對象流的作法是低效的，所以，咱們也能夠直接將數據初始化爲原始類型流，在原始類型流上的操做與對象流相似，咱們只須要記住兩點

1.原始類型流的初始化

2.原始類型流與流對象的轉換

DoubleStream doubleStream;
    IntStream intStream;
 
    /**
     * 原始類型流的初始化
     */
    @Before
    public void testStream1(){
 
        doubleStream = DoubleStream.of(0.1,0.2,0.3,0.8);
        intStream = IntStream.of(1,3,5,7,9);
        IntStream stream1 = IntStream.rangeClosed(0,100);
        IntStream stream2 = IntStream.range(0,100);
    }
 
    /**
     * 流與原始類型流的轉換
     */
    @Test
    public void testStream2(){
        Stream<Double> stream = doubleStream.boxed();
        doubleStream = stream.mapToDouble(Double::new);
    }

6.並行流

能夠將普通順序執行的流轉變爲並行流，只須要調用順序流的parallel() 方法便可，如Stream.iterate(1, x -> x + 1).limit(10).parallel()。

1）並行流的執行順序

咱們調用peek方法來瞧瞧並行流和串行流的執行順序，peek方法顧名思義，就是偷窺流內的數據，peek方法聲明爲Stream<T> peek(Consumer<? super T> action);加入打印程序能夠觀察到經過流內數據，見以下代碼：

public void peek1(int x) {
        System.out.println(Thread.currentThread().getName() + ":->peek1->" + x);
    }
 
    public void peek2(int x) {
        System.out.println(Thread.currentThread().getName() + ":->peek2->" + x);
    }
 
    public void peek3(int x) {
        System.out.println(Thread.currentThread().getName() + ":->final result->" + x);
    }
 
    /**
     * peek，監控方法
     * 串行流和並行流的執行順序
     */
    @org.junit.Test
    public void testPeek() {
        Stream<Integer> stream = Stream.iterate(1, x -> x + 1).limit(10);
        stream.peek(this::peek1).filter(x -> x > 5)
                .peek(this::peek2).filter(x -> x < 8)
                .peek(this::peek3)
                .forEach(System.out::println);
    }
 
    @Test
    public void testPeekPal() {
        Stream<Integer> stream = Stream.iterate(1, x -> x + 1).limit(10).parallel();
        stream.peek(this::peek1).filter(x -> x > 5)
                .peek(this::peek2).filter(x -> x < 8)
                .peek(this::peek3)
                .forEach(System.out::println);
    }

串行流打印結果以下：

並行流打印結果以下：

咋看不必定能看懂，咱們用以下的圖來解釋

咱們將stream.filter(x -> x > 5).filter(x -> x < 8).forEach(System.out::println)的過程想象成上圖的管道，咱們在管道上加入的peek至關於一個閥門，透過這個閥門查看流經的數據，

1）當咱們使用順序流時，數據按照源數據的順序依次經過管道，當一個數據被filter過濾，或者通過整個管道而輸出後，第二個數據纔會開始重複這一過程

2）當咱們使用並行流時，系統除了主線程外啓動了七個線程（個人電腦是4核八線程）來執行處理任務，所以執行是無序的，但同一個線程內處理的數據是按順序進行的。

2） sorted()、distinct()等對並行流的影響

sorted()、distinct()是元素相關方法，和總體的數據是有關係的,map，filter等方法和已經經過的元素是不相關的,不須要知道流裏面有哪些元素，並行執行和sorted會不會產生衝突呢？

結論：1.並行流和排序是不衝突的，2.一個流是不是有序的，對於一些api可能會提升執行效率，對於另外一些api可能會下降執行效率

3.若是想要輸出的結果是有序的，對於並行的流須要使用forEachOrdered(forEach的輸出效率更高)

咱們作以下實驗：

/**
 * 生成一億條0-100之間的記錄
 */
@Before
public void init() {
    Random random = new Random();
    list = Stream.generate(() -> random.nextInt(100)).limit(100000000).collect(toList());
}
 
/**
 * tip
 */
@org.junit.Test
public void test1() {
    long begin1 = System.currentTimeMillis();
    list.stream().filter(x->(x > 10)).filter(x->x<80).count();
    long end1 = System.currentTimeMillis();
    System.out.println(end1-begin1);
    list.stream().parallel().filter(x->(x > 10)).filter(x->x<80).count();
    long end2 = System.currentTimeMillis();
    System.out.println(end2-end1);
 
    long begin1_ = System.currentTimeMillis();
    list.stream().filter(x->(x > 10)).filter(x->x<80).distinct().sorted().count();
    long end1_ = System.currentTimeMillis();
    System.out.println(end1-begin1);
    list.stream().parallel().filter(x->(x > 10)).filter(x->x<80).distinct().sorted().count();
    long end2_ = System.currentTimeMillis();
    System.out.println(end2_-end1_);
 
}

可見，對於串行流.distinct().sorted()方法對於運行時間沒有影響，可是對於串行流，會使得運行時間大大增長，所以對於包含sorted、distinct()等與全局數據相關的操做，不推薦使用並行流。

7.stream vs spark rdd

最初看到stream的一個直觀感覺是和spark像，真的像

val count = sc.parallelize(1 to NUM_SAMPLES).filter { _ =>
  val x = math.random
  val y = math.random
  x*x + y*y < 1}.count()println(s"Pi is roughly ${4.0 * count / NUM_SAMPLES}")

以上代碼摘自spark官網，使用的是scala語言，一個最基礎的word count代碼，這裏咱們簡單介紹一下spark，spark是當今最流行的基於內存的大數據處理框架，spark中的一個核心概念是RDD（彈性分佈式數據集），將分佈於不一樣處理器上的數據抽象成rdd，rdd上支持兩種類型的操做1） Transformation（變換）2） Action（行動），對於rdd的Transformation算子並不會當即執行，只有當使用了Action算子後，纔會觸發。

關於java8中的stream和spark的類似與不一樣咱們以後專門介紹。