Stream

Posted 2021-03-25 quyangyang

概述：

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

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

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

收集器 分组

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

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

并行 fork/join

 

 1 public class Demo {
 2     
 3     public static void main(String[] args) {
 4         int[] nums = {1,2,3};
 5         //外部迭代
 6         int sum = 0;
 7         for(int i :nums) {
 8             sum += i;
 9         }
10         
11         System.out.println(sum);
12         
13         
14         
15         //stream内部迭代
16 
17         int sum2 = IntStream.of(nums).sum();
18         System.out.println(sum2);
19         
20     }
21 
22 }

 

 

流的创建

 1 public class CreatDemo {
 2     
 3     public static void main(String[] args) {
 4         
 5         List<String> list = new ArrayList<>();
 6         
 7         //从集合创建
 8         list.stream();
 9         list.parallelStream();
10         
11         
12         //从数组创建
13         Arrays.stream(new int[]{2,3,5});
14         
15         //创建数字流
16         IntStream.of(1,3,5);
17         IntStream.rangeClosed(1, 10);
18         
19         //使用random创建一个无限流
20         new Random().ints().limit(10);
21         Random random = new Random();
22         
23         
24         //自己生产流
25         Stream.generate(() ->random.nextInt()).limit(20);
26         
27     }
28 
29 }

 

 

流的中间操作

 1 public class StreamDemo3 {
 2 
 3     
 4     public static void main(String[] args) {
 5         String str = "aaa";
 6         Stream.of(str.split(" ")).map(s->s.length())
 7         .forEach(System.out::println);
 8 
 9         
10         
11         //flatMap A元素下面有B元素，最终得到所有A元素的里面的B元素的集合
12         //A元素就是单词，B元素就是一个个字符
13         // intStream/longStream 并不是Stream的子类，所以要进行装箱boxed()
14         Stream.of(str.split(" ")).flatMap(s-> s.chars().boxed())
15         .forEach(i-> System.out.println((char)i.intValue()));
16         
17         
18         //peek用于debug,是中间操作，foreach是终止操作
19         Stream.of(str.split(" ")).peek(System.out::print).forEach(System.out::println);
20         
21         
22         //limit使用用于无限流
23         new Random().ints().filter(i->i>100&&i<1000).limit(10).forEach(System.out::println);
24         
25         
26     }
27 }

 

 

流的终止操作

 1 public class EndDemo {
 2     public static void main(String[] args) {
 3     
 4         String str = "abcd ddd";
 5         
 6         //使用并行流
 7         str.chars().parallel().forEachOrdered(i -> System.out.println((char)i));
 8         
 9         //带初始化值的使用reduce拼接字符串
10         String reduce = Stream.of(str.split(" ")).reduce("",(s1,s2)-> s1+"/"+s2);
11         System.out.println(reduce);
12         
13         //使用reduce拼接字符串
14         Optional<String> optiona = Stream.of(str.split(" ")).reduce((s1,s2)-> s1+"/"+s2);
15         System.out.println(optiona.orElse(""));
16         
17         //计算单词的总长度
18         Integer len = Stream.of(str.split(" ")).map(s -> s.length()).reduce(0,(s1,s2) -> s1+s2);
19         System.out.println(len);
20         
21         
22         //使用findList短路操作,返回一个结果中断流
23         OptionalInt findFirst = new Random().ints().findFirst();
24         System.out.println(findFirst.getAsInt());
25     
26      
27     }
28 }

 

并行流

 1 public class ParallelDemo {
 2     
 3     public static void main(String[] args) {
 4         
 5         /**
 6          * 多次调用parallel/sequential以最后一次为准
 7          * 
 8          * 并行流使用线程池：ForkjoinPool.commonPool
 9          * 默认线程数是当前机器的cpu数
10          * 
11          */
12         
13         //调用parallel产生并行流
14     //    IntStream.range(1,100).parallel().peek(ParallelDemo::debug).count();
15         
16     //    IntStream.range(1,100).peek(ParallelDemo::debug).count();
17         
18         //使用自己的线程池,防止线程阻塞
19         ForkJoinPool pool = new ForkJoinPool(20);
20         pool.submit(()->IntStream.range(1, 100).parallel()
21                 .peek(ParallelDemo::debug).count()
22                 );
23         pool.shutdown();
24         synchronized (pool) {
25         
26             try {
27                 pool.wait();
28             } catch (InterruptedException e) {
29                 // TODO Auto-generated catch block
30                 e.printStackTrace();
31             }
32         }
33         
34         }
35     
36     public static void debug(int i) {
37         System.out.println("debug"+i);
38         try {
39             TimeUnit.SECONDS.sleep(3);
40         } catch (InterruptedException e) {
41             // TODO Auto-generated catch block
42             e.printStackTrace();
43         }
44         
45         
46         
47     }
48 
49 }

 

 

 

 

完