Spring WebFlux案例

wen java案例 2026-08-03 202

本文目录导读：

Spring WebFlux案例

项目结构
Maven依赖 (pom.xml)
主应用类
模型类
DTO类
Repository层
Service层
Controller层
响应式编程示例控制器
全局异常处理
配置类
测试类
application.yml
响应式最佳实践总结

我将为您提供一个完整的Spring WebFlux实战案例,包括响应式编程的基本概念和具体实现。

项目结构

reactor-demo
├── pom.xml
├── src/main/java/com/example/webflux
│   ├── WebFluxApplication.java
│   ├── config
│   │   └── WebFluxConfig.java
│   ├── controller
│   │   ├── UserController.java
│   │   └── ReactiveController.java
│   ├── service
│   │   ├── UserService.java
│   │   └── ReactiveService.java
│   ├── repository
│   │   └── UserRepository.java
│   ├── model
│   │   └── User.java
│   └── dto
│       ├── UserDTO.java
│       └── ApiResponse.java
└── src/main/resources
    └── application.yml

Maven依赖 (pom.xml)

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 
         http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>3.2.0</version>
        <relativePath/>
    </parent>
    <groupId>com.example</groupId>
    <artifactId>webflux-demo</artifactId>
    <version>1.0.0</version>
    <name>webflux-demo</name>
    <description>Spring WebFlux Demo</description>
    <properties>
        <java.version>17</java.version>
    </properties>
    <dependencies>
        <!-- WebFlux -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-webflux</artifactId>
        </dependency>
        <!-- Reactive MongoDB -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-data-mongodb-reactive</artifactId>
        </dependency>
        <!-- Validation -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-validation</artifactId>
        </dependency>
        <!-- Lombok -->
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <optional>true</optional>
        </dependency>
        <!-- Test -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
        <!-- Reactor Test -->
        <dependency>
            <groupId>io.projectreactor</groupId>
            <artifactId>reactor-test</artifactId>
            <scope>test</scope>
        </dependency>
    </dependencies>
    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
                <configuration>
                    <excludes>
                        <exclude>
                            <groupId>org.projectlombok</groupId>
                            <artifactId>lombok</artifactId>
                        </exclude>
                    </excludes>
                </configuration>
            </plugin>
        </plugins>
    </build>
</project>

主应用类

package com.example.webflux;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.data.mongodb.repository.config.EnableReactiveMongoRepositories;
@SpringBootApplication
@EnableReactiveMongoRepositories
public class WebFluxApplication {
    public static void main(String[] args) {
        SpringApplication.run(WebFluxApplication.class, args);
    }
}

模型类

package com.example.webflux.model;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.springframework.data.annotation.Id;
import org.springframework.data.mongodb.core.mapping.Document;
import java.time.LocalDateTime;
@Data
@Builder
@NoArgsConstructor
@AllArgsConstructor
@Document(collection = "users")
public class User {
    @Id
    private String id;
    private String username;
    private String email;
    private Integer age;
    private String fullName;
    private LocalDateTime createdAt;
    private LocalDateTime updatedAt;
}

DTO类

package com.example.webflux.dto;
import jakarta.validation.constraints.Email;
import jakarta.validation.constraints.Min;
import jakarta.validation.constraints.NotBlank;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
@Data
@Builder
@NoArgsConstructor
@AllArgsConstructor
public class UserDTO {
    private String id;
    @NotBlank(message = "Username is required")
    private String username;
    @Email(message = "Email should be valid")
    @NotBlank(message = "Email is required")
    private String email;
    @Min(value = 18, message = "Age should be at least 18")
    private Integer age;
    private String fullName;
}

package com.example.webflux.dto;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
import java.time.LocalDateTime;
@Data
@Builder
@NoArgsConstructor
@AllArgsConstructor
public class ApiResponse<T> {
    private boolean success;
    private String message;
    private T data;
    private LocalDateTime timestamp;
    public static <T> ApiResponse<T> success(T data) {
        return ApiResponse.<T>builder()
                .success(true)
                .message("Operation successful")
                .data(data)
                .timestamp(LocalDateTime.now())
                .build();
    }
    public static <T> ApiResponse<T> error(String message) {
        return ApiResponse.<T>builder()
                .success(false)
                .message(message)
                .timestamp(LocalDateTime.now())
                .build();
    }
}

