Spring WebFlux 响应式：非阻塞与异步流

Spring WebFlux 响应式编程入门：构建非阻塞异步流应用

在微服务和高并发场景下，传统的同步阻塞模型正面临性能瓶颈。Spring WebFlux 作为 Spring 生态中的响应式 Web 框架，基于 Reactor 库实现了完全非阻塞的异步流处理，让开发者能更高效地利用系统资源。本教程将带你从零掌握 WebFlux 的核心思想与实践。

1. 为什么需要响应式编程？

1.1 传统 Servlet 模型的局限性

传统的 Spring MVC 基于 Servlet API，每个请求会绑定一个线程（如 Tomcat 线程池），若业务中包含阻塞操作（如数据库查询、远程调用），线程将被挂起，直到操作完成。当并发量增大时，线程池可能耗尽，导致系统吞吐量急剧下降。

1.2 非阻塞与异步流的优势

响应式编程采用事件驱动和异步非阻塞的方式。线程不会因等待 I/O 而阻塞，而是在 I/O 就绪时接收通知，从而使用少量线程处理海量并发请求。WebFlux 底层可运行在 Netty、Undertow 等非阻塞服务器上，天然适配高吞吐服务。

2. 核心基石：Reactor 与背压

WebFlux 使用 Project Reactor 作为其响应式库，核心类型是 Mono 和 Flux。

• Mono：表示 0 或 1 个元素的异步序列。
• Flux：表示 0 到 N 个元素的异步序列。

2.1 创建响应式流

Mono<String> mono = Mono.just("Hello");
Flux<Integer> flux = Flux.just(1, 2, 3, 4, 5);

流本身是惰性的（冷流），只有在被订阅时才会触发数据生产。

mono.subscribe(data -> System.out.println("收到: " + data));

2.2 操作符与链式处理

Reactor 提供了丰富的操作符，对数据进行转换、过滤、组合，形成异步处理链。

Flux.range(1, 10)
    .filter(n -> n % 2 == 0)
    .map(n -> n * 10)
    .subscribe(System.out::println);  // 输出 20 40 60...

2.3 背压（Backpressure）机制

背压是响应式系统的关键能力，当下游消费速度慢于上游生产速度时，下游可以通过 request(n) 通知上游只发送 n 个元素，避免内存溢出。

Flux.range(1, 100)
    .log()
    .subscribe(new BaseSubscriber<Integer>() {
        @Override
        protected void hookOnSubscribe(Subscription subscription) {
            request(5); // 仅请求 5 个元素
        }
        // 可动态控制 request 数量
    });

3. WebFlux 应用开发模型

WebFlux 提供两种编程模型：注解式控制器（与 Spring MVC 风格一致）和函数式端点（Router Functions）。两者底层均运行在响应式栈上。

3.1 搭建项目依赖

在 pom.xml 中加入 spring-boot-starter-webflux，Spring Boot 会自动配置 Netty 作为默认服务器。

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-webflux</artifactId>
</dependency>

3.2 注解式控制器

用法与 MVC 类似，但返回值类型为 Mono 或 Flux。

@RestController
@RequestMapping("/users")
public class UserController {
    private final UserService userService;

    public UserController(UserService userService) {
        this.userService = userService;
    }

    @GetMapping("/{id}")
    public Mono<User> getUser(@PathVariable String id) {
        return userService.findById(id);
    }

    @GetMapping
    public Flux<User> listUsers() {
        return userService.findAll();
    }

    @PostMapping
    @ResponseStatus(HttpStatus.CREATED)
    public Mono<User> createUser(@RequestBody Mono<User> userMono) {
        return userService.save(userMono);
    }
}

处理器方法不会阻塞线程，findById 返回的 Mono 会异步将结果写入响应。

3.3 函数式端点（Router & Handler）

将路由与处理逻辑分离，更加函数式风格。

@Configuration
public class UserRouter {
    @Bean
    public RouterFunction<ServerResponse> route(UserHandler handler) {
        return RouterFunctions
            .route(GET("/users"), handler::listUsers)
            .andRoute(GET("/users/{id}"), handler::getUser)
            .andRoute(POST("/users"), handler::createUser);
    }
}

@Component
public class UserHandler {
    private final UserRepository repository;

    public UserHandler(UserRepository repository) {
        this.repository = repository;
    }

    public Mono<ServerResponse> listUsers(ServerRequest request) {
        return ServerResponse.ok().body(repository.findAll(), User.class);
    }

    public Mono<ServerResponse> getUser(ServerRequest request) {
        String id = request.pathVariable("id");
        return repository.findById(id)
                .flatMap(user -> ServerResponse.ok().bodyValue(user))
                .switchIfEmpty(ServerResponse.notFound().build());
    }

