22.Java Stream API（Stream 的获取方式、Stream 常用方法、Stream 数据收集、Stream 数据计算、Stream 并行流）

一、问题分析

1、集合处理数据的弊端

• 在需要对集合中的元素进行操作时，集合遍历是典型的操作

package com.my.stream;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

public class StreamTest1 {
    public static void main(String[] args) {
        List<String> list = Arrays.asList("jack", "tom", "smith", "mery");
        System.out.println(list);

        // 获取名称中有 t 的用户
        ArrayList<String> arrayList1 = new ArrayList<>();
        for (String s : list) {
            if (s.contains("t")) {
                arrayList1.add(s);
            }
        }
        System.out.println(arrayList1);

        // 获取名称长度为 4 的用户
        ArrayList<String> arrayList2 = new ArrayList<>();
        for (String s : list) {
            if (s.length() == 4) {
                arrayList2.add(s);
            }
        }
        System.out.println(arrayList2);
    }
}

2、解决方案

• 针对不同的需求总是会进行遍历操作，可以通过 Stream API 进行高效处理

package com.my.stream;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

public class StreamTest2 {
    public static void main(String[] args) {
        List<String> list = Arrays.asList("jack", "tom", "smith", "mery");
        System.out.println(list);

        // 获取名称中有 t 的用户
        ArrayList<String> arrayList1 = new ArrayList<>();
        list.stream()
                .filter(s -> s.contains("t"))
                .forEach(s -> arrayList1.add(s));
        System.out.println(arrayList1);

        // 获取名称长度为 4 的用户
        ArrayList<String> arrayList2 = new ArrayList<>();
        list.stream()
                .filter(s -> s.length() == 4)
                .forEach(s -> arrayList2.add(s));
        System.out.println(arrayList2);
    }
}

---

二、Stream 的获取方式

1、Collection 的默认方法 stream

（1）基本介绍

• java.util.Collection 接口中存在默认方法 stream

default Stream<E> stream() {
    return StreamSupport.stream(spliterator(), false);
}

（2）演示

• 所有 Collection 接口的实现都可以通过 stream 方法来获取 Stream

ArrayList<Object> arrayList = new ArrayList<>();
Stream<Object> stream1 = arrayList.stream();
System.out.println(stream1);

HashSet<Object> hashSet = new HashSet<>();
Stream<Object> stream2 = hashSet.stream();
System.out.println(stream2);

Vector<Object> vector = new Vector<>();
Stream<Object> stream3 = vector.stream();
System.out.println(stream3);

• Map 接口实现获取 Stream

HashMap<Object, Object> hashMap = new HashMap<>();

Stream<Object> stream4 = hashMap.keySet().stream();
System.out.println(stream4);

Stream<Object> stream5 = hashMap.values().stream();
System.out.println(stream5);

Stream<Map.Entry<Object, Object>> stream6 = hashMap.entrySet().stream();
System.out.println(stream6);

2、Stream 的静态方法 of

（1）基本介绍

• 在实际开发中会不可避免的使用到数组操作，Stream 接口中提供了默认方法 of 来对数组进行相关操作

public static<T> Stream<T> of(T t) {
    return StreamSupport.stream(new Streams.StreamBuilderImpl<>(t), false);
}

@SafeVarargs
@SuppressWarnings("varargs") // Creating a stream from an array is safe
public static<T> Stream<T> of(T... values) {
    return Arrays.stream(values);
}

（2）演示

• 数组获取 Stream

Stream<String> stream1 = Stream.of("a1", "a2", "a3");
System.out.println(stream1);

String[] arr1 = {"aa", "bb", "cc"};
Stream<String> stream2 = Stream.of(arr1);
System.out.println(stream2);

// 不能使用基本数据类型的数组
Integer[] arr2 = {1, 2, 3, 4};
Stream<Integer> stream3 = Stream.of(arr2);
System.out.println(stream3);

---

三、Stream 方法概述

1、终结方法

• 返回值不是 Stream 类型的方法，不再支持链式调用