Repository层

package com.example.webflux.repository;
import com.example.webflux.model.User;
import org.springframework.data.mongodb.repository.ReactiveMongoRepository;
import org.springframework.stereotype.Repository;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
@Repository
public interface UserRepository extends ReactiveMongoRepository<User, String> {
    Mono<User> findByUsername(String username);
    Mono<User> findByEmail(String email);
    Flux<User> findByAgeGreaterThan(int age);
    Flux<User> findByFullNameContaining(String keyword);
    Mono<Boolean> existsByUsername(String username);
    Mono<Long> countByAgeGreaterThan(int age);
}

Service层

package com.example.webflux.service;
import com.example.webflux.dto.UserDTO;
import com.example.webflux.model.User;
import com.example.webflux.repository.UserRepository;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeanUtils;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import java.time.LocalDateTime;
@Slf4j
@Service
public class UserService {
    private final UserRepository userRepository;
    public UserService(UserRepository userRepository) {
        this.userRepository = userRepository;
    }
    /**
     * 创建用户
     */
    @Transactional
    public Mono<UserDTO> createUser(UserDTO userDTO) {
        log.info("Creating user: {}", userDTO.getUsername());
        User user = new User();
        BeanUtils.copyProperties(userDTO, user);
        user.setCreatedAt(LocalDateTime.now());
        user.setUpdatedAt(LocalDateTime.now());
        return userRepository.save(user)
                .map(this::convertToDTO)
                .doOnSuccess(dto -> log.info("Created user with id: {}", dto.getId()))
                .onErrorResume(ex -> {
                    log.error("Error creating user", ex);
                    return Mono.error(new RuntimeException("Failed to create user"));
                });
    }
    /**
     * 获取所有用户
     */
    public Flux<UserDTO> getAllUsers() {
        return userRepository.findAll()
                .map(this::convertToDTO)
                .doOnComplete(() -> log.info("Retrieved all users"));
    }
    /**
     * 根据ID获取用户
     */
    public Mono<UserDTO> getUserById(String id) {
        return userRepository.findById(id)
                .map(this::convertToDTO)
                .switchIfEmpty(Mono.error(new RuntimeException("User not found with id: " + id)));
    }
    /**
     * 根据用户名获取用户
     */
    public Mono<UserDTO> getUserByUsername(String username) {
        return userRepository.findByUsername(username)
                .map(this::convertToDTO)
                .switchIfEmpty(Mono.error(new RuntimeException("User not found with username: " + username)));
    }
    /**
     * 更新用户
     */
    @Transactional
    public Mono<UserDTO> updateUser(String id, UserDTO userDTO) {
        return userRepository.findById(id)
                .flatMap(existingUser -> {
                    BeanUtils.copyProperties(userDTO, existingUser, "id", "createdAt");
                    existingUser.setUpdatedAt(LocalDateTime.now());
                    return userRepository.save(existingUser);
                })
                .map(this::convertToDTO)
                .switchIfEmpty(Mono.error(new RuntimeException("User not found with id: " + id)));
    }
    /**
     * 删除用户
     */
    @Transactional
    public Mono<Void> deleteUser(String id) {
        return userRepository.findById(id)
                .flatMap(user -> userRepository.delete(user))
                .switchIfEmpty(Mono.error(new RuntimeException("User not found with id: " + id)));
    }
    /**
     * 根据年龄筛选用户
     */
    public Flux<UserDTO> getUsersByAgeGreaterThan(int age) {
        return userRepository.findByAgeGreaterThan(age)
                .map(this::convertToDTO);
    }
    /**
     * 搜索用户
     */
    public Flux<UserDTO> searchUsers(String keyword) {
        return userRepository.findByFullNameContaining(keyword)
                .map(this::convertToDTO);
    }
    /**
     * 统计用户数量
     */
    public Mono<Long> countUsers(int minAge) {
        return userRepository.countByAgeGreaterThan(minAge);
    }
    /**
     * 并发处理示例：批量创建用户
     */
    public Flux<UserDTO> createUsersBatch(Flux<UserDTO> userDTOFlux) {
        return userDTOFlux
                .flatMap(this::createUser)
                .doOnComplete(() -> log.info("Batch creation completed"));
    }
    /**
     * 响应式流程示例：处理用户数据流
     */
    public Flux<UserDTO> processUsersWithOperators() {
        return userRepository.findAll()
                .filter(user -> user.getAge() >= 18)
                .map(this::convertToDTO)
                .doOnNext(dto -> log.info("Processing user: {}", dto.getUsername()))
                .distinct()
                .buffer(10)
                .flatMap(Flux::fromIterable);
    }
    /**
     * 转换实体到DTO
     */
    private UserDTO convertToDTO(User user) {
        UserDTO userDTO = new UserDTO();
        BeanUtils.copyProperties(user, userDTO);
        return userDTO;
    }
}