    public Mono<ServerResponse> createUser(ServerRequest request) {
        return request.bodyToMono(User.class)
                .flatMap(repository::save)
                .flatMap(saved -> ServerResponse.created(URI.create("/users/" + saved.getId()))
                        .bodyValue(saved));
    }
}

4. 响应式数据访问

WebFlux 需要与响应式数据库驱动配合，才能实现端到端的非阻塞。常用支持响应式的数据库有 MongoDB、Redis、R2DBC（关系型数据库）等。

4.1 响应式 MongoDB 示例

引入 spring-boot-starter-data-mongodb-reactive。

@Repository
public interface UserRepository extends ReactiveMongoRepository<User, String> {
    Flux<User> findByAgeGreaterThan(int age);
}

4.2 响应式 Redis

使用 spring-boot-starter-data-redis-reactive，操作均返回 Mono/Flux。

4.3 R2DBC 连接关系型数据库

依赖 spring-boot-starter-data-r2dbc，并配置连接工厂，例如：

spring.r2dbc.url=r2dbc:postgresql://localhost:5432/mydb
spring.r2dbc.username=user
spring.r2dbc.password=password

定义响应式仓库：

public interface UserR2dbcRepository extends ReactiveCrudRepository<User, Long> {
    Mono<User> findByUsername(String username);
}

5. 错误处理与调试

5.1 声明式错误处理

在流中通过操作符处理异常：

• onErrorReturn：返回默认值。
• onErrorResume：切换到备选流。
• doOnError：仅记录日志，不改变流走向。

public Mono<User> findUserWithFallback(String id) {
    return repository.findById(id)
            .onErrorResume(e -> {
                log.error("查询失败", e);
                return Mono.just(User.ANONYMOUS);
            });
}

5.2 在控制器层面处理异常

使用 @ExceptionHandler 注解在 @RestControllerAdvice 中全局处理：

@RestControllerAdvice
public class GlobalExceptionHandler {
    @ExceptionHandler(UserNotFoundException.class)
    @ResponseStatus(HttpStatus.NOT_FOUND)
    public Mono<ErrorResponse> handleNotFound(UserNotFoundException ex) {
        return Mono.just(new ErrorResponse(404, ex.getMessage()));
    }
}

5.3 调试技巧

• 在操作链中插入 .log() 打印信号流（订阅、请求、onNext、onError等）。
• 使用 Hooks.onOperatorDebug() 启用操作符调试模式（生产慎用，有性能开销）。
• 借助 BlockHound（Java Agent）检测代码中的意外阻塞调用。

6. 测试响应式组件

Spring 提供了 WebTestClient 进行集成测试，可使用 StepVerifier 对 Reactor 流进行断言。

6.1 使用 WebTestClient

@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
class UserRouterTest {
    @Autowired
    WebTestClient webTestClient;

    @Test
    void testGetUser() {
        webTestClient.get().uri("/users/1")
                .exchange()
                .expectStatus().isOk()
                .expectBody(User.class)
                .consumeWith(response -> 
                    Assertions.assertEquals("张三", response.getResponseBody().getName()));
    }
}

6.2 使用 StepVerifier

@Test
void testFluxElements() {
    StepVerifier.create(userService.findAll())
            .expectNextCount(3)
            .verifyComplete();
}

7. 传统 Spring MVC 与 WebFlux 对比

选择建议：如果现有系统大量依赖 JDBC 等同步阻塞 API，强制迁移无法获得响应式收益，可保留 MVC；若服务为高并发网关、流处理、或使用响应式数据持久化，WebFlux 是更合适的选择。

8. 实战注意事项

• 全链响应式：从控制器到数据库（或远程调用）整个链路必须是非阻塞，否则一处阻塞会使整体退化为同步。
• 避免阻塞代码：不要在 Reactor 流内调用 Thread.sleep() 或 JDBC 等阻塞 API，若有必须的阻塞操作，使用 subscribeOn(Schedulers.boundedElastic()) 将其调度到弹性线程池。
• Operators 顺序：操作符的顺序会影响行为和性能，需要仔细规划。
• 资源清理：使用 using 或 doFinally 确保连接、文件等资源被正确释放。

9. 总结

Spring WebFlux 为 Java 开发者打开了响应式编程的大门，通过非阻塞异步流极大提升了资源利用率和吞吐量。掌握 Reactor 的 Mono 与 Flux、注解/函数式两种开发模型以及全链路响应式思想，你便能够在高并发项目中游刃有余地构建现代微服务。下一步可以深入探索 WebFlux 与 RSocket、WebSocket 的结合，以及响应式安全配置等高级主题。