WebFlux 核心组件

注解式编程模型

WebFlux 支持与 Spring MVC 类似的注解式编程，使用 @Controller 和 @RestController。

基本控制器

@RestController
@RequestMapping("/api/users")
public class UserController {
    
    private final UserService userService;
    
    public UserController(UserService userService) {
        this.userService = userService;
    }
    
    @GetMapping
    public Flux<User> getAllUsers() {
        return userService.findAll();
    }
    
    @GetMapping("/{id}")
    public Mono<User> getUserById(@PathVariable String id) {
        return userService.findById(id);
    }
    
    @PostMapping
    @ResponseStatus(HttpStatus.CREATED)
    public Mono<User> createUser(@RequestBody User user) {
        return userService.save(user);
    }
    
    @PutMapping("/{id}")
    public Mono<User> updateUser(@PathVariable String id, @RequestBody User user) {
        return userService.update(id, user);
    }
    
    @DeleteMapping("/{id}")
    @ResponseStatus(HttpStatus.NO_CONTENT)
    public Mono<Void> deleteUser(@PathVariable String id) {
        return userService.deleteById(id);
    }
}

响应式返回类型

返回类型	说明
Mono<T>	返回单个对象
Flux<T>	返回多个对象（流式）
Mono<Void>	无返回值
Mono<ResponseEntity<T>>	自定义响应

WebClient - 响应式 HTTP 客户端

WebClient 是 WebFlux 提供的非阻塞 HTTP 客户端，用于替代 RestTemplate。

创建 WebClient

// 基本创建
WebClient client = WebClient.create();

// 带基础 URL
WebClient client = WebClient.create("http://localhost:8080");

// 使用 Builder 自定义
WebClient client = WebClient.builder()
    .baseUrl("http://localhost:8080")
    .defaultHeader(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE)
    .defaultHeader(HttpHeaders.AUTHORIZATION, "Bearer token")
    .filter(ExchangeFilterFunction.ofRequestProcessor(
        request -> {
            log.info("Request: {} {}", request.method(), request.url());
            return Mono.just(request);
        }
    ))
    .build();

GET 请求

// 获取单个对象
Mono<User> user = webClient.get()
    .uri("/api/users/{id}", userId)
    .retrieve()
    .bodyToMono(User.class);

// 获取列表
Flux<User> users = webClient.get()
    .uri("/api/users")
    .retrieve()
    .bodyToFlux(User.class);

// 带查询参数
Flux<User> users = webClient.get()
    .uri(uriBuilder -> uriBuilder
        .path("/api/users")
        .queryParam("page", 0)
        .queryParam("size", 10)
        .build())
    .retrieve()
    .bodyToFlux(User.class);

POST 请求

Mono<User> createdUser = webClient.post()
    .uri("/api/users")
    .contentType(MediaType.APPLICATION_JSON)
    .bodyValue(newUser)
    .retrieve()
    .bodyToMono(User.class);

// 使用 Mono 作为请求体
Mono<User> createdUser = webClient.post()
    .uri("/api/users")
    .body(userMono, User.class)
    .retrieve()
    .bodyToMono(User.class);

错误处理

Mono<User> user = webClient.get()
    .uri("/api/users/{id}", userId)
    .retrieve()
    .onStatus(HttpStatusCode::is4xxClientError, response -> 
        Mono.error(new NotFoundException("User not found")))
    .onStatus(HttpStatusCode::is5xxServerError, response -> 
        Mono.error(new ServerException("Server error")))
    .bodyToMono(User.class);

使用 exchange() 获取完整响应

Mono<ResponseEntity<User>> response = webClient.get()
    .uri("/api/users/{id}", userId)
    .exchangeToMono(clientResponse -> {
        if (clientResponse.statusCode().is2xxSuccessful()) {
            return clientResponse.toEntity(User.class);
        } else {
            return clientResponse.createException().flatMap(Mono::error);
        }
    });

ServerRequest 和 ServerResponse

这两个类用于函数式端点编程。

ServerRequest

// 获取路径变量
String id = request.pathVariable("id");

// 获取查询参数
Optional<String> name = request.queryParam("name");

// 获取请求头
String contentType = request.headers().contentType()
    .map(MediaType::toString)
    .orElse("unknown");

// 获取请求体
Mono<User> user = request.bodyToMono(User.class);
Flux<User> users = request.bodyToFlux(User.class);

ServerResponse

// 成功响应
ServerResponse.ok()
    .contentType(MediaType.APPLICATION_JSON)
    .bodyValue(user);

// 返回 Mono
ServerResponse.ok().body(userMono, User.class);

// 返回 Flux
ServerResponse.ok().body(userFlux, User.class);

// 创建响应
ServerResponse.created(URI.create("/api/users/" + id))
    .bodyValue(user);

// 无内容响应
ServerResponse.noContent().build();

// 错误响应
ServerResponse.notFound().build();
ServerResponse.badRequest().bodyValue("Invalid request");

异常处理

使用 @ExceptionHandler

@RestControllerAdvice
public class GlobalExceptionHandler {
    
    @ExceptionHandler(NotFoundException.class)
    @ResponseStatus(HttpStatus.NOT_FOUND)
    public Mono<ErrorResponse> handleNotFound(NotFoundException ex) {
        return Mono.just(new ErrorResponse(ex.getMessage()));
    }
    
    @ExceptionHandler(Exception.class)
    @ResponseStatus(HttpStatus.INTERNAL_SERVER_ERROR)
    public Mono<ErrorResponse> handleException(Exception ex) {
        return Mono.just(new ErrorResponse("Internal server error"));
    }
}

使用 WebExceptionHandler

@Component
@Order(-2)
public class GlobalWebExceptionHandler implements WebExceptionHandler {
    
    @Override
    public Mono<Void> handle(ServerWebExchange exchange, Throwable ex) {
        if (ex instanceof NotFoundException) {
            exchange.getResponse().setStatusCode(HttpStatus.NOT_FOUND);
            // 写入响应体...
        }
        return Mono.error(ex);
    }
}

过滤器（WebFilter）

@Component
public class LoggingFilter implements WebFilter {
    
    @Override
    public Mono<Void> filter(ServerWebExchange exchange, WebFilterChain chain) {
        long startTime = System.currentTimeMillis();
        
        return chain.filter(exchange)
            .doFinally(signalType -> {
                long duration = System.currentTimeMillis() - startTime;
                log.info("Request {} {} completed in {}ms",
                    exchange.getRequest().getMethod(),
                    exchange.getRequest().getPath(),
                    duration);
            });
    }
}

相关链接

• 上一篇：Reactor核心API
• 下一篇：函数式端点