Controller层

package com.example.webflux.controller;
import com.example.webflux.dto.ApiResponse;
import com.example.webflux.dto.UserDTO;
import com.example.webflux.service.UserService;
import jakarta.validation.Valid;
import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.*;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import java.util.Map;
@RestController
@RequestMapping("/api/users")
public class UserController {
    private final UserService userService;
    public UserController(UserService userService) {
        this.userService = userService;
    }
    /**
     * 创建用户
     */
    @PostMapping
    @ResponseStatus(HttpStatus.CREATED)
    public Mono<ApiResponse<UserDTO>> createUser(@Valid @RequestBody UserDTO userDTO) {
        return userService.createUser(userDTO)
                .map(ApiResponse::success)
                .onErrorResume(ex -> Mono.error(new RuntimeException("Failed to create user: " + ex.getMessage())));
    }
    /**
     * 获取所有用户
     */
    @GetMapping
    public Flux<ApiResponse<UserDTO>> getAllUsers() {
        return userService.getAllUsers()
                .map(ApiResponse::success);
    }
    /**
     * 根据ID获取用户
     */
    @GetMapping("/{id}")
    public Mono<ResponseEntity<ApiResponse<UserDTO>>> getUserById(@PathVariable String id) {
        return userService.getUserById(id)
                .map(userDTO -> ResponseEntity.ok(ApiResponse.success(userDTO)))
                .defaultIfEmpty(ResponseEntity.notFound().build());
    }
    /**
     * 根据用户名获取用户
     */
    @GetMapping("/username/{username}")
    public Mono<ResponseEntity<ApiResponse<UserDTO>>> getUserByUsername(@PathVariable String username) {
        return userService.getUserByUsername(username)
                .map(userDTO -> ResponseEntity.ok(ApiResponse.success(userDTO)))
                .defaultIfEmpty(ResponseEntity.notFound().build());
    }
    /**
     * 更新用户
     */
    @PutMapping("/{id}")
    public Mono<ResponseEntity<ApiResponse<UserDTO>>> updateUser(
            @PathVariable String id,
            @Valid @RequestBody UserDTO userDTO) {
        return userService.updateUser(id, userDTO)
                .map(updatedUser -> ResponseEntity.ok(ApiResponse.success(updatedUser)))
                .defaultIfEmpty(ResponseEntity.notFound().build());
    }
    /**
     * 删除用户
     */
    @DeleteMapping("/{id}")
    public Mono<ResponseEntity<Void>> deleteUser(@PathVariable String id) {
        return userService.deleteUser(id)
                .then(Mono.just(ResponseEntity.noContent().build()));
    }
    /**
     * 按年龄筛选用户
     */
    @GetMapping("/age/above/{age}")
    public Flux<ApiResponse<UserDTO>> getUsersAboveAge(@PathVariable int age) {
        return userService.getUsersByAgeGreaterThan(age)
                .map(ApiResponse::success);
    }
    /**
     * 搜索用户
     */
    @GetMapping("/search")
    public Flux<ApiResponse<UserDTO>> searchUsers(@RequestParam String keyword) {
        return userService.searchUsers(keyword)
                .map(ApiResponse::success);
    }
    /**
     * 统计用户数量
     */
    @GetMapping("/count")
    public Mono<ApiResponse<Long>> countUsers(@RequestParam(defaultValue = "0") int minAge) {
        return userService.countUsers(minAge)
                .map(count -> ApiResponse.success(count));
    }
    /**
     * 错误处理示例
     */
    @GetMapping("/error-test")
    public Mono<ApiResponse<String>> testError() {
        return Mono.error(new RuntimeException("Test exception"))
                .map(result -> ApiResponse.success("Should never reach here"))
                .onErrorResume(ex -> {
                    return Mono.just(ApiResponse.error(ex.getMessage()));
                });
    }
}

