响应式编程与 WebFlux——从 Reactor 到高并发架构

一、引言

传统 Servlet 容器（Tomcat、Jetty）基于”一个请求一个线程”的阻塞 I/O 模型。当面对高并发、I/O 密集型的场景时，大量线程的上下文切换和资源占用成为性能瓶颈。响应式编程（Reactive Programming）正是为应对这类挑战而生——它通过异步、非阻塞和背压（Backpressure）机制，以极少的线程资源处理海量并发请求。

Spring WebFlux 是 Spring 5 引入的响应式 Web 框架，底层基于 Project Reactor，支持 Netty、Undertow 等非阻塞服务器。本文将深入分析响应式编程理念、Project Reactor 的核心类型、WebFlux 架构、背压机制，并通过性能对比展示响应式架构的优势与适用场景。

二、响应式编程理念

2.1 什么是响应式编程

响应式编程是一种基于数据流和变化传播的异步编程范式。其核心思想是：声明如何处理数据，而非控制数据的流动。

响应式宣言（Reactive Manifesto）定义了响应式系统的四个特征：
1. Responsive（响应灵敏）：系统及时响应
2. Resilient（韧性）：系统在故障时保持可用
3. Elastic（弹性）：系统在不同负载下保持响应
4. Message Driven（消息驱动）：组件间通过异步消息传递

// 命令式编程 vs 响应式编程
// 命令式：主动拉取（Pull）
List<User> users = userRepository.findAll();
List<String> names = new ArrayList<>();
for (User u : users) {
    if (u.getAge() > 18) {
        names.add(u.getName().toUpperCase());
    }
}

// 响应式：声明式推流（Push）
Flux<User> userFlux = reactiveUserRepository.findAll();
Flux<String> namesFlux = userFlux
    .filter(u -> u.getAge() > 18)
    .map(u -> u.getName().toUpperCase());

2.2 Reactive Streams 规范

Reactive Streams 是响应式编程的标准规范，定义了四个核心接口：

接口	角色	职责
`Publisher`	发布者	产生数据流
`Subscriber`	订阅者	消费数据流
`Subscription`	订阅契约	管理请求与取消
`Processor`	处理器	既是发布者也是订阅者

// Reactive Streams 核心接口定义
public interface Publisher<T> {
    void subscribe(Subscriber super T> subscriber);
}

public interface Subscriber<T> {
    void onSubscribe(Subscription s);      // 收到订阅令牌
    void onNext(T t);                      // 收到下一个元素
    void onError(Throwable t);             // 发生错误
    void onComplete();                     // 流结束
}

public interface Subscription {
    void request(long n);  // 请求 n 个元素（背压）
    void cancel();         // 取消订阅
}

三、Project Reactor 核心类型

3.1 Mono 和 Flux

Reactor 提供了两种核心反应式类型：

Mono：包含 0 或 1 个元素的异步序列
Flux：包含 0 到 N 个元素的异步序列

graph LR
    subgraph Mono["Mono - 0..1 个元素"]
        M1[--onNext-->]
        M2[--onComplete-->]
    end

    subgraph Flux["Flux - 0..N 个元素"]
        F1[--onNext-->]
        F2[--onNext-->]
        F3[--onNext-->]
        F4[...-->]
        F5[--onComplete-->]
    end

创建 Flux/Mono：

// 从值创建
Mono<String> mono = Mono.just("Hello");
Flux<String> flux = Flux.just("A", "B", "C");
Flux<Integer> range = Flux.range(1, 10);

// 从集合创建
Flux<User> users = Flux.fromIterable(userList);
Mono<List<User>> listMono = Flux.fromIterable(userList).collectList();

// 从 Callable/Supplier
Mono<String> fromCallable = Mono.fromCallable(() -> {
    Thread.sleep(1000);
    return "result";
});

// 从 Future
Mono<String> fromFuture = Mono.fromFuture(
    CompletableFuture.supplyAsync(() -> "async result"));

3.2 操作符

转换操作符：

// map：同步转换
Flux<String> names = users
    .map(User::getName)            // 1:1 转换
    .map(String::toUpperCase);

// flatMap：异步展开
Flux<Order> orders = Flux.fromIterable(userIds)
    .flatMap(id -> orderService.getOrders(id))  // 每个 id 返回 Flux
    .flatMap(order -> orderService.getDetails(order), 5);  // 并发度 5

// concatMap：保持顺序的 flatMap
Flux<Order> ordered = Flux.fromIterable(userIds)
    .concatMap(id -> orderService.getOrders(id));

// flatMapSequential：按订阅顺序 emit，保持原始顺序

过滤操作符：

