Reactive 编程-Project Reactor
Reactive 编程与 Project Reactor
Reactive 编程是一种编程范式,主要用于处理异步数据流。它旨在通过声明式的编程方式处理事件驱动的非阻塞任务,特别适合于构建响应式、可扩展、高并发的应用。随着互联网应用规模的扩大和响应速度的提升需求,Reactive 编程变得越来越流行。
在 Java 生态中,Project Reactor 是支持 Reactive 编程的核心库之一,基于 Reactive Streams 规范,并被 Spring 5 中的 WebFlux 采用为核心反应式框架。它提供了强大的 API 用于处理异步数据流,同时保持良好的性能和响应性。
一、Reactive 编程的核心概念
在传统的编程模型中,异步处理通常会涉及回调函数、线程池和复杂的同步控制,这样的代码不仅难以维护,还容易出现阻塞和性能瓶颈。Reactive 编程的出现解决了这些问题,它通过响应式的数据流,允许程序按需、非阻塞地处理数据和事件。
Reactive 编程的几个重要特点:
-
异步与非阻塞:程序可以异步处理任务,而不会因为等待结果而阻塞线程。这种非阻塞机制非常适合处理 I/O 密集型任务。
-
事件驱动:通过数据流来处理一系列事件。当有新数据产生时,系统会自动响应和处理这些数据。
-
流式处理:处理数据流中的数据项,类似于 Java 8 中的
Stream
API,但不同的是,Reactive 流可以处理动态、无限的数据流。 -
背压(Backpressure):Reactive Streams 规范提供了背压机制,允许消费者根据自身的能力,按需请求数据,避免生产者产生过多的数据而导致内存溢出。
二、Project Reactor 概述
Project Reactor 是一个支持 Reactive Streams 规范的响应式编程库。它是构建在 Reactive Streams 基础上的高性能框架,提供了几种关键的异步数据流处理类型:Mono
和 Flux
。
- Mono:表示 0 或 1 个元素的异步数据流。适用于返回单个结果的场景,例如 HTTP 请求的响应。
- Flux:表示 0 到 N 个元素的异步数据流,适用于处理多个结果或无限流的场景。
Project Reactor 中的 Mono
和 Flux
提供了强大的操作符(类似于 Java 的 Stream
API 中的操作符),用于组合、转换、过滤和操作异步数据流。
三、Mono 和 Flux 的基本用法
1. Mono 的使用
Mono
代表一个包含最多一个元素的异步流。它可以用于表示单个异步任务的结果,如数据库查询、HTTP 请求等。
import reactor.core.publisher.Mono;public class MonoExample {public static void main(String[] args) {// 创建一个包含数据的 MonoMono<String> mono = Mono.just("Hello, Reactive World!");// 订阅并消费 Monomono.subscribe(System.out::println);}
}
在上述示例中,Mono.just("Hello, Reactive World!")
创建了一个包含单个元素的 Mono。当调用 subscribe()
方法时,Mono 开始执行并将数据输出到控制台。
2. Flux 的使用
Flux
是表示 0 到 N 个元素的异步流,适合处理多个数据项或无限的数据流。
import reactor.core.publisher.Flux;public class FluxExample {public static void main(String[] args) {// 创建一个包含多个元素的 FluxFlux<String> flux = Flux.just("Spring", "Reactor", "WebFlux");// 订阅并消费 Fluxflux.subscribe(System.out::println);}
}
在这个例子中,Flux.just()
创建了一个包含多个元素的 Flux,当 subscribe()
被调用时,Flux 会依次发射每个元素并输出它们。
四、背压(Backpressure)机制
背压是 Reactive Streams 中的一个关键概念,旨在解决生产者与消费者之间速率不匹配的问题。当生产者产生数据的速度快于消费者处理的速度时,背压允许消费者按自己的能力请求数据,避免数据积压导致内存问题。
在 Project Reactor 中,背压由订阅者(即消费者)通过 request(n)
方法来控制。例如,Flux 通过 onBackpressureBuffer()
或 onBackpressureDrop()
来处理背压。
import reactor.core.publisher.Flux;public class BackpressureExample {public static void main(String[] args) {Flux.range(1, 1000).onBackpressureBuffer(10) // 当消费者处理不过来时,使用缓冲区.subscribe(data -> {System.out.println("Processing " + data);try {Thread.sleep(100); // 模拟处理时间} catch (InterruptedException e) {e.printStackTrace();}},error -> System.err.println("Error: " + error),() -> System.out.println("Complete"));}
}
在该示例中,Flux 将发射 1000 个元素,但消费者处理的速度较慢,因此我们使用 onBackpressureBuffer(10)
将多余的数据存放到缓冲区,以便消费者可以按需处理数据。
五、常用的操作符
Project Reactor 提供了许多操作符,用于处理和转换数据流。以下是一些常用的操作符示例:
1. map()
map()
操作符用于将每个元素进行转换,类似于 Java 8 的 Stream.map()
。
Flux<Integer> numbers = Flux.just(1, 2, 3, 4);
numbers.map(n -> n * 2).subscribe(System.out::println); // 输出 2, 4, 6, 8
2. flatMap()
flatMap()
可以用于异步转换,并返回新的 Mono
或 Flux
。
Flux<String> names = Flux.just("Tom", "Jerry", "Spike");
names.flatMap(name -> Flux.just(name.toUpperCase())).subscribe(System.out::println); // 输出 TOM, JERRY, SPIKE
3. filter()
filter()
用于过滤掉不满足条件的元素。
Flux<Integer> numbers = Flux.range(1, 10);
numbers.filter(n -> n % 2 == 0).subscribe(System.out::println); // 输出 2, 4, 6, 8, 10
4. reduce()
reduce()
用于将多个值组合成一个值,类似于 Stream.reduce()
。
Flux<Integer> numbers = Flux.just(1, 2, 3, 4, 5);
numbers.reduce((a, b) -> a + b).subscribe(System.out::println); // 输出 15
5. zip()
zip()
操作符用于合并多个流,并将其元素“打包”成一个对象或元组。
Flux<String> names = Flux.just("Tom", "Jerry");
Flux<Integer> ages = Flux.just(5, 6);
Flux.zip(names, ages).subscribe(tuple -> System.out.println("Name: " + tuple.getT1() + ", Age: " + tuple.getT2()));
六、Project Reactor 的错误处理
Reactive 编程中的错误处理非常重要,因为异步流程中的异常不会像同步代码那样直接抛出。在 Project Reactor 中,有几种方式处理错误:
1. onErrorReturn()
在发生错误时返回默认值。
Flux<Integer> flux = Flux.just(1, 2, 0, 4).map(i -> 10 / i) // 会产生除以 0 的异常.onErrorReturn(-1);
flux.subscribe(System.out::println); // 输出 10, 5, -1
2. onErrorResume()
onErrorResume()
允许你在发生错误时切换到一个新的 Flux
或 Mono
。
Flux<Integer> flux = Flux.just(1, 2, 0, 4).map(i -> 10 / i).onErrorResume(e -> Flux.just(-1, -2));
flux.subscribe(System.out::println); // 输出 10, 5, -1, -2
七、Reactive 编程的优势与适用场景
Reactive 编程在处理高并发、I/O 密集型任务时表现尤为出色,特别适用于以下场景:
-
微服务架构:在微服务中,服务之间的通信常常需要通过非阻塞 I/O,Reactive 编程能够显著提升系统的吞吐量。
-
**高并发的
Web 应用**:Reactive 编程可以处理大量同时进行的用户请求,而不会因为线程阻塞而限制系统性能。
- 事件驱动系统:如物联网(IoT)系统,Reactive 编程可以很好地处理流式数据和异步事件。
八、总结
Reactive 编程为构建高性能、响应式系统提供了一种全新的方式,而 Project Reactor 则是 Java 生态中实现 Reactive Streams 规范的强大工具。通过 Mono
和 Flux
的流式 API,开发者可以简洁高效地处理异步任务,并借助背压机制避免过度生产数据导致的资源问题。
Project Reactor 的丰富操作符、错误处理机制和与 Spring WebFlux 的无缝集成,使其成为开发现代高并发应用的得力助手。掌握 Reactive 编程及其在 Project Reactor 中的实现,能够显著提高应用的扩展性、性能和响应性。
相关文章:
Reactive 编程-Project Reactor
Reactive 编程与 Project Reactor Reactive 编程是一种编程范式,主要用于处理异步数据流。它旨在通过声明式的编程方式处理事件驱动的非阻塞任务,特别适合于构建响应式、可扩展、高并发的应用。随着互联网应用规模的扩大和响应速度的提升需求࿰…...

splice用法
天行健,君子以自强不息;地势坤,君子以厚德载物。 每个人都有惰性,但不断学习是好好生活的根本,共勉! 文章均为学习整理笔记,分享记录为主,如有错误请指正,共同学习进步。…...

Redis - 缓存
文章目录 目录 文章目录 1. 什么是缓存? 2. 使用Redis作为缓存 2.1 关系型数据库的缺点 3. 缓存的更新策略 3.1 定期生成 3.2 实时生成 缓存淘汰策略 4. 缓存预热, 缓存穿透, 缓存雪崩 和 缓存击穿 缓存预热 缓存穿透 缓存雪崩 缓存击穿 总结 1. 什么…...

基于SpringBoot+Vue的养老院管理系统
作者:计算机学姐 开发技术:SpringBoot、SSM、Vue、MySQL、JSP、ElementUI、Python、小程序等,“文末源码”。 专栏推荐:前后端分离项目源码、SpringBoot项目源码、SSM项目源码 系统展示 【2025最新】基于JavaSpringBootVueMySQL的…...
多线程爬虫接入代理IP:高效数据抓取的秘诀
在现代网络环境中,爬虫已经成为获取信息的利器。然而,随着网站反爬措施的不断升级,单线程爬虫往往无法满足需求。多线程爬虫与代理IP的结合,不仅能提高效率,还能有效规避IP封禁问题。本文将详细探讨多线程爬虫接入代理…...
[网络][CISCO]Cisco-PIX配置详解
Cisco PIX防火墙配置指南 任何企业安全策略的一个主要部分都是实现和维护防火墙,因此防火墙在网络安全的实现当中扮演着重要的角色。防火墙通常位于企业网络的边缘,使内部网络与Internet之间或与其他外部网络互相隔离,并限制网络互访&#x…...

拒绝千篇一律,AI帮你定制独一无二的个人写真
每个女人都渴望展现最美的自己,你是否厌倦了拍出千篇一律的照片?今天,我要告诉你一个秘密,用简单三步,即可打造属于你的独一无二个人写真!文生图、蒙版换脸、图生图,三步化身超级模特࿰…...
在云服务器上安装 RabbitMQ:从零到一的最佳实践
🛠 1. RabbitMQ 简介 RabbitMQ 是一个开源的消息代理中间件,广泛应用于高并发、异步任务队列的场景中。在分布式系统架构中,RabbitMQ 可以充当消息的中转站,帮助不同服务之间进行高效的消息通信。 在这篇文章中,我们…...

【nginx】搭配okhttp 配置反向代理
nginx的默认是一个反向代理。 nginx会默认把输入的请求,转向其他的服务器执行。 这些转向的服务器与客户端发起的服务器不是同一个。 客户端只认识nginx,不知道ngiix转向何方。 正向代理修改okhttp的proxy,实际上很多代理都是正向的。 反向代理修改请求路径到nginx。 感觉还…...

Android V 广播注册和配置注意事项问题
现象 在Android V平台上,应用注册非Protected广播时,如果没有加导出flag会抛出异常导致进程crash。 E/AndroidRuntime: FATAL EXCEPTION: main java.lang.SecurityException: com.demo.myapplication: One of RECEIVER_EXPORTED or RECEIVER_NOT_EXPORT…...

深入解读Docker核心原理:Namespace资源隔离机制详解
在容器技术中,资源隔离 是容器化能够实现轻量级虚拟化的关键技术之一。通过资源隔离,容器可以拥有自己的独立环境,确保容器之间互不干扰,从而实现应用的安全和稳定。Docker作为主流的容器平台,其核心的资源隔离机制依赖…...

学习通、智慧职教刷课脚本
🐐个人主页 可惜已不在 🐋可以分享给身边有需要的人🐶 🐉有用的话就留下一个三连吧😼 目录 一.安装 脚本运行器 篡改猴 - Microsoft Edge Addons 二.安装脚本 三.扩展 一.安装 脚本运行器 安装浏览器 Microsoft E…...

SEO写作:从实战到精进的全方位指南
在数字化浪潮中,SEO不再是简单的关键词堆砌,而是成为企业品牌建设与市场拓展的核心策略。作为一名深耕SEO领域的实践者,我深知其中的门道与奥秘。今天,我将结合过往实战经验,以独特视角,带你一窥SEO写作的精…...

解决 git 不是内部或外部命令,也不是可运行的程序
目录 报错提示: 一、解决办法 1、从git官网下载windows版本的git 2、安装 3、注意事项 二、报错 1、解决 fatal: Not a git repository (or any of the parent directories): .git 问题 报错提示: 一、解决办法 Windows下配置Git: 1…...

【卷起来】VUE3.0教程-07-异步请求处理(springboot后端)
🌲 服务端接口准备 pom文件,引入mybatis/mybatis-plus相关依赖 <dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency>&…...

初一信息科技2024指南辅助教学软件(抓包软件)
专门针对信息科技20204指南写的程序,互联网和直播等知识中包含tcp/ip和udp,三次握手等原理,需要简单明了的实验来说明,在机房中需要用抓包软件,可能需要安装windump npcap等软件非常繁琐,还需要接触保护卡&…...

上汽大众:存储成本节约85%,查询性能提升5倍|OceanBase案例
近日,上汽大众汽车有限公司(简称“上汽大众”)的积分卡券等关键业务系统,已成功升级至 OB Cloud 云数据库。借助 OceanBase 原生分布式数据库的卓越性能与先进技术,实现了存储成本的大幅降低,高达85%&#…...
如何快准稳 实现MySQL大表历史数据迁移?
历史迁移解决方案以微服务架构为基础,使用多种设计模式,如:单例、桥接、工厂、模板、策略等。其中涉及的核心技术有多线程、过滤器等,致力于解决MySQL大表迁移的问题,提供多种迁移模式,如:库到库…...
C和指针:函数
函数定义 函数体就是一个代码块,它在函数被调用时执行。 类型 函数名(形式参数) 代码块 与函数定义相反,函数声明出现在函数被调用的地方。 函数声明 编译器是如何知道该函数期望接受的是什么类型和多少数量的参数。 原型 int *find_int( int key…...

Linux——分离部署,分化压力
PQS/TPS 每秒请求数/ 每秒事务数 // 流量衡量参数 可以根据预估QPS 和 服务器的支持的最高QPS 对照计算 就可以得出 需要上架的服务器的最小数量 PV 页面浏览数 UV 独立用户访问量 // 对于网站的总体访问量 response time 响应时间 // 每个请求的响应时间…...
【解密LSTM、GRU如何解决传统RNN梯度消失问题】
解密LSTM与GRU:如何让RNN变得更聪明? 在深度学习的世界里,循环神经网络(RNN)以其卓越的序列数据处理能力广泛应用于自然语言处理、时间序列预测等领域。然而,传统RNN存在的一个严重问题——梯度消失&#…...

相机从app启动流程
一、流程框架图 二、具体流程分析 1、得到cameralist和对应的静态信息 目录如下: 重点代码分析: 启动相机前,先要通过getCameraIdList获取camera的个数以及id,然后可以通过getCameraCharacteristics获取对应id camera的capabilities(静态信息)进行一些openCamera前的…...

select、poll、epoll 与 Reactor 模式
在高并发网络编程领域,高效处理大量连接和 I/O 事件是系统性能的关键。select、poll、epoll 作为 I/O 多路复用技术的代表,以及基于它们实现的 Reactor 模式,为开发者提供了强大的工具。本文将深入探讨这些技术的底层原理、优缺点。 一、I…...

Linux 内存管理实战精讲:核心原理与面试常考点全解析
Linux 内存管理实战精讲:核心原理与面试常考点全解析 Linux 内核内存管理是系统设计中最复杂但也最核心的模块之一。它不仅支撑着虚拟内存机制、物理内存分配、进程隔离与资源复用,还直接决定系统运行的性能与稳定性。无论你是嵌入式开发者、内核调试工…...
uniapp 字符包含的相关方法
在uniapp中,如果你想检查一个字符串是否包含另一个子字符串,你可以使用JavaScript中的includes()方法或者indexOf()方法。这两种方法都可以达到目的,但它们在处理方式和返回值上有所不同。 使用includes()方法 includes()方法用于判断一个字…...

C# 表达式和运算符(求值顺序)
求值顺序 表达式可以由许多嵌套的子表达式构成。子表达式的求值顺序可以使表达式的最终值发生 变化。 例如,已知表达式3*52,依照子表达式的求值顺序,有两种可能的结果,如图9-3所示。 如果乘法先执行,结果是17。如果5…...
LCTF液晶可调谐滤波器在多光谱相机捕捉无人机目标检测中的作用
中达瑞和自2005年成立以来,一直在光谱成像领域深度钻研和发展,始终致力于研发高性能、高可靠性的光谱成像相机,为科研院校提供更优的产品和服务。在《低空背景下无人机目标的光谱特征研究及目标检测应用》这篇论文中提到中达瑞和 LCTF 作为多…...
HybridVLA——让单一LLM同时具备扩散和自回归动作预测能力:训练时既扩散也回归,但推理时则扩散
前言 如上一篇文章《dexcap升级版之DexWild》中的前言部分所说,在叠衣服的过程中,我会带着团队对比各种模型、方法、策略,毕竟针对各个场景始终寻找更优的解决方案,是我个人和我司「七月在线」的职责之一 且个人认为,…...

Unity中的transform.up
2025年6月8日,周日下午 在Unity中,transform.up是Transform组件的一个属性,表示游戏对象在世界空间中的“上”方向(Y轴正方向),且会随对象旋转动态变化。以下是关键点解析: 基本定义 transfor…...

水泥厂自动化升级利器:Devicenet转Modbus rtu协议转换网关
在水泥厂的生产流程中,工业自动化网关起着至关重要的作用,尤其是JH-DVN-RTU疆鸿智能Devicenet转Modbus rtu协议转换网关,为水泥厂实现高效生产与精准控制提供了有力支持。 水泥厂设备众多,其中不少设备采用Devicenet协议。Devicen…...