Spring WebFlux案例

wen java案例 2026-08-10 169

本文目录导读：

Spring WebFlux案例

项目依赖配置 (pom.xml)
主应用类
实体类
数据传输对象 (DTO)
响应封装类
Mapper 工具类
异常处理类
Repository 接口
Service 层
Controller 层
配置类
数据初始化配置
测试类
响应式流操作示例
启动类配置
使用说明

我将为您提供一个完整的Spring WebFlux案例,涵盖响应式编程的核心概念和实际应用。

项目依赖配置 (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.1.5</version>
        <relativePath/>
    </parent>
    <groupId>com.example</groupId>
    <artifactId>webflux-demo</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <name>webflux-demo</name>
    <description>Spring WebFlux Demo Project</description>
    <properties>
        <java.version>17</java.version>
    </properties>
    <dependencies>
        <!-- WebFlux -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-webflux</artifactId>
        </dependency>
        <!-- MongoDB Reactive -->
        <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>
        <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>
            </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 WebFluxDemoApplication {
    public static void main(String[] args) {
        SpringApplication.run(WebFluxDemoApplication.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 javax.validation.constraints.Email;
import javax.validation.constraints.NotBlank;
import java.time.LocalDateTime;
@Data
@Builder
@NoArgsConstructor
@AllArgsConstructor
@Document(collection = "users")
public class User {
    @Id
    private String id;
    @NotBlank(message = "用户名不能为空")
    private String username;
    @Email(message = "邮箱格式不正确")
    @NotBlank(message = "邮箱不能为空")
    private String email;
    private String avatar;
    private Integer age;
    private LocalDateTime createdAt;
    private LocalDateTime updatedAt;
}

数据传输对象 (DTO)

package com.example.webflux.dto;
import lombok.*;
import javax.validation.constraints.Email;
import javax.validation.constraints.NotBlank;
import javax.validation.constraints.Size;
@Data
@Builder
@NoArgsConstructor
@AllArgsConstructor
public class UserDTO {
    private String id;
    @NotBlank(message = "用户名不能为空")
    @Size(min = 3, max = 20, message = "用户名长度必须在3-20之间")
    private String username;
    @Email(message = "邮箱格式不正确")
    @NotBlank(message = "邮箱不能为空")
    private String email;
    private String avatar;
    private Integer age;
    private String createdAt;
    private String updatedAt;
}

响应封装类

package com.example.webflux.dto;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
@Data
@Builder
@NoArgsConstructor
@AllArgsConstructor
public class ApiResponse<T> {
    private Integer code;
    private String message;
    private T data;
    public static <T> ApiResponse<T> success(T data) {
        return ApiResponse.<T>builder()
                .code(200)
                .message("success")
                .data(data)
                .build();
    }
    public static <T> ApiResponse<T> error(Integer code, String message) {
        return ApiResponse.<T>builder()
                .code(code)
                .message(message)
                .build();
    }
}

Mapper 工具类

package com.example.webflux.mapper;
import com.example.webflux.dto.UserDTO;
import com.example.webflux.model.User;
public class UserMapper {
    public static UserDTO toDTO(User user) {
        if (user == null) {
            return null;
        }
        return UserDTO.builder()
                .id(user.getId())
                .username(user.getUsername())
                .email(user.getEmail())
                .avatar(user.getAvatar())
                .age(user.getAge())
                .createdAt(user.getCreatedAt() != null ? 
                    user.getCreatedAt().toString() : null)
                .updatedAt(user.getUpdatedAt() != null ? 
                    user.getUpdatedAt().toString() : null)
                .build();
    }
    public static User toEntity(UserDTO dto) {
        if (dto == null) {
            return null;
        }
        return User.builder()
                .id(dto.getId())
                .username(dto.getUsername())
                .email(dto.getEmail())
                .avatar(dto.getAvatar())
                .age(dto.getAge())
                .build();
    }
}

异常处理类

package com.example.webflux.exception;
import lombok.extern.slf4j.Slf4j;
import org.springframework.http.HttpStatus;
import org.springframework.web.bind.annotation.ExceptionHandler;
import org.springframework.web.bind.annotation.ResponseStatus;
import org.springframework.web.bind.annotation.RestControllerAdvice;
import org.springframework.web.bind.support.WebExchangeBindException;
import com.example.webflux.dto.ApiResponse;
@Slf4j
@RestControllerAdvice
public class GlobalExceptionHandler {
    @ExceptionHandler(WebExchangeBindException.class)
    @ResponseStatus(HttpStatus.BAD_REQUEST)
    public ApiResponse<Void> handleValidationException(WebExchangeBindException e) {
        String message = e.getBindingResult()
                .getFieldErrors()
                .stream()
                .map(error -> error.getField() + ": " + error.getDefaultMessage())
                .reduce((a, b) -> a + "; " + b)
                .orElse("参数错误");
        log.error("参数校验失败: {}", message);
        return ApiResponse.error(400, message);
    }
    @ExceptionHandler(UserNotFoundException.class)
    @ResponseStatus(HttpStatus.NOT_FOUND)
    public ApiResponse<Void> handleUserNotFoundException(UserNotFoundException e) {
        log.error("用户不存在: {}", e.getMessage());
        return ApiResponse.error(404, e.getMessage());
    }
    @ExceptionHandler(Exception.class)
    @ResponseStatus(HttpStatus.INTERNAL_SERVER_ERROR)
    public ApiResponse<Void> handleException(Exception e) {
        log.error("系统异常", e);
        return ApiResponse.error(500, "系统内部错误");
    }
}
// 自定义异常类
package com.example.webflux.exception;
public class UserNotFoundException extends RuntimeException {
    public UserNotFoundException(String message) {
        super(message);
    }
    public UserNotFoundException(String message, Throwable cause) {
        super(message, cause);
    }
}

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(Integer age);
    Mono<Boolean> existsByUsername(String username);
    Mono<Boolean> existsByEmail(String email);
}

Service 层

package com.example.webflux.service;
import com.example.webflux.dto.UserDTO;
import com.example.webflux.exception.UserNotFoundException;
import com.example.webflux.mapper.UserMapper;
import com.example.webflux.model.User;
import com.example.webflux.repository.UserRepository;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import java.time.LocalDateTime;
@Slf4j
@Service
@RequiredArgsConstructor
public class UserService {
    private final UserRepository userRepository;
    public Mono<UserDTO> createUser(UserDTO userDTO) {
        return validateUser(userDTO)
                .flatMap(dto -> {
                    User user = UserMapper.toEntity(dto);
                    user.setCreatedAt(LocalDateTime.now());
                    user.setUpdatedAt(LocalDateTime.now());
                    return userRepository.save(user)
                            .map(UserMapper::toDTO)
                            .doOnSuccess(saved -> 
                                log.info("用户创建成功: {}", saved.getUsername()));
                });
    }
    public Mono<UserDTO> getUserById(String id) {
        return userRepository.findById(id)
                .switchIfEmpty(Mono.error(
                    new UserNotFoundException("用户不存在: " + id)))
                .map(UserMapper::toDTO);
    }
    public Mono<UserDTO> getByUsername(String username) {
        return userRepository.findByUsername(username)
                .switchIfEmpty(Mono.error(
                    new UserNotFoundException("用户不存在: " + username)))
                .map(UserMapper::toDTO);
    }
    public Flux<UserDTO> getAllUsers() {
        return userRepository.findAll()
                .map(UserMapper::toDTO);
    }
    public Mono<UserDTO> updateUser(String id, UserDTO userDTO) {
        return userRepository.findById(id)
                .switchIfEmpty(Mono.error(
                    new UserNotFoundException("用户不存在: " + id)))
                .flatMap(user -> {
                    user.setUsername(userDTO.getUsername());
                    user.setEmail(userDTO.getEmail());
                    user.setAvatar(userDTO.getAvatar());
                    user.setAge(userDTO.getAge());
                    user.setUpdatedAt(LocalDateTime.now());
                    return userRepository.save(user)
                            .map(UserMapper::toDTO)
                            .doOnSuccess(updated -> 
                                log.info("用户更新成功: {}", updated.getUsername()));
                });
    }
    public Mono<Void> deleteUser(String id) {
        return userRepository.findById(id)
                .switchIfEmpty(Mono.error(
                    new UserNotFoundException("用户不存在: " + id)))
                .flatMap(user -> userRepository.delete(user)
                    .doOnSuccess(v -> 
                        log.info("用户删除成功: {}", user.getUsername())));
    }
    public Mono<Long> countUsers() {
        return userRepository.count();
    }
    public Flux<UserDTO> getUsersByAgeGreaterThan(Integer age) {
        return userRepository.findByAgeGreaterThan(age)
                .map(UserMapper::toDTO);
    }
    private Mono<UserDTO> validateUser(UserDTO userDTO) {
        return userRepository.existsByUsername(userDTO.getUsername())
                .flatMap(exists -> {
                    if (exists) {
                        return Mono.error(new RuntimeException(
                            "用户名已存在: " + userDTO.getUsername()));
                    }
                    return userRepository.existsByEmail(userDTO.getEmail());
                })
                .flatMap(exists -> {
                    if (exists) {
                        return Mono.error(new RuntimeException(
                            "邮箱已被注册: " + userDTO.getEmail()));
                    }
                    return Mono.just(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 lombok.RequiredArgsConstructor;
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;
@RestController
@RequestMapping("/api/users")
@RequiredArgsConstructor
public class UserController {
    private final UserService userService;
    @PostMapping
    @ResponseStatus(HttpStatus.CREATED)
    public Mono<ApiResponse<UserDTO>> createUser(@Valid @RequestBody UserDTO userDTO) {
        return userService.createUser(userDTO)
                .map(ApiResponse::success);
    }
    @GetMapping("/{id}")
    public Mono<ResponseEntity<ApiResponse<UserDTO>>> getUserById(@PathVariable String id) {
        return userService.getUserById(id)
                .map(user -> ResponseEntity.ok(ApiResponse.success(user)))
                .defaultIfEmpty(ResponseEntity.notFound().build());
    }
    @GetMapping("/username/{username}")
    public Mono<ResponseEntity<ApiResponse<UserDTO>>> getByUsername(
            @PathVariable String username) {
        return userService.getByUsername(username)
                .map(user -> ResponseEntity.ok(ApiResponse.success(user)))
                .defaultIfEmpty(ResponseEntity.notFound().build());
    }
    @GetMapping
    public Mono<ApiResponse<Flux<UserDTO>>> getAllUsers() {
        return Mono.just(ApiResponse.success(userService.getAllUsers()));
    }
    @PutMapping("/{id}")
    public Mono<ResponseEntity<ApiResponse<UserDTO>>> updateUser(
            @PathVariable String id, 
            @Valid @RequestBody UserDTO userDTO) {
        return userService.updateUser(id, userDTO)
                .map(user -> ResponseEntity.ok(ApiResponse.success(user)))
                .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("/count")
    public Mono<ApiResponse<Long>> countUsers() {
        return userService.countUsers()
                .map(ApiResponse::success);
    }
    @GetMapping("/age/{age}")
    public Mono<ApiResponse<Flux<UserDTO>>> getUsersByAge(@PathVariable Integer age) {
        return Mono.just(ApiResponse.success(
            userService.getUsersByAgeGreaterThan(age)));
    }
}

配置类

package com.example.webflux.config;
import org.springframework.boot.autoconfigure.web.WebProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.mongodb.config.EnableReactiveMongoAuditing;
@Configuration
@EnableReactiveMongoAuditing
public class WebFluxConfig {
    @Bean
    public WebProperties.Resources resources() {
        return new WebProperties.Resources();
    }
}

数据初始化配置

package com.example.webflux.config;
import com.example.webflux.model.User;
import com.example.webflux.repository.UserRepository;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.CommandLineRunner;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.stereotype.Component;
import reactor.core.publisher.Flux;
import java.time.LocalDateTime;
import java.util.Arrays;
@Slf4j
@Configuration
@RequiredArgsConstructor
public class DataInitializer {
    private final UserRepository userRepository;
    @Bean
    public CommandLineRunner initData() {
        return args -> {
            userRepository.deleteAll()
                .thenMany(
                    Flux.just(
                        User.builder()
                            .username("admin")
                            .email("admin@example.com")
                            .age(30)
                            .createdAt(LocalDateTime.now())
                            .updatedAt(LocalDateTime.now())
                            .build(),
                        User.builder()
                            .username("user1")
                            .email("user1@example.com")
                            .age(25)
                            .createdAt(LocalDateTime.now())
                            .updatedAt(LocalDateTime.now())
                            .build(),
                        User.builder()
                            .username("user2")
                            .email("user2@example.com")
                            .age(35)
                            .createdAt(LocalDateTime.now())
                            .updatedAt(LocalDateTime.now())
                            .build()
                    )
                )
                .flatMap(userRepository::save)
                .subscribe(
                    user -> log.info("初始化用户: {}", user.getUsername()),
                    error -> log.error("初始化数据失败: {}", error.getMessage()),
                    () -> log.info("数据初始化完成")
                );
        };
    }
}

测试类

package com.example.webflux.controller;
import com.example.webflux.dto.UserDTO;
import com.example.webflux.service.UserService;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.autoconfigure.web.reactive.WebFluxTest;
import org.springframework.boot.test.mock.mockito.MockBean;
import org.springframework.http.MediaType;
import org.springframework.test.web.reactive.server.WebTestClient;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.when;
@WebFluxTest(UserController.class)
public class UserControllerTest {
    @Autowired
    private WebTestClient webTestClient;
    @MockBean
    private UserService userService;
    @Autowired
    private ObjectMapper objectMapper;
    @Test
    public void testGetUserById() {
        String userId = "123";
        UserDTO userDTO = UserDTO.builder()
                .id(userId)
                .username("testUser")
                .email("test@example.com")
                .build();
        when(userService.getUserById(userId))
                .thenReturn(Mono.just(userDTO));
        webTestClient.get()
                .uri("/api/users/{id}", userId)
                .exchange()
                .expectStatus().isOk()
                .expectBody()
                .jsonPath("$.code").isEqualTo(200)
                .jsonPath("$.data.username").isEqualTo("testUser");
    }
    @Test
    public void testGetAllUsers() {
        UserDTO user1 = UserDTO.builder()
                .id("1")
                .username("user1")
                .email("user1@example.com")
                .build();
        UserDTO user2 = UserDTO.builder()
                .id("2")
                .username("user2")
                .email("user2@example.com")
                .build();
        when(userService.getAllUsers())
                .thenReturn(Flux.just(user1, user2));
        webTestClient.get()
                .uri("/api/users")
                .accept(MediaType.APPLICATION_JSON)
                .exchange()
                .expectStatus().isOk()
                .expectBody()
                .jsonPath("$.code").isEqualTo(200)
                .jsonPath("$.data").isArray()
                .jsonPath("$.data.length()").isEqualTo(2);
    }
    @Test
    public void testCreateUser() {
        UserDTO request = UserDTO.builder()
                .username("newUser")
                .email("new@example.com")
                .age(25)
                .build();
        UserDTO response = UserDTO.builder()
                .id("generate-id")
                .username("newUser")
                .email("new@example.com")
                .age(25)
                .build();
        when(userService.createUser(any(UserDTO.class)))
                .thenReturn(Mono.just(response));
        webTestClient.post()
                .uri("/api/users")
                .contentType(MediaType.APPLICATION_JSON)
                .body(Mono.just(request), UserDTO.class)
                .exchange()
                .expectStatus().isCreated()
                .expectBody()
                .jsonPath("$.code").isEqualTo(200)
                .jsonPath("$.data.username").isEqualTo("newUser");
    }
    @Test
    public void testDeleteUser() {
        String userId = "123";
        when(userService.deleteUser(userId))
                .thenReturn(Mono.empty());
        webTestClient.delete()
                .uri("/api/users/{id}", userId)
                .exchange()
                .expectStatus().isNoContent();
    }
}

响应式流操作示例

package com.example.webflux.example;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;
import org.springframework.stereotype.Component;
import lombok.extern.slf4j.Slf4j;
@Slf4j
@Component
public class ReactiveOperations {
    // 基本操作
    public void basicOperations() {
        // 创建 Flux
        Flux<String> colors = Flux.just("red", "green", "blue");
        // 创建 Mono
        Mono<String> single = Mono.just("Single value");
        // 过滤和转换
        Flux<Integer> numbers = Flux.range(1, 10);
        numbers
            .filter(n -> n % 2 == 0)      // 过滤偶数
            .map(n -> n * n)               // 平方
            .flatMap(n -> Flux.just(n, n)) // 扁平化
            .doOnNext(n -> log.info("处理: {}", n))
            .subscribe();
    }
    // 错误处理
    public void errorHandling() {
        Flux<String> flux = Flux.just("element");
        flux
            .map(value -> {
                if (value.equals("error")) {
                    throw new RuntimeException("发生错误");
                }
                return value.toUpperCase();
            })
            .onErrorResume(error -> Flux.just("fallback1", "fallback2"))
            .onErrorReturn("default")
            .retry(3)
            .timeout(Mono.delay(java.time.Duration.ofSeconds(1)))
            .onErrorReturn("timeout fallback")
            .subscribe(
                value -> log.info("得到: {}", value),
                error -> log.error("错误: {}", error.getMessage())
            );
    }
    // 背压处理
    public void backpressureHandling() {
        Flux.range(1, 100)
            .limitRate(10)  // 每次最多处理10个
            .buffer(5)      // 收集5个元素为一个列表
            .doOnNext(list -> log.info("处理批次: {}", list))
            .subscribe();
    }
    // 并行处理
    public void parallelProcessing() {
        Flux.range(1, 10)
            .parallel()
            .runOn(Schedulers.parallel())
            .map(i -> i * 2)
            .sequential()
            .subscribe(result -> log.info("并行结果: {}", result));
    }
    // 组合操作
    public void combineOperations() {
        Flux<String> flux1 = Flux.just("A", "B", "C");
        Flux<String> flux2 = Flux.just("1", "2");
        // zip 组合
        flux1.zipWith(flux2, (a, b) -> a + b)
            .subscribe(result -> log.info("Zip: {}", result));
        // 合并
        flux1.mergeWith(flux2)
            .subscribe(result -> log.info("Merge: {}", result));
        // 连接
        flux1.concatWith(flux2)
            .subscribe(result -> log.info("Concat: {}", result));
    }
}

启动类配置

# application.yml
spring:
  application:
    name: webflux-demo
  data:
    mongodb:
      uri: mongodb://localhost:27017/webflux_demo
  webflux:
    base-path: /api
server:
  port: 8080
  netty:
    max-initial-line-length: 4096
logging:
  level:
    root: INFO
    com.example.webflux: DEBUG
    org.springframework.data.mongodb: DEBUG
    io.netty: WARN
management:
  endpoints:
    web:
      exposure:
        include: health,info,metrics

使用说明

环境要求：
- JDK 17+
- MongoDB 4.0+
- Maven 3.6+

运行步骤：

# 启动MongoDB
docker run -d -p 27017:27017 mongo
# 构建项目
mvn clean package
# 运行应用
mvn spring-boot:run

测试API：

# 创建用户
curl -X POST http://localhost:8080/api/users \
  -H "Content-Type: application/json" \
  -d '{"username":"test","email":"test@example.com","age":25}'
# 获取所有用户
curl http://localhost:8080/api/users
# 获取特定用户
curl http://localhost:8080/api/users/{id}

这个案例涵盖了Spring WebFlux的主要特性：

响应式编程模型
非阻塞数据流
MongoDB集成
流式处理
背压机制
WebFlux测试

本文地址： https://bmjup.com/post/42507.html