概念
Stream 不是 集合元素，也不是数据结构，它相当于一个 高级版本的 Iterator，不可以重复遍历里面的数据，像水一样，流过了就一去不复返。它和普通的 Iterator 不同的是，它可以并行遍历，普通的 Iterator 只能是串行，在一个线程中执行。
内部迭代和外部迭代
使用for等进行迭代我们叫做外部迭代，使用stream流迭代叫做内部迭代，内部迭代有什么好处，当数量很大是我们不需要对数据进行拆分。
中间操作和终止操作
中间操作：返回stream的操作
终止操作：得到特定的结果
惰性求值：终止没有调用的情况下，中间操作不会执行
import java.util.stream.IntStream;
public class StreamDemo1 {
public static void main(String[] args) {
int[] nums = { 1, 2, 3 };
// 外部迭代
int sum = 0;
for (int i : nums) {
sum += i;
}
System.out.println("结果为：" + sum);
// 使用stream的内部迭代
// map就是中间操作（返回stream的操作）
// sum就是终止操作
int sum2 = IntStream.of(nums).map(StreamDemo1::doubleNum).sum();
System.out.println("结果为：" + sum2);
System.out.println("惰性求值就是终止没有调用的情况下，中间操作不会执行");
IntStream.of(nums).map(StreamDemo1::doubleNum);
}
public static int doubleNum(int i) {
System.out.println("执行了乘以2");
return i * 2;
}
}
流的创建
集合创建：使用collection.stream()、collection.parallelStream()（parallelStream()是函数接口提供者接口的一个特殊接口提供一个流的接口）
数组创建：Arrays.stream(数组);
数字流：IntStream.of()、IntStream.rangeClosed(1,10)创建一个1到10数字流。使用random创建一个无限流，无限流会报错我们必须限制这个流的大小new Random().ints().limit(10)
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Random;
import java.util.stream.IntStream;
import java.util.stream.Stream;
public class StreamDemo2 {
public static void main(String[] args) {
List<String> list = new ArrayList<>();
// 从集合创建
list.stream();
list.parallelStream();
// 从数组创建
Arrays.stream(new int[] { 2, 3, 5 });
// 创建数字流
IntStream.of(1, 2, 3);
IntStream.rangeClosed(1, 10);
// 使用random创建一个无限流
new Random().ints().limit(10);
Random random = new Random();
// 自己产生流
Stream.generate(() -> random.nextInt()).limit(20);
}
}
流的中间操作
中间操作主要包含2类：有状态操作和无状态操作。
有状态操作：当前操作与其他操作有依赖关系。(一般而言有状态操作都是有2个参数传入)
无状态操作：当前操作与其他操作没有依赖关系。
相关示例直接看代码
import java.util.Random;
import java.util.stream.Collectors;
import java.util.stream.Stream;
public class StreamDemo3 {
public static void main(String[] args) {
String str = "my name is 007";
// 把每个单词的长度大于2的单词的长度打印出来出来
Stream.of(str.split(" ")).filter(s -> s.length() > 2)
.map(s -> s.length()).forEach(System.out::println);
// flatMap A->B属性(是个集合), 最终得到所有的A元素里面的所有B属性集合
// intStream/longStream 并不是Stream的子类, 所以要进行装箱 boxed
Stream.of(str.split(" ")).flatMap(s -> s.chars().boxed())
.forEach(i -> System.out.println((char) i.intValue()));
// peek 用于debug. 是个中间操作,和 forEach 是终止操作
System.out.println("--------------peek------------");
Stream.of(str.split(" ")).peek(System.out::println)
.forEach(System.out::println);
// limit 使用, 主要用于无限流
new Random().ints().filter(i -> i > 100 && i < 1000).limit(10)
.forEach(System.out::println);
}
}
终止操作
包括短路操作和非短路操作
相关api直接看代码
import java.util.List;
import java.util.Optional;
import java.util.OptionalInt;
import java.util.Random;
import java.util.stream.Collectors;
import java.util.stream.Stream;
public class StreamDemo4 {
public static void main(String[] args) {
String str = "my name is 007";
// 使用并行流
str.chars().parallel().forEach(i -> System.out.print((char) i));
System.out.println();
// 使用 forEachOrdered 保证顺序
str.chars().parallel().forEachOrdered(i -> System.out.print((char) i));
// 收集到list
List<String> list = Stream.of(str.split(" "))
.collect(Collectors.toList());
System.out.println(list);
// 使用 reduce 拼接字符串
Optional<String> letters = Stream.of(str.split(" "))
.reduce((s1, s2) -> s1 + "|" + s2);
System.out.println(letters.orElse(""));
// 带初始化值的reduce
String reduce = Stream.of(str.split(" ")).reduce("",
(s1, s2) -> s1 + "|" + s2);
System.out.println(reduce);
// 计算所有单词总长度
Integer length = Stream.of(str.split(" ")).map(s -> s.length())
.reduce(0, (s1, s2) -> s1 + s2);
System.out.println(length);
// max 的使用
Optional<String> max = Stream.of(str.split(" "))
.max((s1, s2) -> s1.length() - s2.length());
System.out.println(max.get());
// 使用 findFirst 短路操作
OptionalInt findFirst = new Random().ints().findFirst();
System.out.println(findFirst.getAsInt());
}
}
并行流
先看下不使用并行操作的方式
import java.util.List;
import java.util.Optional;
import java.util.OptionalInt;
import java.util.Random;
import java.util.stream.Collectors;
import java.util.stream.Stream;
public class StreamDemo4 {
public static void main(String[] args) {
// 调用parallel 产生一个并行流
IntStream.range(1, 100).parallel().peek(StreamDemo5::debug).count();
}
public static void debug(int i) {
System.out.println(Thread.currentThread().getName() + " debug " + i);
TimeUnit.SECONDS.sleep(3);
}
}
运行结果为debug1 debug2 debug3 。。。
如果采用并行的方式
IntStream.range(1, 100).parallel().peek(StreamDemo5::debug).count();
则debug后的数字是乱序的
多次调用 parallel / sequential, 以最后一次调用为准.
// 多次调用 parallel / sequential, 以最后一次调用为准.
IntStream.range(1, 100)
// 调用parallel产生并行流
.parallel().peek(StreamDemo5::debug)
// 调用sequential 产生串行流
.sequential().peek(StreamDemo5::debug2)
.count();
并行流使用的线程池: ForkJoinPool.commonPool
默认的线程数是 当前机器的cpu个数
使用这个属性可以修改默认的线程数
System.setProperty("java.util.concurrent.ForkJoinPool.common.parallelism","20");
使用自己的线程池, 不使用默认线程池, 防止任务被阻塞
// 线程名字 : ForkJoinPool-1
ForkJoinPool pool = new ForkJoinPool(20);
pool.submit(() -> IntStream.range(1, 100).parallel()
.peek(StreamDemo5::debug).count());
pool.shutdown();
synchronized (pool) {
try {
pool.wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
收集器
将处理后的流收集到集合中
api的演示看代码
import java.util.Arrays;
import java.util.IntSummaryStatistics;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.TreeSet;
import java.util.stream.Collectors;
import org.apache.commons.collections4.MapUtils;
/**
* 学生 对象
*/
class Student {
/**
* 姓名
*/
private String name;
/**
* 年龄
*/
private int age;
/**
* 性别
*/
private Gender gender;
/**
* 班级
*/
private Grade grade;
public Student(String name, int age, Gender gender, Grade grade) {
super();
this.name = name;
this.age = age;
this.gender = gender;
this.grade = grade;
}
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
public int getAge() {
return age;
}
public void setAge(int age) {
this.age = age;
}
public Grade getGrade() {
return grade;
}
public void setGrade(Grade grade) {
this.grade = grade;
}
public Gender getGender() {
return gender;
}
public void setGender(Gender gender) {
this.gender = gender;
}
@Override
public String toString() {
return "[name=" + name + ", age=" + age + ", gender=" + gender
+ ", grade=" + grade + "]";
}
}
/**
* 性别
*/
enum Gender {
MALE, FEMALE
}
/**
* 班级
*/
enum Grade {
ONE, TWO, THREE, FOUR;
}
public class CollectDemo {
public static void main(String[] args) {
// 测试数据
List<Student> students = Arrays.asList(
new Student("小明", 10, Gender.MALE, Grade.ONE),
new Student("大明", 9, Gender.MALE, Grade.THREE),
new Student("小白", 8, Gender.FEMALE, Grade.TWO),
new Student("小黑", 13, Gender.FEMALE, Grade.FOUR),
new Student("小红", 7, Gender.FEMALE, Grade.THREE),
new Student("小黄", 13, Gender.MALE, Grade.ONE),
new Student("小青", 13, Gender.FEMALE, Grade.THREE),
new Student("小紫", 9, Gender.FEMALE, Grade.TWO),
new Student("小王", 6, Gender.MALE, Grade.ONE),
new Student("小李", 6, Gender.MALE, Grade.ONE),
new Student("小马", 14, Gender.FEMALE, Grade.FOUR),
new Student("小刘", 13, Gender.MALE, Grade.FOUR));
// 得到所有学生的年龄列表
// s -> s.getAge() --> Student::getAge , 不会多生成一个类似 lambda$0这样的函数
Set<Integer> ages = students.stream().map(Student::getAge)
.collect(Collectors.toCollection(TreeSet::new));
System.out.println("所有学生的年龄:" + ages);
// 统计汇总信息
IntSummaryStatistics agesSummaryStatistics = students.stream()
.collect(Collectors.summarizingInt(Student::getAge));
System.out.println("年龄汇总信息:" + agesSummaryStatistics);
// 分块
Map<Boolean, List<Student>> genders = students.stream().collect(
Collectors.partitioningBy(s -> s.getGender() == Gender.MALE));
// System.out.println("男女学生列表:" + genders);
MapUtils.verbosePrint(System.out, "男女学生列表", genders);
// 分组
Map<Grade, List<Student>> grades = students.stream()
.collect(Collectors.groupingBy(Student::getGrade));
MapUtils.verbosePrint(System.out, "学生班级列表", grades);
// 得到所有班级学生的个数
Map<Grade, Long> gradesCount = students.stream().collect(Collectors
.groupingBy(Student::getGrade, Collectors.counting()));
MapUtils.verbosePrint(System.out, "班级学生个数列表", gradesCount);
}
}