• 终结方法包括 forEach、count、match 方法

2、非终结方法

• 返回值是 Stream 类型的方法，支持链式调用

• 除了终结方法外，其余方法均为非终结方法

---

四、Stream 常用方法

1、forEach 方法

（1）基本介绍

• 用来遍历流中的元素

void forEach(Consumer<? super T> action);

• 该方法接收一个 Consumer 函数式接口，将流中每一个元素交给 Consumer 处理

（2）演示

Stream<String> stream = Stream.of("a1", "a2", "a3");

stream.forEach(s -> {
    System.out.println(s);
});

2、count 方法

（1）基本介绍

• 用来统计流中元素的个数

long count();

（2）演示

Stream<String> stream = Stream.of("a1", "a2", "a3");

long count = stream.count();
System.out.println(count);

3、filter 方法

（1）基本介绍

• 过滤流中的元素

Stream<T> filter(Predicate<? super T> predicate);

• 该方法接收一个 Predicate 函数式接口，将流中每一个元素交给 Predicate 处理

（2）演示

Stream<String> stream = Stream.of("a1", "a2", "a3");

stream.filter(s -> {
    return s.contains("1");
}).forEach(s -> {
    System.out.println(s);
});

4、limit 方法

（1）基本介绍

• 可以对流中的元素进行截取，只取前 N 个元素

Stream<T> limit(long maxSize);

（2）演示

stream.limit(2).forEach(s -> System.out.println(s));

5、skip 方法

（1）基本介绍

• 跳过流中的前 N 个元素

Stream<T> skip(long n);

（2）演示

stream.skip(2).forEach(s -> System.out.println(s));

6、map 相关方法

（1）基本介绍

• 将当前流中的元素的数据类型转换成另一种数据类型

<R> Stream<R> map(Function<? super T, ? extends R> mapper);

// 将当前流中的元素的数据类型转换成 int 类型
IntStream mapToInt(ToIntFunction<? super T> mapper);

// 将当前流中的元素的数据类型转换成 long 类型
LongStream mapToLong(ToLongFunction<? super T> mapper);

// 将当前流中的元素的数据类型转换成 double 类型
DoubleStream mapToDouble(ToDoubleFunction<? super T> mapper);

• 该方法接收一个 Function 函数式接口，将流中每一个元素交给 Function 处理

（2）演示

Stream<String> stream = Stream.of("10", "20", "30", "40", "50");

stream.map(s -> Integer.parseInt(s))
        .forEach(num -> {
            num += 5;
            System.out.println(num);
        });

• 将当前流中的元素的数据类型转换成 int 类型

Stream<String> stream = Stream.of("10", "20", "30", "40", "50");

IntStream intStream = stream.map(Integer::parseInt).mapToInt(num -> num.intValue());
intStream.forEach(System.out::println);

• 将当前流中的元素的数据类型转换成 long 类型

Stream<String> stream = Stream.of("10", "20", "30", "40", "50");

LongStream longStream = stream.map(Integer::parseInt).mapToLong(num -> num.longValue());
longStream.forEach(System.out::println);

• 将当前流中的元素的数据类型转换成 double 类型

Stream<String> stream = Stream.of("10", "20", "30", "40", "50");

DoubleStream doubleStream = stream.map(Integer::parseInt).mapToDouble(num -> num.doubleValue());
doubleStream.forEach(System.out::println);

7、sort 相关方法

（1）基本介绍

• 将流中的元素进行排序

// 自然排序
Stream<T> sorted();

// 指定规则排序
Stream<T> sorted(Comparator<? super T> comparator);

（2）演示

• 自然排序

Stream<String> stream = Stream.of("6", "5", "1", "-7");

stream.map(Integer::parseInt)
        .sorted()
        .forEach(System.out::println);

• 指定规则排序

Stream<String> stream = Stream.of("6", "5", "1", "-7");

stream.map(Integer::parseInt)
        .sorted((o1, o2) -> o2 - o1)
        .forEach(System.out::println);

8、distinct 方法

（1）基本介绍

