Reactive Programming in Java

2025 年 9 月 5 日 | 阅读 5 分钟

当今的应用程序需要出色的并发管理能力，以及可扩展性和快速响应能力。鉴于当今应用程序的需求，响应式编程作为一种处理异步数据流的解决方案应运而生。

这种方法允许应用程序在无需等待响应的情况下处理实时事件，从而提高应用程序的速度并最大限度地减少资源使用。

来自 Java 的 Project Reactor 通过其用于多值的 Flux API 和用于单个值的 Mono，帮助开发人员执行非阻塞操作。以下文章将解释如何使用Java Reactor 实现响应式编程。

什么是响应式编程？

响应式编程的编程模型侧重于事件驱动系统的高效运行，这些系统会管理数据流。与传统的阻塞调用（其中线程等待响应）不同，响应式系统可以在不阻塞执行的情况下同时处理多个任务。应用程序将受益于改进的可伸缩性、弹性和更佳的响应能力。

响应式编程模型在 Reactive Streams 规范定义的四项原则下运行

响应性 (Responsiveness)：系统必须能够及时响应。
弹性 (Resilience)：系统在故障条件下应保持稳定。
弹性 (Elasticity)：系统应根据需求动态扩展。
消息驱动 (Message-Driven)：与基于轮询的方法相比，基于事件的消息系统在通信方面是一种更好的方法。

为什么响应式编程有用？

响应式编程对于需要以下能力的应用程序非常有用：

高并发：能够高效地并行处理多个请求。
低延迟：通过避免阻塞调用来减少响应时间。
可伸缩性：能够动态适应工作负载的变化。
高效的资源利用：通过减少空闲线程来最大限度地利用 CPU 和内存。

Project Reactor 中的关键概念

Flux：处理一系列多个元素（如数据流）。

Mono：处理单个值或空结果。

背压 (Backpressure)：控制数据流，以防止系统过载。

错误处理 (Error Handling)：提供在数据流中优雅处理错误的策略。

依赖关系

要在 Spring Boot WebFlux 项目中使用 Project Reactor，请在 xml 文件中包含以下依赖项。

文件名：pom.xml

 
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-webflux</artifactId>
</dependency>
<dependency>
    <groupId>io.projectreactor</groupId>
    <artifactId>reactor-core</artifactId>
    <version>3.6.7</version>
</dependency>   

文件名：Main.java

 
package org.example;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.publisher.BaseSubscriber;
public class Main {
    public static void main(String[] args) {
        System.out.println("=== Flux Example ===");
        fluxExample();
        System.out.println("\n=== Mono Example ===");
        monoExample();
        System.out.println("\n=== Error Handling Example ===");
        errorHandlingExample();
        System.out.println("\n=== Backpressure Example ===");
        backpressureExample();
    }
    private static void fluxExample() {
        Flux<String> namesFlux = Flux.just("Alice", "Bob", "Charlie", "David")
                .map(String::toUpperCase)
                .filter(name -> name.startsWith("A") || name.startsWith("D"));
        namesFlux.subscribe(
                name -> System.out.println("Received: " + name),
                error -> System.err.println("Error: " + error),
                () -> System.out.println("Processing complete!")
        );
    }
    private static void monoExample() {
        Mono<String> singleItem = Mono.just("Hello, Reactive World!");
        singleItem.subscribe(
                item -> System.out.println("Mono Output: " + item),
                error -> System.err.println("Error: " + error),
                () -> System.out.println("Mono Processing complete!")
        );
    }
    private static void errorHandlingExample() {
        Flux<Integer> numbers = Flux.just(1, 2, 3, 4, 5)
                .map(n -> {
                    if (n == 3) throw new RuntimeException("Error at 3");
                    return n * 2;
                })
                .onErrorReturn(-1);
        numbers.subscribe(System.out::println);
    }
    private static void backpressureExample() {
        Flux<Integer> flux = Flux.range(1, 10);
        flux.subscribe(new BaseSubscriber<Integer>() {
            @Override
            protected void hookOnSubscribe(org.reactivestreams.Subscription subscription) {
                System.out.println("Requesting 3 items...");
                request(3);
            }
            @Override
            protected void hookOnNext(Integer value) {
                System.out.println("Received: " + value);
            }
        });
    }
}

输出

 
=== Flux Example ===
Received: ALICE
Received: DAVID
Processing complete!
=== Mono Example ===
Mono Output: Hello, Reactive World!
Mono Processing complete!
=== Error Handling Example ===
2
4
-1
=== Backpressure Example ===
Requesting 3 items...
Received: 1
Received: 2
Received: 3

