WebFlux响应式编程基础之 3 Stream 流编程

流水线
外部迭代和内部迭代
惰性求值
惰性求值就是终止没有调用的情况下，中间操作不会执行

package 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;}}

短路操作

流的创建

package stream;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);}}

流的中间操作

package stream;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";// 把每个单词的长度调用出来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的子类, 所以要进行装箱 boxedStream.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);}}

package stream;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));// 收集到listList<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(""));// 带初始化值的reduceString 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());}}

并行流

package stream;import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.TimeUnit;
import java.util.stream.IntStream;public class StreamDemo5 {public static void main(String[] args) {// 调用parallel 产生一个并行流// IntStream.range(1, 100).parallel().peek(StreamDemo5::debug).count();// 现在要实现一个这样的效果: 先并行,再串行// 多次调用 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");// IntStream.range(1, 100).parallel().peek(StreamDemo5::debug).count();// 使用自己的线程池, 不使用默认线程池, 防止任务被阻塞// 线程名字 : ForkJoinPool-1ForkJoinPool 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();}}}public static void debug(int i) {System.out.println(Thread.currentThread().getName() + " debug " + i);try {TimeUnit.SECONDS.sleep(3);} catch (InterruptedException e) {e.printStackTrace();}}public static void debug2(int i) {System.err.println("debug2 " + i);try {TimeUnit.SECONDS.sleep(3);} catch (InterruptedException e) {e.printStackTrace();}}
}

收集器

package stream;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;}@Overridepublic 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);}}

Stream运行机制

package stream;import java.util.Random;
import java.util.concurrent.TimeUnit;
import java.util.stream.Stream;/*** 验证stream运行机制* * 1. 所有操作是链式调用, 一个元素只迭代一次 * 2. 每一个中间操作返回一个新的流. 流里面有一个属性sourceStage *     指向同一个 地方,就是Head * 3. Head->nextStage->nextStage->... -> null* 4. 有状态操作会把无状态操作阶段,单独处理* 5. 并行环境下, 有状态的中间操作不一定能并行操作.* * 6. parallel/ sequetial 这2个操作也是中间操作(也是返回stream)*      但是他们不创建流, 他们只修改 Head的并行标志* **/
public class RunStream {public static void main(String[] args) {Random random = new Random();// 随机产生数据Stream<Integer> stream = Stream.generate(() -> random.nextInt())// 产生500个 ( 无限流需要短路操作. ).limit(500)// 第1个无状态操作.peek(s -> print("peek: " + s))// 第2个无状态操作.filter(s -> {print("filter: " + s);return s > 1000000;})// 有状态操作.sorted((i1, i2) -> {print("排序: " + i1 + ", " + i2);return i1.compareTo(i2);})// 又一个无状态操作.peek(s -> {print("peek2: " + s);}).parallel();// 终止操作stream.count();}/*** 打印日志并sleep 5 毫秒* * @param s*/public static void print(String s) {// System.out.println(s);// 带线程名(测试并行情况)System.out.println(Thread.currentThread().getName() + " > " + s);try {TimeUnit.MILLISECONDS.sleep(5);} catch (InterruptedException e) {}}}

有状态操作会把无状态操作阶段截断,单独处理

无状态操作基本都是一个参数

有转态操作一般两个参数

中间操作返回流

终止操作短路操作

WebFlux响应式编程基础之 3 Stream 流编程相关推荐

16ava基础---Lambda，Stream流编程题
命名要求:1. 类名,接口名,枚举名,注解名使用大驼峰2. 变量名,方法名,包名均使用小驼峰3. 常量名全大写,多个单词下划线分割4. 名字要见名知意,如果不知道对应的英文,可以使用拼音代替.不可使用 ...
springboot 使用webflux响应式开发教程（二）
本篇是对springboot 使用webflux响应式开发教程(一)的进一步学习. 分三个部分: 数据库操作 webservice websocket 创建项目,artifactId = tradin ...
函数式编程（JAVA）——Stream流
函数式编程(JAVA)--Stream流概述 Java8的Stream使用的是函数式编程模式,如同它的名字一样,它可以被用来对集合或数组进行链状流式的操作.可以更方便的让我们对集合或数组操作. 下述 ...
计算机科学与技术学习路线编程基础四大件应用实践编程（含C++学习路线）
计算机科学与技术学习路线&编程基础四大件&应用实践编程(含C++学习路线) 基本介绍本人211科班出身,目前大学临近毕业,想给迷茫的同行者或者后来人一些建议和推荐,少走弯路.想想自己 ...
WebFlux响应式编程基础之 4 reactive stream 响应式流
reactive stream 响应式流 - 简而言之,就是多了一个沟通的渠道发布订阅者背压交流 Reactive Stream主要接口 java.util.concurrent.Flow 源码 ...
Spring Webflux 响应式编程 (二) - WebFlux编程实战
第一章 Reactive Stream 第1节 jdk9的响应式流就是reactive stream,也就是flow.其实和jdk8的stream没有一点关系.说白了就一个发布-订阅模式,一共只有4 ...
Spring WebFlux 响应式编程学习笔记(一)
各位Javaer们,大家都在用SpringMVC吧?当我们不亦乐乎的用着SpringMVC框架的时候,Spring5.x又悄(da)无(zhang)声(qi)息(gu)的推出了Spring WebFl ...
Spring：Webflux响应式编程
目录简介响应式编程 Reactor 信号操作符 SpringWebflux 执行流程和API 注解实现SpringWebflux 说明:基于atguigu学习笔记. 简介 Webflux是 Sp ...
WebFlux响应式编程
文章目录概要入门WebFlux WebClient 概要在SpringMvc框架下,http的请求是同步的,在某些场景下为了提供性能,可以采用异步的方式来进行优化.WebFlux便是提供了异步的 ...

WebFlux响应式编程基础之 3 Stream 流编程

WebFlux响应式编程基础之 3 Stream 流编程相关推荐

最新文章

热门文章

WebFlux响应式编程基础之 3 Stream 流 编程

WebFlux响应式编程基础之 3 Stream 流 编程相关推荐

最新文章

热门文章

WebFlux响应式编程基础之 3 Stream 流编程

WebFlux响应式编程基础之 3 Stream 流编程相关推荐