Spring WebFlux案例

本文目录导读：

1. 项目结构
2. Maven依赖 (pom.xml)
3. 主应用类
4. 模型类
5. DTO类
6. Repository层
7. Service层
8. Controller层
9. 响应式编程示例控制器
10. 全局异常处理
11. 配置类
12. 测试类
13. application.yml
14. 响应式最佳实践总结

我将为您提供一个完整的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案例包含了：

1. 响应式编程基础：Mono和Flux的创建与操作
2. 完整的CRUD操作：用户管理
3. 响应式MongoDB：数据持久化
4. 全局异常处理：错误处理机制
5. 响应式测试：测试用例示例
6. 性能优化：背压、调度器、批处理等最佳实践
7. 高级特性：并行处理、缓存、错误恢复等

这个案例可以作为您学习和使用Spring WebFlux的参考模板。