Java学习笔记-day06-响应式编程Reactor API大全（上）

Reactor 是一个基于响应式编程的库，主要用于构建异步和事件驱动的应用程序。Reactor 提供了丰富的 API，包括创建、转换、过滤、组合等操作符，用于处理异步数据流。以下是一些 Reactor 的主要 API 示例：

pom依赖

   <dependencyManagement><dependencies><dependency><groupId>io.projectreactor</groupId><artifactId>reactor-bom</artifactId><version>2023.0.0</version><type>pom</type><scope>import</scope></dependency></dependencies></dependencyManagement><dependencies><dependency><groupId>io.projectreactor</groupId><artifactId>reactor-core</artifactId></dependency><dependency><groupId>io.projectreactor</groupId><artifactId>reactor-test</artifactId><scope>test</scope></dependency><dependency><groupId>org.junit.jupiter</groupId><artifactId>junit-jupiter</artifactId><version>5.7.2</version><scope>test</scope></dependency></dependencies>

1. 创建 Mono 和 Flux

  • Mono: 用于表示包含零个或一个元素的异步序列。
  • Flux: 用于表示包含零个或多个元素的异步序列。
import reactor.core.publisher.Mono;
import reactor.core.publisher.Flux;public class ReactorCreateExample {public static void main(String[] args) {// 创建包含单个元素的 MonoMono<String> mono = Mono.just("Hello, Reactor!");// 创建包含多个元素的 FluxFlux<Integer> flux = Flux.fromArray(new Integer[]{1, 2, 3, 4, 5});mono.subscribe(System.out::println); // 输出: Hello, Reactor!flux.subscribe(System.out::println); // 输出: 1, 2, 3, 4, 5}
}

2. 转换操作符

使用转换操作符对数据流进行转换或处理。

import reactor.core.publisher.Flux;public class ReactorTransformExample {public static void main(String[] args) {Flux<Integer> source = Flux.range(1, 5);// 对每个元素进行平方操作Flux<Integer> squared = source.map(x -> x * x);squared.subscribe(System.out::println); // 输出: 1, 4, 9, 16, 25}
}

3. 过滤操作符

使用过滤操作符筛选数据流中的元素。

import reactor.core.publisher.Flux;public class ReactorFilterExample {public static void main(String[] args) {Flux<Integer> source = Flux.range(1, 5);// 筛选偶数Flux<Integer> evenNumbers = source.filter(x -> x % 2 == 0);evenNumbers.subscribe(System.out::println); // 输出: 2, 4}
}

4. 组合操作符

使用组合操作符组合多个数据流。

import reactor.core.publisher.Flux;public class ReactorCombineExample {public static void main(String[] args) {Flux<Integer> source1 = Flux.range(1, 3);Flux<Integer> source2 = Flux.range(4, 3);// 合并两个数据流Flux<Integer> merged = Flux.concat(source1, source2);merged.subscribe(System.out::println); // 输出: 1, 2, 3, 4, 5, 6}
}

这些只是 Reactor API 的一小部分示例。Reactor 提供了丰富的操作符和方法，用于处理复杂的异步数据流。开发人员可以根据具体需求选择适当的操作符进行组合，以构建出符合业务逻辑的异步处理链。

5. 错误处理

Reactor 提供了多种处理错误的方式，例如使用 onErrorResume, onErrorReturn, doOnError 等方法。

import reactor.core.publisher.Flux;public class ReactorErrorHandlingExample {public static void main(String[] args) {Flux<Integer> source = Flux.just(1, 2, 0, 4, 5);// 处理除零异常并提供默认值Flux<Integer> result = source.map(x -> 10 / x).onErrorResume(ex -> Flux.just(-1));result.subscribe(System.out::println); // 输出: 10, 5, -1}
}

6. 背压处理

Reactor 提供了背压处理的支持，允许生产者和消费者之间实现合理的数据流控制。使用 onBackpressureBuffer 或者其他背压操作符可以处理高速生产者和慢速消费者之间的数据流。

import reactor.core.publisher.Flux;public class ReactorBackpressureExample {public static void main(String[] args) {Flux<Integer> source = Flux.range(1, 1000);// 设置缓冲区大小Flux<Integer> buffered = source.onBackpressureBuffer(10);buffered.subscribe(data -> {// 模拟慢速消费者try {Thread.sleep(100);} catch (InterruptedException e) {e.printStackTrace();}System.out.println(data);},error -> System.err.println("Error: " + error),() -> System.out.println("Done"));}
}
  • TODO：未能实现没有背压和有背压的对比

7. 使用 Reactor WebFlux 处理 Web 请求

Reactor 还提供了 WebFlux 模块，用于处理响应式的 Web 请求。以下是一个简单的示例：

import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
import reactor.core.publisher.Mono;@RestController
public class WebFluxController {@GetMapping("/hello")public Mono<ResponseEntity<String>> hello() {return Mono.just(ResponseEntity.ok("Hello, Reactor WebFlux!"));}
}

8. Reactor 核心概念

Reactor 中有一些核心概念，了解这些概念有助于更好地使用 Reactor API。

  • Publisher（发布者）: 代表一个生产数据的源头，通常是 Mono 或 Flux。

  • Subscriber（订阅者）: 用于消费数据流的组件。通过 subscribe 方法订阅 Publisher。

  • Subscription（订阅）: 代表 Subscriber 和 Publisher 之间的连接。Subscriber 可以使用 Subscription 来请求数据，取消订阅等。

  • Processor（处理器）: 既是 Publisher 又是 Subscriber，用于在两者之间进行转换和处理。

public class ReactorCoreConceptsExample {public static void main(String[] args) {// 创建发布者Flux<Integer> source = Flux.range(1, 5);// 创建处理器，并进行数据处理UnicastProcessor<Integer> processor = UnicastProcessor.create();source.map(value -> value * 2)  // Example: doubling the values.subscribe(processor);// 创建订阅者CustomSubscriber<Integer> subscriber = new CustomSubscriber<>();// 订阅并处理数据processor.subscribe(subscriber);}// 自定义订阅者static class CustomSubscriber<T> extends BaseSubscriber<T> {@Overrideprotected void hookOnNext(T value) {System.out.println("Processed Value: " + value);}@Overrideprotected void hookOnError(Throwable throwable) {System.err.println("Error: " + throwable);}@Overrideprotected void hookOnComplete() {System.out.println("Done");}}
}
  • UnicastProcessor.create()已弃用，可以使用Sinks.many().unicast().onBackpressureBuffer()

9. Reactor 调度器

Reactor 提供了多种调度器，用于控制异步操作的执行线程。例如，Schedulers.boundedElastic() 创建了一个弹性线程池，可以根据需要动态调整线程数。

public class ReactorSchedulersExample {public static void main(String[] args) {Flux.range(1, 5).publishOn(Schedulers.boundedElastic())  // 在弹性线程池上发布.map(x -> x * x).subscribeOn(Schedulers.parallel())  // 在并行线程池上订阅.subscribe(System.out::println);}
}
  • 经测试，大概率只使用了一个线程

11. 组合多个 Mono 或 Flux

使用 zip 操作符可以组合多个 Mono 或 Flux，将它们的元素进行组合。

import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;public class ReactorZipExample {public static void main(String[] args) {Mono<String> mono1 = Mono.just("Hello");Mono<String> mono2 = Mono.just("Reactor");// 将两个 Mono 合并为一个 FluxFlux<String> result = Flux.zip(mono1, mono2).map(tuple -> tuple.getT1() + " " + tuple.getT2());result.subscribe(System.out::println); // 输出: Hello Reactor}
}

12. 超时操作

使用 timeout 操作符可以在指定的时间内等待数据流产生结果，如果超时，则触发错误。

public class ReactorTimeoutExample {public static void main(String[] args) throws InterruptedException {Flux<Integer> source = Flux.just(1, 2, 3).delayElements(Duration.ofSeconds(2)); // 模拟延迟// 在指定时间内等待数据流产生结果，否则触发超时source.timeout(Duration.ofSeconds(1)).subscribe(data -> System.out.println("Received: " + data),error -> System.err.println("Error: " + error),() -> System.out.println("Done"));//睡一会，等待任务执行完成Thread.sleep(3333);}
}

13. 并行操作

使用 parallel 操作符可以将一个数据流并行处理，提高处理速度。

public class ReactorParallelExample {public static void main(String[] args) throws InterruptedException {Flux.range(1, 10).parallel().runOn(Schedulers.parallel()).map(x -> x * x).sequential().subscribe(System.out::println);//睡一会，等待任务执行完成Thread.sleep(1111);}
}

14. 与 Java Stream 集成

Reactor 与 Java Stream 可以方便地进行集成。

import reactor.core.publisher.Flux;
import java.util.stream.Stream;public class ReactorJavaStreamIntegrationExample {public static void main(String[] args) {Flux<Integer> flux = Flux.fromStream(Stream.of(1, 2, 3, 4, 5));flux.subscribe(System.out::println); // 输出: 1, 2, 3, 4, 5}
}

15. 使用 Mono 和 Flux 进行条件操作

Reactor 提供了条件操作符，例如 switchIfEmpty 和 filter，用于根据条件处理数据流。

public class ReactorConditionalOperatorsExample {public static void main(String[] args) {Flux<Integer> empty = Flux.range(1, 0);Flux<Integer> source = Flux.range(1, 5);// 如果数据流为空，则切换到另一个数据流empty.switchIfEmpty(Flux.range(6, 3)).subscribe(System.out::println); // 输出: 6,7,8// 使用 filter 过滤元素source.filter(x -> x % 2 == 0).subscribe(System.out::println); // 输出: 2, 4}
}

16. 使用 Reactor StepVerifier 进行测试

代码需要写在test测试目录下！！！

Reactor 提供了 StepVerifier 类，用于测试异步操作的行为。

public class ReactorTestingExample {public static void main(String[] args) {Flux<Integer> flux = Flux.range(1, 5);// 使用 StepVerifier 验证数据流的行为StepVerifier.create(flux).expectNext(1, 1, 3, 4, 5)//正确顺序应该是12345.expectComplete().verify();}
}

17. 使用 Mono 和 Flux 进行重试

Reactor 提供了 retryWhen 方法，结合 Backoff 操作符，用于在发生错误时进行重试。

public class ReactorRetryExample {public static void main(String[] args) throws InterruptedException {Mono<Object> source = Mono.fromCallable(() -> {throw new RuntimeException("Simulated error");})//最大重试次数为3次，初始重试间隔为1秒，并且采用指数回退策略，直到达到最大的回退时间（这里是5秒）。.retryWhen(Retry.backoff(3, Duration.ofSeconds(1)).maxBackoff(Duration.ofSeconds(5)));source.subscribe(data -> System.out.println("Received: " + data),error -> System.err.println("Error: " + error.getMessage()));//得多睡会儿，让它跑完最大重试时间Thread.sleep(999999);}
}

19. 使用 Reactor Context 进行上下文传递

Reactor 提供了 Context 类，用于在操作链中传递上下文信息。这对于在异步操作中共享信息非常有用。

import reactor.core.publisher.Mono;
import reactor.util.context.Context;public class ReactorContextExample {public static void main(String[] args) {Mono<String> mono = Mono.deferContextual(contextView ->Mono.just("Hello, " + contextView.get("user")));String result = mono.contextWrite(Context.of("user", "John")).block();System.out.println(result); // 输出: Hello, John}
}

20. 使用 Reactor 的 doOn 方法进行副作用处理

doOn 系列方法允许在数据流的不同生命周期阶段执行副作用操作，如日志记录、统计等。

import reactor.core.publisher.Flux;public class ReactorDoOnExample {public static void main(String[] args) {Flux<Integer> source = Flux.range(1, 5);source.doOnNext(value -> System.out.println("Processing element: " + value)).doOnComplete(() -> System.out.println("Processing complete")).subscribe(System.out::println);}
}

21. 使用 Reactor 的 transform 方法进行操作链重用

transform 方法允许对操作链进行重用，将一系列操作组合为一个新的 Function。

import reactor.core.publisher.Flux;import java.util.function.Function;public class ReactorTransformExample {public static void main(String[] args) {Flux<Integer> source = Flux.range(1, 5);// 定义一个操作链Function<Flux<Integer>, Flux<Integer>> customTransform = flux ->flux.filter(x -> x % 2 == 0).map(x -> x * 2);// 使用 transform 应用自定义操作链source.transform(customTransform).subscribe(System.out::println); // 输出: 4, 8}
}

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.mzph.cn/news/612140.shtml

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

191. 位1的个数

191. 位1的个数

编写一个函数&#xff0c;输入是一个无符号整数&#xff08;以二进制串的形式&#xff09;&#xff0c;返回其二进制表达式中数字位数为 1 的个数&#xff08;也被称为汉明重量&#xff09;。 提示&#xff1a; 请注意&#xff0c;在某些语言&#xff08;如 Java&#xff09;中…
阅读更多...
LeetCode 2645. 构造有效字符串的最少插入数

LeetCode 2645. 构造有效字符串的最少插入数

一、题目 1、题目描述 LeetCode 给你一个字符串 word &#xff0c;你可以向其中任何位置插入 "a"、"b" 或 "c" 任意次&#xff0c;返回使 word 有效 需要插入的最少字母数。如果2645. 构造有效字符串的最少插入数 2、接口描述 ​ class Solut…
阅读更多...
SpringCloud微服务 【实用篇】| RabbitMQ快速入门、SpringAMQP

SpringCloud微服务 【实用篇】| RabbitMQ快速入门、SpringAMQP

目录 一&#xff1a;初始RabbitMQ 1. 同步和异步通讯 1.1 同步调用 1.2 异步调用 2. MQ常见框架 二&#xff1a;RabbitMQ快速入门 1. RabbitMQ概述和安装 2. 常见消息队列模型 3. 快速入门案例 三&#xff1a;SpringAMQP 1. Basic Queue 简单队列模型 2. Work Queu…
阅读更多...
Hive事务表转换为非事务表

Hive事务表转换为非事务表

环境&#xff1a;hive3.1.0 由于建表时默认会建为非事务表 CREATE TABLE bucket_text_table2(column1 string,column2 string,column3 int) CLUSTERED BY (column3) into 5 BUCKETS STORED AS TEXTFILE; 执行完成后&#xff0c;查看默认建表语句&#xff1a; ---------------…
阅读更多...
PHP 微信小程序获取 手机号码

PHP 微信小程序获取 手机号码

PHP代码 $param $_POST; $app_id ""; $app_secret "";$url_get https://api.weixin.qq.com/cgi-bin/token?grant_typeclient_credential&appid . $app_id . &secret . $app_secret;$tmptoken json_decode(curl($url_get), true);$token $tm…
阅读更多...
ORACLE RAC DG文件路径错乱解决办法

ORACLE RAC DG文件路径错乱解决办法

最近接手了一个客户的RAC-RAC dg环境的维护,登录上去之后发现dg延迟了8天,由于主库的空间非常紧张,归档日志早就删除了,所以准备使用rman基于scn点的备份恢复的方案恢复dg同步 在备份完成之后,使用新的控制文件进行数据恢复的时候报错datafile 43 not found: 检查了一下发现当…
阅读更多...
SpringBoot中使用单例模式+ScheduledExecutorService实现异步多线程任务(若依源码学习)

SpringBoot中使用单例模式+ScheduledExecutorService实现异步多线程任务(若依源码学习)

场景 若依前后端分离版手把手教你本地搭建环境并运行项目&#xff1a; 若依前后端分离版手把手教你本地搭建环境并运行项目_本地运行若依前后端分离-CSDN博客 设计模式-单例模式-饿汉式单例模式、懒汉式单例模式、静态内部类在Java中的使用示例&#xff1a; 设计模式-单例模…
阅读更多...
数据库 MySQL 索引的原理

数据库 MySQL 索引的原理

在数据库中&#xff0c;索引是一种重要的数据结构&#xff0c;它用于加快数据的检索速度和提高查询性能。在 MySQL 中&#xff0c;索引的实现基于 B树结构。 索引的基本思想是通过维护一个有序的数据结构&#xff0c;来快速定位和访问表中的数据。B树是一种自平衡的二叉搜索树…
阅读更多...
Python requirements.txt 详解

Python requirements.txt 详解

文章目录 1 概述1.1 作用1.2 注意 2 操作2.1 生成 requirements.txt2.2 安装 requirements.txt 3 示例3.1 新建 Django 项目3.2 找到 Scripts 目录&#xff0c;执行生成 requirements.txt 命令 1 概述 1.1 作用 作用&#xff1a;记录 当前项目下 所有 依赖包及其版本号&#…
阅读更多...
Polars使用指南（一）

Polars使用指南（一）

pandas是Python数据处理中非常经典的一个科学计算库&#xff0c;表形式的数据结构、丰富的API和灵活的编程语法使得pandas成为最常用的的数据分析工具。但是pandas也有一个最致命的缺陷&#xff0c;就是效率问题&#xff0c;尤其是不支持并行计算。pandas2在性能方面有了极大的…
阅读更多...
不知道题目是啥

不知道题目是啥

本题是学校的集训里的题&#xff0c;所有不知道题目名字是啥&#xff0c;直接看题目就好 解题思路&#xff1a;因为字符串只含有小写字母&#xff0c;所以可以创建两个数组分别来存s和t的每个字母出现次数&#xff0c;然后遍历数组&#xff0c;如果s字符串中的某个字母比t的小&…
阅读更多...
输电线路分布式故障诊断装置的四大特点介绍-深圳鼎信

输电线路分布式故障诊断装置的四大特点介绍-深圳鼎信

输电线路分布式故障诊断装置是一种利用行波测距、无线通信等技术手段实现电网故障定位的设备。这对于电网的故障处理和恢复具有重要意义&#xff0c;可以帮助运维人员提高故障处理的效率&#xff0c;缩短故障处理时间&#xff0c;减少停电时间&#xff0c;提高用户的供电可靠性…
阅读更多...
premiere简约大气3D动画logo片头Pr模板Mogrt免费下载

premiere简约大气3D动画logo片头Pr模板Mogrt免费下载

Premiere简约大气3D动画logo片头pr模板mogrt下载&#xff0c;无需插件&#xff0c;高清分辨率&#xff0c;易于自定义&#xff0c;包括教程&#xff0c;不包括音频和图像。免费下载&#xff1a;https://prmuban.com/37065.html
阅读更多...
Linux学习（1）：目录结构、编辑器和用户管理

Linux学习（1）：目录结构、编辑器和用户管理

Linux学习&#xff08;1&#xff09;&#xff1a;目录结构、编辑器和用户管理 1 Linux目录结构2 vi和vim编辑器2.1 快捷键练习 3 用户管理3.1 添加用户3.2 删除用户即主目录3.3 切换用户 4 用户组 1 Linux目录结构 在linux世界里&#xff0c;一切皆为文件。 linux目录结构&a…
阅读更多...
test fuzz-05-模糊测试 kelinci AFL-based fuzzing for Java

test fuzz-05-模糊测试 kelinci AFL-based fuzzing for Java

拓展阅读 开源 Auto generate mock data for java test.(便于 Java 测试自动生成对象信息) 开源 Junit performance rely on junit5 and jdk8.(java 性能测试框架。性能测试。压测。测试报告生成。) test fuzz-01-模糊测试&#xff08;Fuzz Testing&#xff09; test fuzz-…
阅读更多...
Gin CORS 跨域请求资源共享与中间件

Gin CORS 跨域请求资源共享与中间件

Gin CORS 跨域请求资源共享与中间件 文章目录 Gin CORS 跨域请求资源共享与中间件一、同源策略1.1 什么是浏览器的同源策略&#xff1f;1.2 同源策略判依据1.3 跨域问题三种解决方案 二、CORS:跨域资源共享简介(后端技术)三 CORS基本流程1.CORS请求分类2.基本流程 四、CORS两种…
阅读更多...
信息安全面试攻略

信息安全面试攻略

在神秘的代码江湖里&#xff0c;有一群行走在钢丝上的信息安全程序员。有人说它神秘&#xff0c;有人说它刺激&#xff0c;也有人觉得它充满了挑战&#xff0c;他们不是普通的码农&#xff0c;而是掌握了“隐匿术”的忍者编程者。今天&#xff0c;就让我们一起探讨如何在这场没…
阅读更多...
Java项目：02 基于ssm超市订单管理系统

Java项目：02 基于ssm超市订单管理系统

项目介绍 基于ssm超市订单管理系统 环境&#xff1a;jdk1.8&#xff0c;mysql5.7&#xff0c;tomcat8.5&#xff0c;maven3.6 软件&#xff1a;IDEA 功能&#xff1a;超市后台管理系统&#xff0c;有订单管理&#xff0c;供应商管理&#xff0c;用户管理&#xff0c;密码修改&…
阅读更多...
阿赵UE学习笔记——9、材质和材质实例

阿赵UE学习笔记——9、材质和材质实例

阿赵UE学习笔记目录 大家好&#xff0c;我是阿赵。   继续学习虚幻引擎&#xff0c;这次来了解一下UE里面关于材质的一些概念性的东西。 一、材质 材质这个概念&#xff0c;在所有三维软件里面都会有&#xff0c;比如3Dsmax里面的材质球&#xff0c;或者Unity里面的Material…
阅读更多...
解决docker run报错：Error response from daemon: No command specified.

解决docker run报错：Error response from daemon: No command specified.

将docker镜像export/import之后&#xff0c;对新的镜像执行docker run时报错&#xff1a; docker: Error response from daemon: No command specified. 解决方法&#xff1a; 方案1&#xff1a; 查看容器的command&#xff1a; docker ps --no-trunc 在docker run命令上增加…
阅读更多...
最新文章

Copyright @ 2022~2023