Project Reactor 响应式范式编程

// 创建一个Mono对象，包含一个字符串元素
Mono<String> mono = Mono.just("Hello World");

// 订阅这个Mono对象，并打印元素值
mono.subscribe(System.out::println);

// 创建一个Mono对象，包含一个整数元素
Mono<Integer> mono = Mono.just(1);

// 使用flux()方法把Mono转换成Flux
Flux<Integer> flux = mono.flux();

// 订阅这个Flux对象，并打印元素值
flux.subscribe(System.out::println); // 输出1

// 创建一个Mono对象，包含一个整数元素
Mono<Integer> mono = Mono.just(1);

// 创建一个Flux对象，包含两个整数元素
Flux<Integer> flux = Flux.just(2, 3);

// 使用concatWith()方法把Mono和Flux连接起来
Flux<Integer> result = mono.concatWith(flux);

// 订阅这个Flux对象，并打印元素值
result.subscribe(System.out::println); // 输出1, 2, 3

// 从一个固定的值创建一个Mono
Mono.just("Hello").subscribe(System.out::println); // 输出Hello

// 从一个Callable对象创建一个Mono
Callable<String> callable = () -> "World";
Mono.fromCallable(callable).subscribe(System.out::println); // 输出World

// 从一个Supplier对象创建一个Mono
Supplier<String> supplier = () -> "Supplier!";
Mono.fromSupplier(supplier).subscribe(System.out::println); // 输出Supplier!

// 对Mono发出的元素进行映射操作
Mono.just("Hello").map(s -> s + " World").subscribe(System.out::println); // 输出Hello World

// 对Mono发出的元素进行扁平化操作
Mono.just("Hello")
    .flatMap(s -> Mono.just(s + " World"))
    .subscribe(System.out::println); // 输出Hello World

// 对Mono发出的元素进行过滤操作
Mono.just(1).filter(i -> i > 0).subscribe(System.out::println); // 输出1

// 将多个Mono合并为一个Mono
Mono.zip(Mono.just("Hello"), Mono.just("World"))
    .subscribe(tuple -> System.out.println(tuple.getT1() + " " + tuple.getT2())); // 输出Hello World

// 将多个Mono合并为一个Flux
Mono.just("Hello")
    .mergeWith(Mono.just("World"))
    .subscribe(System.out::println); // 输出Hello World

// 在这个Mono完成后，继续处理另一个发布者
Mono.just("Hello").then(Mono.just("World")).subscribe(System.out::println); // 输出World

// 在这个Mono发出元素时，执行一个副作用操作
Mono.just("Hello")
    .doOnNext(s -> System.out.println("Before: " + s))
    .map(s -> s + " World")
    .doOnNext(s -> System.out.println("After: " + s))
    .subscribe(); 
// 输出
// Before: Hello 
// After: Hello World

// 创建一个Flux对象，包含三个整数元素
Flux<Integer> flux = Flux.just(1, 2, 3);

// 订阅这个Flux对象，并打印元素值
flux.subscribe(System.out::println);

// 创建一个Flux对象，包含三个整数元素
Flux<Integer> flux = Flux.just(1, 2, 3);

// 使用next()方法把Flux转换成Mono
Mono<Integer> mono = flux.next();

// 订阅这个Mono对象，并打印元素值
mono.subscribe(System.out::println); // 输出1

// 创建一个Flux对象，包含三个整数元素
Flux<Integer> flux = Flux.just(1, 2, 3);

// 使用collectList()方法把Flux转换成Mono
Mono<List<Integer>> mono = flux.collectList();

// 订阅这个Mono对象，并打印元素值
mono.subscribe(System.out::println); // 输出[1, 2, 3]

// 从多个固定的值创建一个Flux
Flux.just("Hello", "World").subscribe(System.out::println); // 输出Hello World

// 从一个数组对象创建一个Flux
String[] array = {"Hello", "World"};
Flux.fromArray(array).subscribe(System.out::println); // 输出Hello World

// 从一个Iterable对象创建一个Flux
List<String> list = Arrays.asList("Hello", "World");
Flux.fromIterable(list).subscribe(System.out::println); // 输出Hello World

// 从一个Stream对象创建一个Flux
Stream<String> stream = Stream.of("Hello", "World");
Flux.fromStream(stream).subscribe(System.out::println); // 输出Hello World

// 创建一个包含指定范围内整数的Flux
Flux.range(1, 5).subscribe(System.out::println); // 输出1 2 3 4 5

// 创建一个按照指定时间间隔从0整数递增的Flux
Duration duration = Duration.ofSeconds(1);
Flux<Long> interval = Flux.interval(duration);
interval.subscribe(System.out::println);
// 使用blockLast阻塞主线程，防止程序立即退出
interval.blockLast();
// 输出结果每秒打印一次
// 0
// 1
// 2
// 3
// 4
// ...

// 对Flux发出的每个元素进行映射操作
Flux.just("Hello", "World").map(s -> s + "!")
    .subscribe(System.out::println); // 输出Hello! World!

// 对Flux发出的每个元素进行扁平化操作
Flux.just("Hello", "World")
    .flatMap(s -> Flux.just(s + "!"))
    .subscribe(System.out::println); //输出Hello! World!

// 对Flux发出的每个元素进行过滤操作
Flux.range(1, 5).filter(i -> i % 2 == 0).subscribe(System.out::println); 
// 输出2 4

// 将多个Flux合并为一个Flux
Flux.zip(Flux.just("Hello"), Flux.just("World"))
    .subscribe(tuple -> System.out.println(tuple.getT1() + " " + tuple.getT2())); 
// 输出Hello World

// 将多个Flux合并为一个Flux
Flux.just("Hello").mergeWith(Flux.just("World")).subscribe(System.out::println); 
//  输出Hello World

// 将多个Flux合并为一个Flux
Flux.just("Hello").concatWith(Flux.just("World")).subscribe(System.out::println); 
// 输出Hello World

// 将所有元素收集到一个List中
Flux.just("Hello", "World").collectList().subscribe(list -> System.out.println(list)); // 输出[Hello, World]

  Flux<String> flux1 = Flux.just("A", "B", "C");
  Flux<Integer> flux2 = Flux.just(1, 2, 3, 4);
  Flux<Tuple2<String, Integer>> flux3 = Flux.zip(flux1, flux2);
  flux3.subscribe(tuple -> System.out.println(tuple.getT1() + " " + tuple.getT2())); 
  // 输出A 1 B 2 C 3
  // 4不会输出，因为最短的Flux是flux1，长度是3

  Duration duration1 = Duration.ofMillis(100);
  Duration duration2 = Duration.ofMillis(200);
  Flux<String> flux1 = Flux.interval(duration1).map(i -> "A" + i);
  Flux<String> flux2 = Flux.interval(duration2).map(i -> "B" + i);
  Flux<String> flux = flux1.mergeWith(flux2);
  flux.subscribe(System.out::println); 
  // 输出A0 B0 A1 B1 A2 A3 B2 A4 B3 A5 ...

  Duration duration1 = Duration.ofMillis(100);
  Duration duration2 = Duration.ofMillis(200);
  // 每100ms递增1，打印5次结束
  Flux<String> flux1 = Flux.interval(duration1).map(i -> "A" + i).take(5);
  Flux<String> flux2 = Flux.interval(duration2).map(i -> "B" + i).take(5);
  Flux<String> flux3 = flux1.concatWith(flux2);
  flux3.subscribe(System.out::println);
  // 避免程序立即退出
  flux3.blockLast();
  // 输出 
  // A0
  // A1
  // A2
  // A3
  // A4
  // B0
  // B1
  // B2
  // B3
  // B4