Flux<Integer> filtered = Flux.range(1, 100)
    .filter(i -> i % 2 == 0)
    .skip(10)           // 跳过前 10 个
    .take(20)           // 只取 20 个
    .distinct()         // 去重
    .distinctUntilChanged();  // 连续重复消除

组合操作符：

// zip：合并多个流（一一对应）
Flux<String> zipped = Flux.zip(
    Flux.just("A", "B", "C"),
    Flux.just("1", "2", "3"),
    (letter, number) -> letter + number
);  // A1, B2, C3

// merge：交错合并（按到达时间）
Flux<String> merged = Flux.merge(
    service1.getData(),
    service2.getData()
);

// concat：串联（保持顺序）
Flux<String> concatenated = Flux.concat(
    service1.getData(),
    service2.getData()
);

错误处理：

Flux<String> safe = flux
    .onErrorReturn("fallback")              // 发生错误时返回默认值
    .onErrorResume(e -> {
        log.error("Error occurred", e);
        return Flux.just("recovered1", "recovered2");
    })                                      // 从错误恢复
    .onErrorMap(e -> new BusinessException("Wrapped", e))  // 转换异常
    .retry(3)                               // 重试 3 次
    .timeout(Duration.ofSeconds(5));        // 超时

四、Spring WebFlux 架构

4.1 WebFlux vs MVC

graph TD
    subgraph MVC[Spring MVC - Servlet Stack]
        Req1[Request] --> TC[Tomcat Thread Pool]
        TC --> S1[Servlet]
        S1 --> C1[Controller]
        C1 --> S2[Service - blocking JDBC]
        S2 --> DB[(Database)]
    end

    subgraph WebFlux[Spring WebFlux - Reactive Stack]
        Req2[Request] --> EL[Event Loop<br/>Netty Worker Threads]
        EL --> H[Handler/Controller]
        H --> S3[Reactive Service]
        S3 --> R2DBC[(R2DBC - reactive DB)]
        EL --> EH[Event Handler]
    end

特性	Spring MVC	Spring WebFlux
IO 模型	阻塞（Blocking）	非阻塞（Non-blocking）
线程模型	每个请求一个线程	Event Loop（少量线程）
服务器	Tomcat、Jetty、Undertow	Netty、Undertow、Servlet 3.1+
数据库访问	JDBC（阻塞）	R2DBC、MongoDB Reactive
处理方式	`@RequestMapping` + `@ResponseBody`	`@RequestMapping` + `Mono/Flux`
背压	无	原生支持

4.2 WebFlux Controller 示例

@RestController
@RequestMapping("/api/orders")
public class OrderReactiveController {

    @Autowired
    private ReactiveOrderService orderService;
    @Autowired
    private ReactiveUserService userService;

    // 返回 Mono（0 或 1 个元素）
    @GetMapping("/{id}")
    public Mono<ResponseEntity<OrderDTO>> getOrder(@PathVariable String id) {
        return orderService.findById(id)
            .map(ResponseEntity::ok)
            .defaultIfEmpty(ResponseEntity.notFound().build());
    }

    // 返回 Flux（多元素流）
    @GetMapping
    public Flux<OrderDTO> listOrders(@RequestParam(required = false) String status) {
        return orderService.findByStatus(status);
    }

    // 组合多个异步调用
    @GetMapping("/detail/{id}")
    public Mono<OrderDetailDTO> getOrderDetail(@PathVariable String id) {
        return orderService.findById(id)
            .flatMap(order -> 
                Mono.zip(
                    userService.findById(order.getUserId()),
                    productService.findByIds(order.getProductIds()).collectList(),
                    (user, products) -> OrderDetailDTO.builder()
                        .order(order)
                        .user(user)
                        .products(products)
                        .build()
                )
            );
    }
}

4.3 Reactive Repository

// MongoDB Reactive
public interface ReactiveOrderRepository 
    extends ReactiveMongoRepository<Order, String> {

    Flux<Order> findByUserId(String userId);
    Flux<Order> findByStatusOrderByCreatedAtDesc(String status);
    Mono<Long> countByStatus(String status);
}

// R2DBC（响应式关系型数据库）
public interface ReactiveUserRepository 
    extends ReactiveCrudRepository<User, Long> {

    Flux<User> findByAgeGreaterThan(int age);
    Mono<User> findByEmail(String email);
}

4.4 WebClient（响应式 HTTP 客户端）

@Service
public class ReactiveProductClient {

    private final WebClient webClient;

    public ReactiveProductClient() {
        this.webClient = WebClient.builder()
            .baseUrl("http://product-service")
            .defaultHeader(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE)
            .build();
    }

    public Mono<ProductDTO> getProduct(Long id) {
        return webClient.get()
            .uri("/api/products/{id}", id)
            .retrieve()
            .bodyToMono(ProductDTO.class)
            .timeout(Duration.ofSeconds(5))
            .onErrorResume(e -> {
                log.warn("Failed to get product {}: {}", id, e.getMessage());
                return Mono.empty();
            });
    }

