https://cloud.tencent.com/developer/article/1073888 Java 平台反应式编程（Reactive Programming）入门

流式编程

个人理解，流式编程就是 lambda 函数式编程的使用。其中函数式编程，supplier、consumer、function、predicate，派生出的 BiFunction，unaryOperator 等；
stream 流，流的产生，中间计算，结束。
函数式接口 @FunctionalInterface
流式编程、反应式编程 - 图1

  1. package com.demo;
  3. import java.util.Arrays;
  4. import java.util.function.BiConsumer;
  5. import java.util.function.BiFunction;
  6. import java.util.function.BiPredicate;
  7. import java.util.function.BinaryOperator;
  8. import java.util.function.Consumer;
  9. import java.util.function.Function;
  10. import java.util.function.Predicate;
  11. import java.util.function.Supplier;
  12. import java.util.stream.Stream;
  14. public class TestStreamCode {
  16.     private static void testFunction() {
  17.         Function<String, String> function = i -> "testFunction: 你输入的是："+i;
  18.         System.out.println(function.apply("hello world"));
  19.     }
  21.     private static void testBiFunction() {
  22.         BiFunction<Integer, String, String> function = (i, s) -> String.format("testBiFunction: %s 今年 %d", s, i);
  23.         System.out.println(function.apply(30, "xiaohui"));
  24.     }
  26.     private static void testSuppier() {
  27.         Supplier<String> supplier = () -> new String("testSuppier: xiaohui");
  28.     }
  30.     private static void testConsumer() {
  31.         Consumer<String> consumer = (s) -> System.out.println("testConsumer: "+s);
  32.         consumer.accept("xiaohui");
  33.     }
  35.     private static void testBiConsumer() {
  36.         BiConsumer<Integer, String> consumer = (i, s) -> System.out.println(String.format("testBiConsumer: %s %d", s,
  37.                 i));
  38.         consumer.accept(100, "xiaohui");
  39.     }
  41.     private static void testPredicate() {
  42.         Predicate<String> predicate = s -> s.isBlank();
  43.         System.out.println("testPredicate: "+ predicate.test("xiaohui"));
  44.     }
  46.     private static void testBiPredicate() {
  47.         BiPredicate<Integer, String> predicate = (i, s) -> i > 0 && s.isBlank();
  48.         System.out.println(String.format("testBiPredicate: %s, %d", "xiaohui", 100));
  49.     }
  51.     private static void testBinaryOperator() {
  52.         // public interface BinaryOperator<T> extends BiFunction<T,T,T>
  53.         BinaryOperator<String> operator = (s1,s2) -> s1 + s2;
  54.         System.out.println("testBinaryOperator: " + operator.apply("hello", "world"));
  55.     }
  57.     private static void testStream() {
  58.         Arrays.stream(new String[]{"hello", "hi", "嘿嘿"}).forEach(System.out::println);
  59.         Arrays.asList("哈哈", "呵呵").stream().forEach(System.out::println);
  60.         Stream.iterate(1, i->i+1).limit(10).forEach(System.out::print);
  61.         Stream.of("xiaohui".split("")).forEach(System.out::println);
  62.         Stream.generate(()->"xiaohui").limit(9).forEach(System.out::println);
  63.     }
  66.     public static void main(String[] args) {
  67.         testBiConsumer();
  68.         testBiFunction();
  69.         testBinaryOperator();
  70.         testBiPredicate();
  71.         testConsumer();
  72.         testFunction();
  73.         testPredicate();
  74.         testSuppier();
  75.         ////////////////////////////
  76.         testStream();
  78.     }
  79. }

反应式编程

project reactor
flux、mono
流式编程、反应式编程 - 图2
流的建立、操作、结束
mono、flux
流式编程、反应式编程 - 图3

