Spring Boot + WebFlux 全面使用实践指南

<dependencies>
    <!-- WebFlux 核心 -->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-webflux</artifactId>
    </dependency>
    <!-- 响应式数据库（以 MariaDB 为例） -->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-data-r2dbc</artifactId>
    </dependency>
    <dependency>
        <groupId>org.mariadb</groupId>
        <artifactId>r2dbc-mariadb</artifactId>
        <version>1.1.5</version>
    </dependency>
    <dependency>
        <groupId>org.mariadb</groupId>
        <artifactId>mariadb-java-client</artifactId>
        <scope>runtime</scope>
    </dependency>
    <!-- 测试 -->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-test</artifactId>
        <scope>test</scope>
    </dependency>
    <dependency>
        <groupId>io.projectreactor</groupId>
        <artifactId>reactor-test</artifactId>
        <scope>test</scope>
    </dependency>
</dependencies>

@SpringBootApplication
public class WebfluxDemoApplication {
    public static void main(String[] args) {
        SpringApplication.run(WebfluxDemoApplication.class, args);
    }
}

@RestController
@RequestMapping("/users")
public class UserController {
    @Autowired
    private UserRepository userRepository;
    // 返回单个对象
    @GetMapping("/{id}")
    public Mono<User> getUser(@PathVariable String id) {
        return userRepository.findById(id);
    }
    // 返回列表流
    @GetMapping
    public Flux<User> getAllUsers() {
        return userRepository.findAll();
    }
    // 创建用户
    @PostMapping
    public Mono<User> createUser(@RequestBody User user) {
        return userRepository.save(user);
    }
    // SSE：服务器推送事件
    @GetMapping(value = "/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<String> streamEvents() {
        return Flux.interval(Duration.ofSeconds(1))
                   .map(seq -> "Event #" + seq);
    }
}

@Component
public class UserHandler {
    private final UserRepository userRepository;
    public UserHandler(UserRepository userRepository) {
        this.userRepository = userRepository;
    }
    public Mono<ServerResponse> getUser(ServerRequest request) {
        String id = request.pathVariable("id");
        Mono<User> user = userRepository.findById(id);
        return user
            .flatMap(u -> ServerResponse.ok().bodyValue(u))
            .switchIfEmpty(ServerResponse.notFound().build());
    }
    public Mono<ServerResponse> getAllUsers(ServerRequest request) {
        Flux<User> users = userRepository.findAll();
        return ServerResponse.ok().body(users, User.class);
    }
}

@Configuration
public class UserRouter {
    @Bean
    public RouterFunction<ServerResponse> userRoutes(UserHandler handler) {
        return route()
            .GET("/users/{id}", handler::getUser)
            .GET("/users", handler::getAllUsers)
            .build();
    }
}

@Table("users")
public class User {
    @Id
    private Long id;
    private String name;
    private String email;
    // constructors, getters, setters
}

public interface UserRepository extends ReactiveCrudRepository<User, Long> {
    Flux<User> findByEmail(String email);
}

spring:
  r2dbc:
    url: r2dbc:mariadb://localhost:3306/mydb
    username: root
    password: password

@Service
public class ExternalServiceClient {
    private final WebClient webClient;
    public ExternalServiceClient() {
        this.webClient = WebClient.builder()
            .baseUrl("https://api.example.com")
            .build();
    }
    public Mono<UserProfile> fetchProfile(String userId) {
        return webClient.get()
            .uri("/profiles/{id}", userId)
            .retrieve()
            .bodyToMono(UserProfile.class)
            .onErrorResume(e -> Mono.empty()); // 错误降级
    }
}

@ControllerAdvice
public class GlobalExceptionHandler {
    @ExceptionHandler(UserNotFoundException.class)
    public Mono<ResponseEntity<String>> handleUserNotFound(Exception ex) {
        return Mono.just(ResponseEntity
            .status(HttpStatus.NOT_FOUND)
            .body("User not found"));
    }
    @ExceptionHandler(Exception.class)
    public Mono<ResponseEntity<String>> handleGeneral(Exception ex) {
        return Mono.just(ResponseEntity
            .status(HttpStatus.INTERNAL_SERVER_ERROR)
            .body("Internal error"));
    }
}

@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
class UserControllerTest {
    @Autowired
    private WebTestClient webClient;
    @Test
    void shouldGetUser() {
        webClient.get().uri("/users/1")
            .exchange()
            .expectStatus().isOk()
            .expectBody(User.class)
            .value(user -> assertThat(user.getName()).isEqualTo("Alice"));
    }
}

server:
  netty:
    connection-timeout: 30s
    max-in-memory-size: 10MB  # 防止 OOM

Flux.range(1, 1000)
    .limitRate(100) // 控制上游发射速率
    .onBackpressureBuffer(500); // 缓冲溢出数据