响应式编程示例控制器

package com.example.webflux.controller;
import org.springframework.web.bind.annotation.*;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import java.time.Duration;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;
@RestController
@RequestMapping("/api/reactive")
public class ReactiveController {
    /**
     * Flux 操作示例
     */
    @GetMapping("/flux/{count}")
    public Flux<Integer> fluxExample(@PathVariable int count) {
        return Flux.range(1, count)
                .map(i -> i * 2)
                .filter(i -> i % 3 == 0)
                .doOnNext(i -> System.out.println("Processing: " + i));
    }
    /**
     * Mono 操作示例
     */
    @GetMapping("/mono")
    public Mono<String> monoExample() {
        return Mono.just("Hello, Reactive World!")
                .map(String::toUpperCase)
                .delayElement(Duration.ofMillis(500));
    }
    /**
     * Flux with delay
     */
    @GetMapping("/flux/delayed")
    public Flux<Integer> delayedFlux() {
        return Flux.range(1, 5)
                .delayElements(Duration.ofSeconds(1))
                .map(i -> i * i);
    }
    /**
     * 合并多个 Flux
     */
    @GetMapping("/flux/merge")
    public Flux<String> mergeFluxes() {
        Flux<String> source1 = Flux.just("A", "B", "C");
        Flux<String> source2 = Flux.just("D", "E", "F");
        return Flux.merge(source1, source2);
    }
    /**
     * 组合操作
     */
    @GetMapping("/flux/combine")
    public Flux<String> combineFluxes() {
        Flux<Integer> numbers1 = Flux.range(1, 3);
        Flux<String> letters = Flux.just("a", "b", "c");
        return Flux.zip(numbers1, letters)
                .map(tuple -> tuple.getT1() + ":" + tuple.getT2());
    }
    /**
     * 错误处理
     */
    @GetMapping("/error-handling")
    public Flux<String> errorHandling() {
        return Flux.just("data1", "data2", "error")
                .map(data -> {
                    if (data.equals("error")) {
                        throw new RuntimeException("Data processing failed");
                    }
                    return data.toUpperCase();
                })
                .onErrorResume(ex -> Flux.just("Recovered from: " + ex.getMessage()))
                .doFinally(signalType -> System.out.println("Completed with signal: " + signalType));
    }
    /**
     * 背压处理示例
     */
    @GetMapping("/backpressure")
    public Flux<Integer> backpressure() {
        AtomicInteger count = new AtomicInteger(0);
        return Flux.range(1, Integer.MAX_VALUE)
                .map(it -> {
                    int value = count.incrementAndGet();
                    System.out.println("Produced: " + value);
                    return value;
                })
                .onBackpressureBuffer(100);
    }
    /**
     * 数据转换示例
     */
    @GetMapping("/transform")
    public Flux<List<Integer>> transform() {
        return Flux.range(1, 100)
                .map(i -> i * 2)
                .window(10)  // 每10个元素为一个窗口
                .flatMap(Flux::collectList)
                .doOnNext(list -> System.out.println("Window of: " + list.size()));
    }
    /**
     * 条件操作
     */
    @GetMapping("/condition")
    public Mono<String> conditionalOperations() {
        Flux<Integer> numbers = Flux.range(1, 10);
        return numbers
                .filter(n -> n % 2 == 0)
                .take(3)
                .collectList()
                .map(list -> "Even numbers: " + list);
    }
    /**
     * 并行处理
     */
    @GetMapping("/parallel")
    public Flux<String> parallelProcessing() {
        return Flux.range(1, 10)
                .parallel(4)
                .runOn(reactor.core.scheduler.Schedulers.parallel())
                .map(i -> {
                    try {
                        Thread.sleep(100); // 模拟工作
                    } catch (InterruptedException e) {
                        Thread.currentThread().interrupt();
                    }
                    return "Processed: " + i + " on thread " + Thread.currentThread().getName();
                })
                .sequential();
    }
    /**
     * 缓存示例
     */
    @GetMapping("/cached")
    public Flux<String> cachedData() {
        return Flux.fromIterable(Arrays.asList("cached-data-1", "cached-data-2", "cached-data-3"))
                .cache(Duration.ofMinutes(5))
                .doOnSubscribe(subscription -> System.out.println("Subscription started"));
    }
}

