Stream

概述：

高级的迭代器，数据在流水线中，从而进行高效处理

中间操作，即返回String 的操作，分为有状态的操作和无状态的操作，函数接口有一个参数的是
无状态，有两个参数的是有状态

中止操作 短路（无限流）
　　parallel/sequential 不创建流，只修改流head标志

收集器 分组

运行机制 ：链式调用 ，一个元素只迭代一次，

　　　　head - > nextStage -> nextStage ... -> null

并行 fork/join

public class Demo {
    
    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内部迭代

        int sum2 = IntStream.of(nums).sum();
        System.out.println(sum2);
        
    }

}

流的创建

public class CreatDemo {
    
    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,3,5);
        IntStream.rangeClosed(1, 10);
        
        //使用random创建一个无限流
        new Random().ints().limit(10);
        Random random = new Random();
        
        
        //自己生产流
        Stream.generate(() ->random.nextInt()).limit(20);
        
    }

}

流的中间操作

public class StreamDemo3 {

    
    public static void main(String[] args) {
        String str = "aaa";
        Stream.of(str.split(" ")).map(s->s.length())
        .forEach(System.out::println);

        
        
        //flatMap A元素下面有B元素，最终得到所有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是终止操作
        Stream.of(str.split(" ")).peek(System.out::print).forEach(System.out::println);
        
        
        //limit使用用于无限流
        new Random().ints().filter(i->i>100&&i<1000).limit(10).forEach(System.out::println);
        
        
    }
}

流的终止操作

public class EndDemo {
    public static void main(String[] args) {
    
        String str = "abcd ddd";
        
        //使用并行流
        str.chars().parallel().forEachOrdered(i -> System.out.println((char)i));
        
        //带初始化值的使用reduce拼接字符串
        String reduce = Stream.of(str.split(" ")).reduce("",(s1,s2)-> s1+"/"+s2);
        System.out.println(reduce);
        
        //使用reduce拼接字符串
        Optional<String> optiona = Stream.of(str.split(" ")).reduce((s1,s2)-> s1+"/"+s2);
        System.out.println(optiona.orElse(""));
        
        //计算单词的总长度
        Integer len = Stream.of(str.split(" ")).map(s -> s.length()).reduce(0,(s1,s2) -> s1+s2);
        System.out.println(len);
        
        
        //使用findList短路操作,返回一个结果中断流
        OptionalInt findFirst = new Random().ints().findFirst();
        System.out.println(findFirst.getAsInt());
    
     
    }
}

并行流

public class ParallelDemo {
    
    public static void main(String[] args) {
        
        /**
         * 多次调用parallel/sequential以最后一次为准
         * 
         * 并行流使用线程池：ForkjoinPool.commonPool
         * 默认线程数是当前机器的cpu数
         * 
         */
        
        //调用parallel产生并行流
    //    IntStream.range(1,100).parallel().peek(ParallelDemo::debug).count();
        
    //    IntStream.range(1,100).peek(ParallelDemo::debug).count();
        
        //使用自己的线程池,防止线程阻塞
        ForkJoinPool pool = new ForkJoinPool(20);
        pool.submit(()->IntStream.range(1, 100).parallel()
                .peek(ParallelDemo::debug).count()
                );
        pool.shutdown();
        synchronized (pool) {
        
            try {
                pool.wait();
            } catch (InterruptedException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
        }
        
        }
    
    public static void debug(int i) {
        System.out.println("debug"+i);
        try {
            TimeUnit.SECONDS.sleep(3);
        } catch (InterruptedException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
        
        
        
    }

}

完