    public Flux<ProductDTO> getProducts(List<Long> ids) {
        return Flux.fromIterable(ids)
            .flatMap(this::getProduct, 10);  // 并发 10 个请求
    }
}

五、背压机制

5.1 什么是背压

背压（Backpressure）是响应式流的核心机制，它允许下游 Subscriber 告诉上游 Publisher “我一次只能处理 N 个元素”，从而防止消费者被生产者淹没。

// 没有背压的灾难场景
Flux.interval(Duration.ofMillis(1))  // 每秒产生 1000 个元素
    .subscribe(i -> {
        Thread.sleep(100);  // 每次处理需要 100ms
        System.out.println(i);
    });
// 无缓冲情况下的背压缺失会导致 OutOfMemoryError

// 有背压的控制
Flux.interval(Duration.ofMillis(1))
    .onBackpressureBuffer(1000)  // 缓冲 1000 个
    .onBackpressureDrop()        // 溢出丢弃
    .onBackpressureLatest()      // 保留最新的
    .subscribe(i -> {
        Thread.sleep(100);
        System.out.println(i);
    });

5.2 背压策略

策略	行为	适用场景
BUFFER	无限制缓冲（OOM 风险）	数据量可知且较小
DROP	丢弃无法处理的事件	实时数据，允许丢失
LATEST	只保留最新事件	股票行情、监控数据
ERROR	溢出时抛出异常	数据完整性要求高
IGNORE	忽略背压（不推荐）	绝不用在生产环境

// 手动控制背压
Flux<String> source = Flux.generate(sink -> {
    sink.next(fetchData());
});

source.subscribe(new BaseSubscriber<>() {
    @Override
    protected void hookOnSubscribe(Subscription subscription) {
        request(10);  // 第一次请求 10 个
    }

    @Override
    protected void hookOnNext(String value) {
        process(value);
        if (isReady()) {
            request(5);  // 处理完一批，再请求 5 个
        }
    }
});

六、性能对比与选择建议

6.1 WebFlux vs MVC 性能

在 I/O 密集型场景下（如大量数据库查询、远程调用）：

指标	MVC（Tomcat）	WebFlux（Netty）	提升
最大并发连接数	200（默认线程池）	数千-数万	10-50x
内存占用（1000 并发）	~800MB	~200MB	4x
CPU 利用率	高（线程切换）	低（事件驱动）	2-3x
P99 延迟（高并发）	不稳定	稳定	显著改善

6.2 何时选择 WebFlux

适合 WebFlux 的场景：
– I/O 密集型应用（大量 HTTP 调用、数据库查询）
– 网关和代理服务（Zuul 的替代方案）
– 实时数据流（SSE、WebSocket）
– 需要极高并发的场景

不适合 WebFlux 的场景：
– CPU 密集型任务（计算为主）
– 简单的 CRUD 应用
– 团队对响应式编程不熟悉
– 依赖阻塞式中间件（传统 JDBC 驱动）

七、总结

响应式编程代表了一种范式转变——从”拉取”数据到”推送”数据，从阻塞到非阻塞，从命令式到声明式。

Reactive Streams 规范为不同响应式库之间的互操作性提供了标准
Project Reactor 的 Mono 和 Flux 提供了丰富的操作符，使声明式数据处理成为可能
Spring WebFlux 基于 Netty 的事件循环模型，能以极少的线程资源处理海量请求
背压机制确保了系统的稳定性，防止背压问题导致 OOM

响应式编程并非银弹。对于大多数业务系统，传统的 Spring MVC 已经足够。但在网关、实时流处理和高并发 I/O 场景下，WebFlux 的优势是 MVC 无法替代的。选择哪种技术栈，取决于你的实际问题：”不要把响应式当作时髦的技术来追求，而应在真正需要的时候使用它。”

文章版权归作者所有，未经允许请勿转载。

THE END

Java

响应式编程与 WebFlux——从 Reactor 到高并发架构

响应式编程与 WebFlux——从 Reactor 到高并发架构

一、引言

二、响应式编程理念

2.1 什么是响应式编程

2.2 Reactive Streams 规范

三、Project Reactor 核心类型

3.1 Mono 和 Flux

3.2 操作符

四、Spring WebFlux 架构

4.1 WebFlux vs MVC

4.2 WebFlux Controller 示例

4.3 Reactive Repository

4.4 WebClient（响应式 HTTP 客户端）

五、背压机制

5.1 什么是背压

5.2 背压策略

六、性能对比与选择建议

6.1 WebFlux vs MVC 性能

6.2 何时选择 WebFlux

七、总结

请登录后发表评论