全局异常处理

package com.example.webflux.handler;
import com.example.webflux.dto.ApiResponse;
import org.springframework.boot.web.reactive.error.ErrorWebExceptionHandler;
import org.springframework.core.annotation.Order;
import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseEntity;
import org.springframework.http.server.reactive.ServerHttpResponse;
import org.springframework.stereotype.Component;
import org.springframework.validation.BeanPropertyBindingResult;
import org.springframework.web.bind.support.WebExchangeBindException;
import org.springframework.web.server.ResponseStatusException;
import org.springframework.web.server.ServerWebExchange;
import reactor.core.publisher.Mono;
import lombok.extern.slf4j.Slf4j;
@Slf4j
@Order(-2)
@Component
public class GlobalExceptionHandler implements ErrorWebExceptionHandler {
    @Override
    public Mono<Void> handle(ServerWebExchange exchange, Throwable ex) {
        log.error("Global exception handled", ex);
        ServerHttpResponse response = exchange.getResponse();
        if (ex instanceof WebExchangeBindException) {
            WEBExchangeBindException badRequest = (WebExchangeBindException) ex;
            String message = badRequest.getBindingResult()
                    .getFieldErrors()
                    .stream()
                    .map(error -> error.getField() + ": " + error.getDefaultMessage())
                    .collect(Collectors.joining(", "));
            response.setStatusCode(HttpStatus.BAD_REQUEST);
            return writeResponse(exchange, ApiResponse.error(message));
        }
        if (ex instanceof ResponseStatusException) {
            ResponseStatusException statusException = (ResponseStatusException) ex;
            response.setStatusCode(statusException.getStatusCode());
            return writeResponse(exchange, ApiResponse.error(statusException.getMessage()));
        }
        response.setStatusCode(HttpStatus.INTERNAL_SERVER_ERROR);
        return writeResponse(exchange, ApiResponse.error("Internal server error"));
    }
    private Mono<Void> writeResponse(ServerWebExchange exchange, ApiResponse<?> apiResponse) {
        return Mono.fromRunnable(() -> {
            exchange.getResponse().getHeaders().setContentType(MediaType.APPLICATION_JSON);
            try {
                ObjectMapper mapper = new ObjectMapper();
                byte[] bytes = mapper.writeValueAsBytes(apiResponse);
                DataBuffer buffer = exchange.getResponse().bufferFactory().wrap(bytes);
                exchange.getResponse().writeWith(Mono.just(buffer)).subscribe();
            } catch (Exception e) {
                log.error("Error writing response", e);
            }
        });
    }
}

配置类

package com.example.webflux.config;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.http.server.reactive.ServerHttpRequest;
import org.springframework.web.cors.CorsConfiguration;
import org.springframework.web.cors.reactive.CorsWebFilter;
import org.springframework.web.cors.reactive.UrlBasedCorsConfigurationSource;
import org.springframework.web.server.WebFilter;
import reactor.core.scheduler.Scheduler;
import reactor.core.scheduler.Schedulers;
import java.util.concurrent.Executors;
@Configuration
public class WebFluxConfig {
    /**
     * CORS 配置
     */
    @Bean
    public CorsWebFilter corsWebFilter() {
        CorsConfiguration corsConfig = new CorsConfiguration();
        corsConfig.addAllowedOrigin("*");
        corsConfig.addAllowedMethod("*");
        corsConfig.addAllowedHeader("*");
        corsConfig.setMaxAge(3600L);
        UrlBasedCorsConfigurationSource source = new UrlBasedCorsConfigurationSource();
        source.registerCorsConfiguration("/**", corsConfig);
        return new CorsWebFilter(source);
    }
    /**
     * 自定义 Scheduler
     */
    @Bean
    public Scheduler customScheduler() {
        return Schedulers.fromExecutor(Executors.newFixedThreadPool(10));
    }
}

测试类

