标签:cat sleep reduce 修改 就是 parallel lse amp 一个
概述:
高级的迭代器,数据在流水线中,从而进行高效处理
中间操作,即返回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 }
完
标签:cat sleep reduce 修改 就是 parallel lse amp 一个
原文地址:https://www.cnblogs.com/quyangyang/p/11735596.html