• 去除流中的重复数据

Stream<T> distinct();

（2）演示

• 基本数据类型去重

Stream<String> stream = Stream.of("1", "1", "2", "2");

stream.distinct().forEach(System.out::println);

• 引用数据类型去重（需要重写 equals 和 hashCode 方法）

package com.my.stream;

import java.util.Objects;
import java.util.stream.Stream;

public class StreamMethodTest {
    public static void main(String[] args) {
        Stream<User> stream = Stream.of(
                new User("jack", 20),
                new User("jack", 20),
                new User("tom", 15));

        stream.distinct().forEach(System.out::println);
    }
}

class User {
    private String name;
    private Integer age;

    public String getName() {
        return name;
    }

    public void setName(String name) {
        this.name = name;
    }

    public Integer getAge() {
        return age;
    }

    public void setAge(Integer age) {
        this.age = age;
    }

    public User(String name, Integer age) {
        this.name = name;
        this.age = age;
    }

    @Override
    public boolean equals(Object o) {
        if (this == o) return true;
        if (o == null || getClass() != o.getClass()) return false;
        User user = (User) o;
        return Objects.equals(name, user.name) &&
                Objects.equals(age, user.age);
    }

    @Override
    public int hashCode() {
        return Objects.hash(name, age);
    }

    @Override
    public String toString() {
        return "User{" +
                "name='" + name + '\'' +
                ", age=" + age +
                '}';
    }
}

9、match 相关方法

（1）基本介绍

• 匹配流中的数据

// 是否有任意一个元素满足条件
boolean anyMatch(Predicate<? super T> predicate);

// 是否所有元素都满足条件
boolean allMatch(Predicate<? super T> predicate);

// 是否所有元素都不满足条件
boolean noneMatch(Predicate<? super T> predicate);

（2）演示

Stream<String> stream = Stream.of("6", "5", "1", "-7");

boolean b = stream.map(Integer::parseInt).anyMatch(num -> num > 3);
System.out.println(b);

Stream<String> stream = Stream.of("6", "5", "1", "-7");

boolean b = stream.map(Integer::parseInt).allMatch(num -> num > 3);
System.out.println(b);

Stream<String> stream = Stream.of("6", "5", "1", "-7");

boolean b = stream.map(Integer::parseInt).noneMatch(num -> num > 3);
System.out.println(b);

10、find 相关方法

（1）基本介绍

• 获取流中的元素

// 获取流中第一个元素
Optional<T> findFirst();

// 在默认情况下获取流中第一个元素，在并行流时返回流中随机一个元素
Optional<T> findAny();

（2）演示

Stream<String> stream = Stream.of("6", "5", "1", "-7");

Optional<String> first = stream.findFirst();
System.out.println(first.get());

Stream<String> stream = Stream.of("6", "5", "1", "-7");

Optional<String> any = stream.findAny();
System.out.println(any.get());

11、max 和 min 方法

（1）基本介绍

• 根据规则，获取流中的最大元素

Optional<T> max(Comparator<? super T> comparator);

• 根据规则，获取流中的最小元素

Optional<T> min(Comparator<? super T> comparator);

（2）演示

Stream<String> stream = Stream.of("6", "5", "1", "-7");

Optional<String> max = stream.max((s1, s2) -> {
    return Integer.parseInt(s1) - Integer.parseInt(s2);
});
System.out.println(max.get());

Stream<String> stream = Stream.of("6", "5", "1", "-7");

Optional<String> min = stream.min((s1, s2) -> {
    return Integer.parseInt(s1) - Integer.parseInt(s2);
});
System.out.println(min.get());

12、reduce 方法

（1）基本介绍

• 将流中所有元素归纳为一个元素

T reduce(T identity, BinaryOperator<T> accumulator);

（2）演示

• 求和

Stream<String> stream = Stream.of("6", "5", "1", "-7");

Integer sum = stream.map(Integer::parseInt)
        .reduce(0, (x, y) -> x + y);
System.out.println(sum);

• 统计字符 a 出现的次数

