一、迭代
public static void main(String[] args) {
int[] nums = {1, 2, 3};
//外部迭代
int sum = 0;
for (int i: nums){
sum += i;
}
//内部迭代
int sum2 = IntStream.of(nums).sum();
System.out.println(sum == sum2);
}
- 中间操作：返回Stream
- 终止操作：返回最终结果
惰性求值：若没有执行终止操作，则中间操作不会执行。
流只能遍历一次。
二、流的创建
常见的Stream流创建.png
public class Test {
public static void main(String[] args) {
//从集合创建
List list = new ArrayList();
list.stream();
//从数组创建
Arrays.stream(new int[]{1, 2, 3});
//创建数字流
IntStream.of(1, 2, 3);
IntStream.rangeClosed(1, 10);
//使用Random创建无限流
new Random().ints().limit(10);
//自己创建
Random random = new Random();
Stream.generate(() -> random.nextInt(10)).limit(10);
try(Stream<String> lines = Files.lines(Paths.get(""))){
//由文件生成流
}catch (IOException e){
//显式捕获流中的异常
}
}
}
对象流和数值流的转化：
- 映射到数值流：mapToInt、mapToDouble、mapToLong返回一个数值流IntStream、DoubleStream、LongStream。
- 转换回对象流：使用boxed()方法，如intStream.boxed()转化为Stream<Integer>类型。
三、中间操作
中间操作.png
- map:映射，对值进行处理。
- flatMap:多个stream连成一个stream。
- filter:过滤，保留通过某项测试的对象。
- peak:用于debug,是个中间操作，类似foreach。
- limit:返回一个不超过给定长度的流。有状态-有界。
- skip:跳过前n个元素，若元素不足n个，返回一个空流。有状态-有界。
- sorted:排序。有状态-无界。
- distinct:返回一个元素各异的流。（根据流所生成的元素的hashCode和equal方法实现）。有状态-无界。
- peek:将中间变量的值输出到日志中。
tips：filter和map一起使用时，尽管它们是两个独立的操作，但它们会被合并到同一次遍历中，这种技术叫做循环合并。
有状态操作：需要知道先前的历史。
四、终止操作
终止操作.png
- collect:收集到集合。
- reduce:把流合成一个对象。有状态-有界。
- foreach:消费流中的每个元素，返回void。
- count:返回流中元素的个数。
- allMatch、anyMatch、noneMatch、findFirst、findAny
- min、max
public class Test {
public static void main(String[] args) {
String str = "hello world";
List<Character> list = Stream.of(str.split(" ")).flatMap(s -> s.chars().boxed()).map(c -> (char)c.intValue()).collect(Collectors.toList());
System.out.println(list);
//带初始值的reduce
Integer len = Stream.of(str.split(" "))
.map(s -> s.length())
.reduce(0, (s1, s2) -> s1+s2);
System.out.println(len);
}
}
五、收集器
将流元素归约和汇总为一个值
（1）Collectors.toList()、Collectors.toSet()、Collectors.toCollection(TreeSet::new)，将流转化为集合。
（2）counting()、minBy()、maxBy()、averagingXX()、summarizingXX()，对流进行统计个数、最小值、最大值、平均数、汇总等操作。
（3）joining()对流中的每个对象的toString()得到的字符串连接成一个字符串。
（4）reducing(初始值,转换函数,累积函数)，转换函数的类型为BiFunction<T,T,T>，reducing()方法是上述这些特殊情况的一般化。元素分组
（1）groupingBy(?)返回一个Map<T, List<T>>
（2）多级分组：groupingBy(?, groupingBy(?))
（3）可搭配其他收集器使用：groupingBy(?, counting())、groupingBy(?，mapping() )
（4）mapping(映射函数,收集成集合)。
（5）collectingAndThen(要转换的收集器,转换函数)，将收集器的结果转化为另一种类型。元素分区
partitioningBy(?) 返回一个Map<Boolean, List<T>>
可类比groupingBy()的用法自定义收集器
（1）Collector接口解析
/**
* 前四个方法都会返回一个被collect方法调用的函数
* 第五个方法提供了一个提示列表，告诉collect方法在执行归约操作时可以应用哪些优化
* @param <T>集合中要收集的项目的泛型
* @param <A>累加器的类型，累加器是在收集过程用于累积部分结果的对象
* @param <R>收集操作得到的对象
*/
interface Collector<T, A, R> {
/**
* 建立新的结果容器
* @return
*/
Supplier<A> supplier();
/**
* 将元素添加到结果容器
* @return
*/
BiConsumer<A, T> accumulator();
/**
* 对结果容器应用最终转换
* @return
*/
Function<A, R> finisher();
/**
* 合并两个结果容器
* @return
*/
BinaryOperator<A> combiner();
/**
*Characteristics是一个包含CONCURRENT、UNORDERED、IDENTITY_FINISH三个项目的枚举类
*CONCURRENT accumulator函数可以从多个线程同时调用，且收集器可以并行归约流。如果没有声明为UNORDERED，仅在无序数据源才可以并行归约
* UNORDERED归约结果不受流中项目遍历和累积顺序的影响
* IDENTITY_FINISH表明完成器方法返回的函数是一个恒等函数。这种情况下，累加器对象将会直接用作归约过程的最终结果
* @return
*/
Set<Characteristics> characteristics();
}
（2）自定义收集器：模拟toList()
public class TestDemo {
public static void main(String[] args) {
System.out.println(IntStream.rangeClosed(0, 100).boxed().collect(new ToListCollector<>()));
}
}
class ToListCollector<T> implements Collector<T, List<T>, List<T>> {
/**
* 创建一个空的容器
* @return
*/
@Override
public Supplier<List<T>> supplier() {
return ArrayList::new;
}
/**
* 加元素添加到容器中
* @return
*/
@Override
public BiConsumer<List<T>, T> accumulator() {
return List::add;
}
/**
* 恒等函数
* @return
*/
@Override
public Function<List<T>, List<T>> finisher() {
return Function.identity();
}
/**
* 将两个累加器合并并返回
* @return
*/
@Override
public BinaryOperator<List<T>> combiner() {
return (list1, list2) -> {
list1.addAll(list2);
return list1;
};
}
/**
* 为收集器添加IDENTITY_FINISH和CONCURRENT标志
* @return
*/
@Override
public Set<java.util.stream.Collector.Characteristics> characteristics() {
return Collections.unmodifiableSet(EnumSet.of(java.util.stream.Collector.Characteristics.IDENTITY_FINISH, java.util.stream.Collector.Characteristics.CONCURRENT));
}
}
（3）自定义收集器：筛选质数
import java.util.*;
import java.util.function.*;
import java.util.stream.Collector;
import java.util.stream.IntStream;
/**
* @description: ${todo}
* @author: wxz1997
* @date: 18-9-26 下午7:50
*/
public class TestCollector{
public static void main(String[] args) {
System.out.println(IntStream.rangeClosed(2, 100).boxed().collect(new PrimeNumbersCollector()));
}
}
class PrimeNumbersCollector implements Collector<Integer, Map<Boolean, List<Integer>>, Map<Boolean, List<Integer>>> {
//创建一个HashMap容器，含有两个空的List
@Override
public Supplier<Map<Boolean, List<Integer>>> supplier() {
return () -> new HashMap<Boolean, List<Integer>>(){{
put(true, new ArrayList<>());
put(false, new ArrayList<>());
}};
}
@Override
public BiConsumer<Map<Boolean, List<Integer>>, Integer> accumulator() {
return (Map<Boolean, List<Integer>> acc, Integer candidate) -> {
acc.get(isPrime(acc.get(true), candidate)).add(candidate);
};
}
private static boolean isPrime(List<Integer> primes, Integer candidate) {
int candidateRoot = (int)Math.sqrt((double)candidate);
return takeWhile(primes, i -> i <= candidateRoot)
.stream()
.noneMatch(p -> candidate % p == 0);
}
private static <A> List<A> takeWhile(List<A> list, Predicate<A> p){
int i=0;
for (A item: list){
if (!p.test(item)){
return list.subList(0, i);
}
i++;
}
return list;
}
@Override
public Function<Map<Boolean, List<Integer>>, Map<Boolean, List<Integer>>> finisher() {
return Function.identity();
}
@Override
public BinaryOperator<Map<Boolean, List<Integer>>> combiner() {
return (Map<Boolean, List<Integer>> map1, Map<Boolean, List<Integer>> map2) -> {
map1.get(true).addAll(map2.get(true));
map1.get(false).addAll(map2.get(false));
return map1;
};
}
@Override
public Set<java.util.stream.Collector.Characteristics> characteristics() {
return Collections.unmodifiableSet(EnumSet.of(java.util.stream.Collector.Characteristics.IDENTITY_FINISH));
}
}
六、并行流
- 并行标志parallel()，串行标志sequential()，若多次使用，以最后一个为准决定该流为串行还是并行。
- 并行流内部使用了默认的ForkJoinPool，它默认的线程数量为处理器的数量。
-
自动装箱和拆箱操作会大大降低性能，尽可能使用原始类型流IntStream、LongStream、DoubleStream等来避免这些操作。
根据可分解性判断其是否适合并行
流的数据源和可分解性.png - 工作窃取：每个线程都为分配给它的任务保存一个双向链式队列,每完成一个任务,就会从队列头上取出下一个任务开始执行。基于前面所述的原因,某个线程可能早早完成了分配给它的所有任务,也就是它的队列已经空了,而其他的线程还很忙。这时,这个线程并没有闲下来,而是随机选了一个别的线程,从队列的尾巴上“偷走”一个任务。这个过程一直继续下去,直到所有的任务都执行完毕,所有的队列都清空。这就是为什么要划成许多小任务而不是少数几个大任务,这有助于更好地在工作线程之间平衡负载。