基于流式的restful web请求（基于注解，基于routerFunction）

  1. package com.demo;
  3. import static org.springframework.web.reactive.function.server.RequestPredicates.DELETE;
  4. import static org.springframework.web.reactive.function.server.RequestPredicates.GET;
  5. import static org.springframework.web.reactive.function.server.RequestPredicates.accept;
  6. import static org.springframework.web.reactive.function.server.RouterFunctions.route;
  8. import org.springframework.boot.SpringApplication;
  9. import org.springframework.boot.autoconfigure.SpringBootApplication;
  10. import org.springframework.context.annotation.Bean;
  11. import org.springframework.http.MediaType;
  12. import org.springframework.stereotype.Component;
  13. import org.springframework.web.bind.annotation.GetMapping;
  14. import org.springframework.web.bind.annotation.RequestMapping;
  15. import org.springframework.web.bind.annotation.RestController;
  16. import org.springframework.web.reactive.function.server.RouterFunction;
  17. import org.springframework.web.reactive.function.server.ServerRequest;
  18. import org.springframework.web.reactive.function.server.ServerResponse;
  19. import reactor.core.publisher.Mono;
  21. @SpringBootApplication
  22. public class TestWebFlux {
  24.     public static void main(String[] args) {
  25.         SpringApplication.run(TestWebFlux.class, args);
  26.     }
  28.     @Bean
  29.     public RouterFunction<ServerResponse> monoRouterFunction(UserHandler userHandler) {
  30.         return route(GET("/{user}").and(accept(MediaType.APPLICATION_JSON)), userHandler::getUser)
  31.                 .andRoute(GET("/{user}/customers").and(accept(MediaType.APPLICATION_JSON)),
  32.                         userHandler::getUserCustomers)
  33.                 .andRoute(DELETE("/{user}").and(accept(MediaType.APPLICATION_JSON)), userHandler::deleteUser);
  34.     }
  35.     @Bean
  36.     public RouterFunction<ServerResponse> functionResponse() {
  37.         return route().GET("/function/greeting", request -> ServerResponse.ok().bodyValue("function greeting")).build();
  38.     }
  40. }
  42. @RestController
  43. @RequestMapping("/annotated")
  44. class TestAnnotated {
  46.     @GetMapping("/greeting")
  47.     public Mono<String> greeting() {
  48.         return Mono.just("hello greeting by annotated.");
  49.     }
  53. }
  54. @Component
  55. class UserHandler {
  57.     public Mono<ServerResponse> getUser(ServerRequest request) {
  58.         // ...
  59. //        Flux.just(1).publishOn(Schedulers.immediate())
  60.         return ServerResponse.ok().bodyValue("getUser");
  61.     }
  63.     public Mono<ServerResponse> getUserCustomers(ServerRequest request) {
  64.         // ...
  65.         return ServerResponse.ok().bodyValue("getUserCustomers");
  66.     }
  68.     public Mono<ServerResponse> deleteUser(ServerRequest request) {
  69.         // ...
  70.         return ServerResponse.ok().bodyValue("deleteUser");
  71.     }
  72. }
  73. /*
  74.     <dependency>
  75.       <groupId>org.springframework.boot</groupId>
  76.       <artifactId>spring-boot-starter-webflux</artifactId>
  77.     </dependency>
  78. */
  79. /*
  80. 2021-04-15 17:31:03.250  INFO 11924 --- [           main] com.demo.TestWebFlux                     : Starting TestWebFlux using Java 15.0.2 on 220200700182 with PID 11924 (D:\work\task\my-activiti-demo\scala-second\target\classes started by DELL in D:\work\task\my-activiti-demo\scala-second)
  81. 2021-04-15 17:31:03.253  INFO 11924 --- [           main] com.demo.TestWebFlux                     : No active profile set, falling back to default profiles: default
  82. 2021-04-15 17:31:04.200  INFO 11924 --- [           main] o.s.b.web.embedded.netty.NettyWebServer  : Netty started on port(s): 8080
  83. 2021-04-15 17:31:04.219  INFO 11924 --- [           main] com.demo.TestWebFlux                     : Started TestWebFlux in 1.206 seconds (JVM running for 1.684)
  84. */