Stream<String> stream = Stream.of("a", "b", "c", "d", "a");

Integer sum = stream.map(s -> s.equals("a") ? 1 : 0).reduce(0, (x, y) -> x + y);
System.out.println(sum);

13、concat 方法

（1）基本介绍

• 静态方法，将两个流合并为一个流

public static <T> Stream<T> concat(Stream<? extends T> a, Stream<? extends T> b) {
    Objects.requireNonNull(a);
    Objects.requireNonNull(b);

    @SuppressWarnings("unchecked")
    Spliterator<T> split = new Streams.ConcatSpliterator.OfRef<>(
            (Spliterator<T>) a.spliterator(), (Spliterator<T>) b.spliterator());
    Stream<T> stream = StreamSupport.stream(split, a.isParallel() || b.isParallel());
    return stream.onClose(Streams.composedClose(a, b));
}

（2）演示

Stream<String> stream1 = Stream.of("a", "b", "c");
Stream<String> stream2 = Stream.of("d", "e", "f");

Stream<String> concatStream = Stream.concat(stream1, stream2);
concatStream.forEach(System.out::println);

---

五、Stream 常用方法综合案例

1、需求

• 定义两个集合，在集合中存储多个学生的名称，完成如下操作

1. 第一个集合只保留名字长度为 3 的前三个学生

2. 第二个集合只保留以 t 开头的后两个学生

3. 将两个集合合并为一个集合

4. 根据姓名创建 Student 对象

5. 打印整个集合

2、具体实现

package com.my.stream;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;

import java.util.Arrays;
import java.util.List;
import java.util.stream.Stream;

public class StreamMethodTest2 {
    public static void main(String[] args) {
        List<String> list1 = Arrays.asList("jack", "tom", "smith", "jem", "mei", "len");
        List<String> list2 = Arrays.asList("timi", "serry", "tomas", "terry");

        // 1. 第一个集合只保留名字长度为 3 的前三个学生
        Stream<String> stream1 = list1.stream().filter(s -> s.length() == 3).limit(3);

        // 2. 第二个集合只保留以 t 开头的前两个学生
        Stream<String> stream2 = list2.stream().filter(s -> s.startsWith("t")).limit(2);

        // 3. 将两个集合合并为一个集合
        Stream<String> concatStream = Stream.concat(stream1, stream2);

        // 4. 根据姓名创建 Student 对象
        // 5. 打印整个集合
        concatStream.map(s -> {
            Student student = new Student();
            student.setName(s);
            return student;
        }).forEach(System.out::println);
    }
}

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

---

六、Stream 数据收集

1、将结果收集到集合中

• 将结果收集到 List 集合中

Stream<String> stream = Stream.of("aa", "bb", "cc", "aa");

List<String> list = stream.collect(Collectors.toList());
System.out.println(list);

• 将结果收集到 Set 集合中

Stream<String> stream = Stream.of("aa", "bb", "cc", "aa");

Set<String> set = stream.collect(Collectors.toSet());
System.out.println(set);

• 将结构收集到具体实现 ArrayList 集合中

Stream<String> stream = Stream.of("aa", "bb", "cc", "aa");

ArrayList<String> arrayList = stream.collect(Collectors.toCollection(ArrayList::new));
System.out.println(arrayList);

• 将结构收集到具体实现 HashSet 集合中

Stream<String> stream = Stream.of("aa", "bb", "cc", "aa");

HashSet<String> hashSet = stream.collect(Collectors.toCollection(HashSet::new));
System.out.println(hashSet);

2、将数据收集到数组中

• 收集到 Object 类型的数组中

Stream<String> stream = Stream.of("aa", "bb", "cc", "aa");

Object[] objects = stream.toArray();
System.out.println(Arrays.toString(objects));

• 收集到 String 类型的数组中

Stream<String> stream = Stream.of("aa", "bb", "cc", "aa");

String[] strings = stream.toArray(s -> new String[s]);
System.out.println(Arrays.toString(strings));

---

七、Stream 数据计算

1、聚合计算

• 准备数据