解释

此程序演示了使用 Project Reactor 实现响应式编程的关键概念。它首先创建一个 Flux (fluxExample)，该 Flux 发出名称，将其转换为大写，并仅过滤掉以“A”或“D”开头的名称。

然后，Mono 示例 (monoExample) 创建一个具有问候消息的单值流。错误处理 (errorHandlingExample) 演示了如何使用 .onErrorReturn() 来捕获数据流中的错误并进行替换。

最后，使用 BaseSubscriber 实现背压 (backpressureExample)，允许程序以受控批次请求数据，而不是一次性接收所有数据。这可以防止消费者过载并优化资源使用。

响应式编程的优势

更好的资源利用：非阻塞执行可实现高效的 CPU 和内存使用。

可伸缩性：支持并发操作而不会发生线程争用。

改进的响应能力：即使在高负载下，应用程序也能保持响应。

响应式编程的应用场景

微服务架构：响应式编程为微服务架构提供了一种构建松散耦合的事件驱动系统的解决方案。

流式应用程序：流式应用程序最适合处理涉及股票价格、社交媒体流和传感器信息的实时操作。

数据库交互：通过减少阻塞调用引起的延迟来增强数据库操作。

API 网关：用于 API 网关，以高效地处理成千上万的并发请求。

结论

应用程序需要响应式编程来开发提供高性能、可伸缩性和非阻塞功能的系统。借助 Project Reactor，开发人员可以轻松处理异步数据流，实现错误处理，并高效地控制数据流。通过利用 Flux、Mono 和 Backpressure，应用程序可以无缝地处理大型数据流，同时保持系统稳定。

响应式编程选择题

Q1. 响应式编程的主要目标是什么？

简化数据库查询
处理同步数据流
高效管理异步数据流和事件
使用更多 CPU 线程以获得更好的性能

答案：C

解释：响应式编程旨在高效地管理异步数据流，使应用程序能够在不阻塞执行的情况下处理实时事件。

Q2. 响应式流的哪个原则确保系统在故障条件下保持稳定？

弹性 (Elasticity)
响应性 (Responsiveness)
消息驱动 (Message-Driven)
弹性 (Resilience)

答案： D

解释：弹性 (Resilience) 是确保系统能够从故障中恢复并继续运行的原则。

Q3. 响应式编程中的背压 (backpressure) 是什么？

一种提高数据速度的机制
一种缓存数据以供以后使用的方法
一种控制数据流以防止订阅者过载的方法
一种终止数据流的方法

答案：C

解释：背压 (Backpressure) 是响应式编程中的一个关键概念，它允许订阅者控制接收到的数据量，以避免过载。

Q4. 在 Project Reactor 中，下列哪个用于表示多个值的流？

Mono
Flux
可选
Stream

答案： B

解释：Flux 在 Project Reactor 中用于表示 0 到 N 个元素的流，非常适合异步处理数据序列。

Q5. Project Reactor 中的 Mono 类代表什么？

值列表
可选值
一个或零个值的流
阻塞线程

答案：C

解释：Mono 代表一个可以发出 0 个或 1 个值的发布者。当您期望单个结果或根本没有结果时使用它。

下一个主题Right-truncatable-prime-in-java

Reactive Programming in Java

什么是响应式编程？

为什么响应式编程有用？

Project Reactor 中的关键概念

依赖关系

解释

响应式编程的优势

响应式编程的应用场景

结论

响应式编程选择题

联系信息

关注我们

教程

面试题

在线编译器

Python

Java

.Net Framework

AI, ML and Data Science

Cloud Technology

B.Tech and MCA

Web Technology

PHP

Software Testing

Technical Interview

Java Interview

Python

Web Interview

Database Interview

B.Tech / MCA

Important Interview

Software Testing Interview

Company Interviews

Online Compilers

Multiple Choice Questions

Java Conversion

Java Misc

Reactive Programming in Java

什么是响应式编程？

为什么响应式编程有用？

Project Reactor 中的关键概念

依赖关系

解释

响应式编程的优势

响应式编程的应用场景

结论

响应式编程选择题

相关帖子

Java 中的加一问题

Instance Variable in Java

Right-Truncatable Prime in Java

Could Not Find or Load Main Class in Java

Java 中的 Atomic 与 Synchronized 对比

How to Print in Java

如何解决 Java 中的不兼容类型错误

转换手机数字键盘序列为等效句子

Java 密码生成器

创建 Java 临时文件

订阅 Tpoint Tech

联系信息

关注我们

教程

面试题

在线编译器