package com.example.webflux.controller;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.autoconfigure.web.reactive.AutoConfigureWebTestClient;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.http.MediaType;
import org.springframework.test.web.reactive.server.WebTestClient;
import reactor.core.publisher.Flux;
import reactor.test.StepVerifier;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
@AutoConfigureWebTestClient
class UserControllerTest {
    @Autowired
    private WebTestClient webTestClient;
    @Test
    void testGetAllUsers() {
        webTestClient.get()
                .uri("/api/users")
                .accept(MediaType.APPLICATION_JSON)
                .exchange()
                .expectStatus().isOk()
                .expectBodyList(UserDTO.class)
                .hasSize(1);
    }
    @Test
    void testCreateUser() {
        UserDTO userDTO = UserDTO.builder()
                .username("testuser")
                .email("test@example.com")
                .age(25)
                .fullName("Test User")
                .build();
        webTestClient.post()
                .uri("/api/users")
                .contentType(MediaType.APPLICATION_JSON)
                .bodyValue(userDTO)
                .exchange()
                .expectStatus().isCreated()
                .expectBody(ApiResponse.class)
                .consumeWith(response -> {
                    assertNotNull(response.getResponseBody());
                    assertEquals(true, response.getResponseBody().isSuccess());
                });
    }
    @Test
    void testFluxOperations() {
        Flux<Integer> flux = Flux.range(1, 10)
                .map(i -> i * 2)
                .filter(i -> i % 4 == 0);
        StepVerifier.create(flux)
                .expectNext(4, 8, 12, 16, 20)
                .verifyComplete();
    }
    @Test
    void testReactiveErrorHandling() {
        webTestClient.get()
                .uri("/api/reactive/error-handling")
                .accept(MediaType.APPLICATION_JSON)
                .exchange()
                .expectStatus().isOk()
                .expectBodyList(String.class)
                .consumeWith(response -> {
                    assertNotNull(response.getResponseBody());
                    assertEquals(2, response.getResponseBody().size());
                });
    }
}

application.yml

server:
  port: 8080
  reactive:
    netty:
      max-initial-line-length: 10KB
spring:
  application:
    name: webflux-demo
  data:
    mongodb:
      uri: mongodb://localhost:27017/webflux-demo
  webflux:
    base-path: /api
logging:
  level:
    com.example.webflux: DEBUG
    reactor.netty: WARN
    org.springframework.data: WARN
management:
  endpoints:
    web:
      exposure:
        include: health,info,metrics
  endpoint:
    health:
      show-details: always

响应式最佳实践总结

Flux 和 Mono 常用操作符

// 创建流
Flux.just("A", "B", "C");
Flux.range(1, 10);
Flux.interval(Duration.ofSeconds(1));
Mono.just("single");
Mono.empty();
Mono.error(new RuntimeException("Error"));
// 转换操作
Flux.range(1, 10)
    .map(i -> i * 2)                    // 映射
    .flatMap(i -> Mono.just(i * 3))     // 平面映射
    .filter(i -> i % 5 == 0)            // 过滤
    .take(3)                            // 取前n个
    .skip(2)                            // 跳过前n个
    .distinct()                         // 去重
    .collectList()                      // 收集为List
    .collectMap(i -> i);                // 收集为Map
// 错误处理
Flux.error(new RuntimeException())
    .onErrorReturn("default value")     // 返回默认值
    .onErrorResume(ex -> Flux.just("recovered")) // 恢复
    .onErrorMap(ex -> new BusinessException(ex.getMessage())) // 转换错误
    .doOnError(ex -> log.error("Error occurred", ex)) // 记录错误
// 调度
Flux.range(1, 10)
    .subscribeOn(Schedulers.boundedElastic())
    .publishOn(Schedulers.parallel())
    .parallel(4)
    .runOn(Schedulers.parallel())
    .sequential();
// 组合
Flux.concat(flux1, flux2);             // 连接
Flux.merge(flux1, flux2);              // 合并
Flux.zip(flux1, flux2);                // 压缩
Flux.combineLatest(flux1, flux2, func); // 最新组合

性能优化建议

// 1. 使用流式处理代替一次性加载
// 不要
List<User> allUsers = userRepository.findAll().collectList().block();
// 应该
userRepository.findAll()
    .limitRate(100)  // 背压控制
    .map(this::convertToDTO)
    .subscribe();
// 2. 使用批处理
userRepository.findAll()
    .buffer(100)  // 批处理
    .flatMap(users -> processBatch(users));
// 3. 预编译表达式
Predicate<User> agePredicate = user -> user.getAge() > 18;
userRepository.findAll()
    .filter(agePredicate);
// 4. 使用缓存
userRepository.findById(id)
    .cache();  // 缓存结果

这个完整的Spring WebFlux案例包含了：