package com.my.stream;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;

import java.util.Optional;
import java.util.stream.Collectors;
import java.util.stream.Stream;

public class StreamPolymerizeTest {
    public static void main(String[] args) {
        Stream<Person> personStream = Stream.of(
                new Person("jack", 20),
                new Person("tom", 23),
                new Person("smith", 34),
                new Person("jerry", 15),
                new Person("serry", 21)
        );
    }
}

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

• 获取最大值

Optional<Person> ageMax = personStream.collect(Collectors.maxBy((o1, o2) -> o1.getAge() - o2.getAge()));
System.out.println("最大年龄为 " + ageMax.get().getAge());

• 获取最小值

Optional<Person> ageMin = personStream.collect(Collectors.minBy((o1, o2) -> o1.getAge() - o2.getAge()));
System.out.println("最小年龄为 " + ageMin.get().getAge());

• 求和

Integer ageSum = personStream.collect(Collectors.summingInt(p -> p.getAge()));
System.out.println("年龄总和为 " + ageSum);

• 获取平均值

Double ageAverage = personStream.collect(Collectors.averagingInt(p -> p.getAge()));
System.out.println("平均年龄为 " + ageAverage);

• 统计数量

Long personSum = personStream.collect(Collectors.counting());
System.out.println("人员总数为 " + personSum);

2、分组计算

• 准备数据

package com.my.stream;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;

import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;
import java.util.stream.Stream;

public class StreamGroupTest {
    public static void main(String[] args) {
        Stream<Boy> boyStream = Stream.of(
                new Boy("张三", 15, 20.0),
                new Boy("李四", 19, 40.0),
                new Boy("王五", 13, 20.0),
                new Boy("赵六", 12, 40.0),
                new Boy("孙七", 25, 50.0)
        );
    }
}

@Data
@AllArgsConstructor
@NoArgsConstructor
class Boy {
    private String name;
    private Integer age;
    private Double score;
}

• 根据分数分组

Map<Double, List<Boy>> scoreMap = boyStream.collect(Collectors
        .groupingBy(b -> b.getScore()));
scoreMap.forEach((k, v) -> System.out.println(k + "  " + v));

• 根据年龄分组

Map<String, List<Boy>> ageMap = boyStream.collect(Collectors
        .groupingBy(b -> b.getAge() >= 18 ? "成年" : "未成年"));
ageMap.forEach((k, v) -> System.out.println(k + "  " + v));

• 先根据年龄分组，再根据分数分组

Map<Double, Map<String, List<Boy>>> map = boyStream.collect(Collectors
        .groupingBy(Boy::getScore, Collectors.groupingBy(b -> b.getAge() >= 18 ? "成年" : "未成年")));
map.forEach((k, v) -> {
    System.out.println(k);
    v.forEach((x, y) -> System.out.println(x + "  " + y));
});

3、分区操作

• 准备数据

package com.my.stream;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;

import java.util.stream.Stream;

public class StreamPartTest {
    public static void main(String[] args) {
        Stream<Dog> dogStream = Stream.of(
                new Dog("tom", 5),
                new Dog("jack", 4),
                new Dog("jerry", 3),
                new Dog("mery", 2)
        );
    }
}

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

• 根据年龄分区

// partitioningBy 方法会根据元素是否满足条件分组，分为 true 组，和 false 组
Map<Boolean, List<Dog>> ageMap = dogStream.collect(Collectors
        .partitioningBy(s -> s.getAge() > 3));
ageMap.forEach((k, v) -> System.out.println(k + "  " + v));

4、拼接操作

• 准备数据

package com.my.stream;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;

import java.util.stream.Collectors;
import java.util.stream.Stream;

public class StreamSplitTest {
    public static void main(String[] args) {
        Stream<Cat> catStream = Stream.of(
                new Cat("jack", 2),
                new Cat("tom", 3),
                new Cat("smith", 4)
        );
    }
}

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

• 名字拼接

String str = catStream.map(c -> c.getName()).collect(Collectors.joining());
System.out.println(str);

• 名字拼接（带分隔符）

String str = catStream.map(c -> c.getName()).collect(Collectors
        .joining("-"));
System.out.println(str);

• 名字拼接（带分隔符、前缀、后缀）

String str = catStream.map(c -> c.getName()).collect(Collectors
        .joining("-", "Hello ", " World"));
System.out.println(str);

---

八、Stream 并行流

1、Stream 串行流

• 前面使用的 Stream 流都是串行，即在同一个线程上执行

package com.my.stream;

import java.util.stream.Stream;

public class StreamSerialTest {
    public static void main(String[] args) {
        Stream<Integer> integerStream = Stream.of(1, 2, 3, 4, 5);
        integerStream.filter(num -> {
            System.out.println(Thread.currentThread() + "  " + num);
            return num > 3;
        }).count();
    }
}

2、获取并行流

（1）Collection 的默认方法 parallelStream

ArrayList<String> arrayList = new ArrayList<>();
Stream<String> parallelStream = arrayList.parallelStream();

（2）Stream 的方法 parallel

Stream<Integer> stream = Stream.of(1, 2, 3);
Stream<Integer> parallelStream = stream.parallel();

3、并行流操作

Stream.of(1, 2, 3, 4, 5).parallel()
        .filter(num -> {
            System.out.println(Thread.currentThread() + "  " + num);
            return num > 3;
        }).count();

4、串行流和并行流对比

（1）对比案例

• 对比 for 循环、串行流和并行流的累加耗时

package com.my.stream;

import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.util.stream.LongStream;

public class StreamContrastTest {
    private static long times = 500000000;
    private long start;
    private long end;

    @Before
    public void before() {
        start = System.currentTimeMillis();
    }

    @After
    public void after() {
        end = System.currentTimeMillis();
        System.out.println("耗时：" + (end - start));
    }

    // 耗时：162
    @Test
    public void forTest() {
        System.out.println("for 循环");
        int sum = 0;
        for (long i = 0; i < times; i++) {
            sum += i;
        }
    }

    // 耗时：211
    @Test
    public void serialTest() {
        System.out.println("串行流");
        LongStream.rangeClosed(0, times).reduce(0, (x, y) -> x + y);
    }

    // 耗时：89
    @Test
    public void parallelTest() {
        System.out.println("并行流");
        LongStream.rangeClosed(0, times).parallel().reduce(0, (x, y) -> x + y);
    }
}

（2）小节

• 并行流效率最高，Stream 并行处理的过程是分而治之，将大任务分成多个小任务，每个任务都是一个线程操作

5、线程安全问题

（1）问题分析

ArrayList<Object> arrayList = new ArrayList<>();
IntStream.rangeClosed(1, 1000)
        .parallel()
        .forEach(num -> arrayList.add(num));
System.out.println(arrayList.size());

• 运行结果不是 arrayList 的大小不满 1000，就是直接抛出异常（java.lang.ArrayIndexOutOfBoundsException）

（2）解决方案

• 加同步锁

ArrayList<Object> arrayList = new ArrayList<>();
Object obj = new Object();
IntStream.rangeClosed(1, 100000)
        .forEach(num -> {
            synchronized (obj) {
                arrayList.add(num);
            }
        });
System.out.println(arrayList.size());

• 使用线程安全的容器

Vector<Object> vector = new Vector<>();
IntStream.rangeClosed(1, 1000)
        .parallel()
        .forEach(num -> vector.add(num));
System.out.println(vector.size());

• 将线程不安全的容器包装成线程安全的容器

ArrayList<Object> arrayList = new ArrayList<>();
List<Object> synchronizedList = Collections.synchronizedList(arrayList);
IntStream.rangeClosed(1, 1000)
        .parallel()
        .forEach(num -> synchronizedList.add(num));
System.out.println(synchronizedList.size());

• 使用 Stream 中的 collect 方法

List<Integer> list = IntStream.rangeClosed(1, 1000)
        .parallel()
        .boxed()
        .collect(Collectors.toList());
System.out.println(list.size());