Spring Boot 3.x 响应式编程全攻略：WebFlux 实战

响应式编程概述

什么是响应式编程？

响应式编程是一种异步编程范式，基于数据流和变化传播。它允许你构建非阻塞的、高并发的应用程序。

核心概念

数据流（Data Stream）：

• 数据像水流一样流动
• 可以被观察、转换、过滤
• 支持异步处理

背压（Backpressure）：

• 生产者生产速度 > 消费者消费速度
• 需要机制来控制流量
• 避免内存溢出

Project Reactor

核心 API

1. Mono（0-1 个元素）

import reactor.core.publisher.Mono;

// 创建 Mono
Mono<String> mono1 = Mono.just("Hello");
Mono<String> mono2 = Mono.empty();
Mono<String> mono3 = Mono.error(new RuntimeException("Error"));

// 订阅
mono1.subscribe(
    value -> System.out.println("Value: " + value),
    error -> error.printStackTrace(),
    () -> System.out.println("Completed")
);
​

2. Flux（0-N 个元素）

import reactor.core.publisher.Flux;

// 创建 Flux
Flux<Integer> flux1 = Flux.just(1, 2, 3, 4, 5);
Flux<String> flux2 = Flux.fromIterable(Arrays.asList("a", "b", "c"));
Flux<Long> flux3 = Flux.interval(Duration.ofMillis(100));

// 订阅
flux1.subscribe(
    value -> System.out.println("Value: " + value),
    error -> error.printStackTrace(),
    () -> System.out.println("Completed")
);
​

操作符

转换操作符

// map：一对一转换
Flux<Integer> numbers = Flux.range(1, 5);
Flux<Integer> squared = numbers.map(x -> x * x);

// flatMap：一对多转换
Flux<String> words = Flux.just("Hello", "World");
Flux<String> letters = words.flatMap(word -> 
    Flux.fromArray(word.split(""))
);

// filter：过滤
Flux<Integer> evens = numbers.filter(x -> x % 2 == 0);
​

组合操作符

// zip：组合多个流
Flux<Integer> flux1 = Flux.just(1, 2, 3);
Flux<String> flux2 = Flux.just("a", "b", "c");
Flux<Tuple2<Integer, String>> zipped = Flux.zip(flux1, flux2);

// merge：合并流（保持顺序）
Flux<Integer> merged = Flux.merge(flux1, flux2);

// concat：连接流（顺序执行）
Flux<Integer> concatenated = Flux.concat(flux1, flux2);
​

错误处理操作符

// onErrorReturn：返回默认值
Flux<String> flux = Flux.just("a", "b", "c")
    .concatWith(Flux.error(new RuntimeException("Error")))
    .onErrorReturn("default");

// onErrorResume：恢复错误
Flux<String> recovered = flux.onErrorResume(error -> {
    return Flux.just("recovered1", "recovered2");
});

// retry：重试
Flux<String> retried = flux.retry(3);
​

Spring WebFlux

创建响应式 Controller

import org.springframework.web.bind.annotation.*;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

@RestController
@RequestMapping("/api/users")
public class UserController {
    
    private final UserRepository userRepository;
    
    // 获取所有用户
    @GetMapping
    public Flux<User> getAllUsers() {
        return userRepository.findAll();
    }
    
    // 获取单个用户
    @GetMapping("/{id}")
    public Mono<User> getUserById(@PathVariable Long id) {
        return userRepository.findById(id)
            .switchIfEmpty(Mono.error(new NotFoundException("User not found")));
    }
    
    // 创建用户
    @PostMapping
    public Mono<User> createUser(@RequestBody User user) {
        return userRepository.save(user);
    }
    
    // 更新用户
    @PutMapping("/{id}")
    public Mono<User> updateUser(@PathVariable Long id, @RequestBody User user) {
        return userRepository.findById(id)
            .flatMap(existingUser -> {
                existingUser.setName(user.getName());
                existingUser.setEmail(user.getEmail());
                return userRepository.save(existingUser);
            });
    }
    
    // 删除用户
    @DeleteMapping("/{id}")
    public Mono<Void> deleteUser(@PathVariable Long id) {
        return userRepository.deleteById(id);
    }
}
​

响应式 Service

@Service
public class UserService {
    
    private final UserRepository userRepository;
    private final EmailService emailService;
    
    public Mono<User> getUserWithPosts(Long userId) {
        return userRepository.findById(userId)
            .flatMap(user -> {
                Mono<List<Post>> postsMono = postService
                    .findByUserId(userId)
                    .collectList();
                
                return postsMono.map(posts -> {
                    user.setPosts(posts);
                    return user;
                });
            });
    }
    
    public Mono<Void> registerAndSendEmail(User user) {
        return userRepository.save(user)
            .flatMap(savedUser -> 
                emailService.sendWelcomeEmail(savedUser.getEmail())
                    .then(Mono.just(savedUser))
            )
            .then();
    }
}
​

响应式数据访问

Spring Data R2DBC

配置

# application.yml
spring:
  r2dbc:
    url: r2dbc:postgresql://localhost:5432/mydb
    username: user
    password: password
    pool:
      initial-size: 5
      max-size: 20
​

Repository

import org.springframework.data.r2dbc.repository.R2dbcRepository;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

public interface UserRepository extends R2dbcRepository<User, Long> {
    
    Flux<User> findByEmail(String email);
    
    Flux<User> findByAgeGreaterThan(int age);
    
    @Query("SELECT * FROM users WHERE name LIKE :pattern")
    Flux<User> findByNameLike(@Param("pattern") String pattern);
}
​

自定义查询

@Service
public class CustomUserService {
    
    private final DatabaseClient databaseClient;
    
    public Flux<User> findUsersWithPosts() {
        return databaseClient.sql("""
            SELECT u.*, p.id as post_id, p.title as post_title
            FROM users u
            LEFT JOIN posts p ON u.id = p.user_id
            ORDER BY u.id
            """)
            .fetch()
            .all()
            .flatMap(row -> {
                User user = mapRowToUser(row);
                // 处理关联的 posts
                return Mono.just(user);
            });
    }
    
    public Mono<User> findByIdWithPosts(Long userId) {
        return databaseClient.sql("""
            SELECT u.*, p.id as post_id, p.title as post_title
            FROM users u
            LEFT JOIN posts p ON u.id = p.user_id
            WHERE u.id = :userId
            """)
            .parameter("userId", userId)
            .fetch()
            .all()
            .collectList()
            .map(rows -> {
                // 聚合结果
                User user = null;
                List<Post> posts = new ArrayList<>();
                
                for (Row row : rows) {
                    if (user == null) {
                        user = mapRowToUser(row);
                    }
                    if (row.get("post_id") != null) {
                        posts.add(mapRowToPost(row));
                    }
                }
                
                if (user != null) {
                    user.setPosts(posts);
                }
                
                return user;
            });
    }
}
​

响应式安全

Spring Security 配置

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.security.config.annotation.web.reactive.EnableWebFluxSecurity;
import org.springframework.security.config.web.server.ServerHttpSecurity;
import org.springframework.security.web.server.SecurityWebFilterChain;

@Configuration
@EnableWebFluxSecurity
public class SecurityConfig {
    
    @Bean
    public SecurityWebFilterChain securityWebFilterChain(ServerHttpSecurity http) {
        return http
            .authorizeExchange(exchanges -> exchanges
                .pathMatchers("/api/public/**").permitAll()
                .pathMatchers("/api/admin/**").hasRole("ADMIN")
                .anyExchange().authenticated()
            )
            .oauth2ResourceServer(ServerHttpSecurity.OAuth2ResourceServerSpec::jwt)
            .csrf().disable()
            .build();
    }
}
​

JWT 认证

@Service
public class TokenService {
    
    private final JwtEncoder encoder;
    
    public Mono<String> generateToken(Authentication authentication) {
        return Mono.fromCallable(() -> {
            User user = (User) authentication.getPrincipal();
            
            JwtClaimsSet claims = JwtClaimsSet.builder()
                .issuer("myapp")
                .issuedAt(Instant.now())
                .expiresAt(Instant.now().plus(1, ChronoUnit.HOURS))
                .subject(user.getUsername())
                .claim("scope", user.getAuthorities())
                .build();
            
            return this.encoder.encode(JwtEncoderParameters.from(claims)).getTokenValue();
        });
    }
}
​

函数式路由

Router Function

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.web.reactive.function.server.RouterFunction;
import org.springframework.web.reactive.function.server.ServerResponse;
import static org.springframework.web.reactive.function.server.RouterFunctions.route;
import static org.springframework.web.reactive.function.server.RequestPredicates.*;

@Configuration
public class RouterConfig {
    
    @Bean
    public RouterFunction<ServerResponse> userRouter(UserHandler userHandler) {
        return route()
            .GET("/api/users", userHandler::getAllUsers)
            .GET("/api/users/{id}", userHandler::getUserById)
            .POST("/api/users", userHandler::createUser)
            .PUT("/api/users/{id}", userHandler::updateUser)
            .DELETE("/api/users/{id}", userHandler::deleteUser)
            .build();
    }
}
​

Handler Function

@Component
public class UserHandler {
    
    private final UserRepository userRepository;
    
    public Mono<ServerResponse> getAllUsers(ServerRequest request) {
        return ServerResponse.ok()
            .body(userRepository.findAll(), User.class);
    }
    
    public Mono<ServerResponse> getUserById(ServerRequest request) {
        Long id = Long.parseLong(request.pathVariable("id"));
        
        return userRepository.findById(id)
            .flatMap(user -> ServerResponse.ok().bodyValue(user))
            .switchIfEmpty(ServerResponse.notFound().build());
    }
    
    public Mono<ServerResponse> createUser(ServerRequest request) {
        return request.bodyToMono(User.class)
            .flatMap(userRepository::save)
            .flatMap(user -> ServerResponse.status(HttpStatus.CREATED).bodyValue(user));
    }
}
​

错误处理

全局异常处理

@Configuration
public class GlobalErrorConfig {
    
    @Bean
    public WebExceptionHandler exceptionHandler() {
        return (ServerWebExchange exchange, Throwable ex) -> {
            ServerHttpResponse response = exchange.getResponse();
            
            if (ex instanceof NotFoundException) {
                response.setStatusCode(HttpStatus.NOT_FOUND);
            } else if (ex instanceof BadRequestException) {
                response.setStatusCode(HttpStatus.BAD_REQUEST);
            } else {
                response.setStatusCode(HttpStatus.INTERNAL_SERVER_ERROR);
            }
            
            return response.writeWith(
                Mono.just(new ErrorResponse(ex.getMessage()))
                    .flatMap(body -> {
                        DataBuffer buffer = response.bufferFactory().wrap(body);
                        return response.writeWith(Mono.just(buffer));
                    })
            );
        };
    }
}

record ErrorResponse(String error, int status) {}
​

测试

WebTestClient

@SpringBootTest
@AutoConfigureWebTestClient
class UserControllerTest {
    
    @Autowired
    private WebTestClient webTestClient;
    
    @Test
    void getAllUsers() {
        webTestClient.get()
            .uri("/api/users")
            .exchange()
            .expectStatus().isOk()
            .expectBodyList(User.class)
            .hasSize(3);
    }
    
    @Test
    void getUserById() {
        webTestClient.get()
            .uri("/api/users/1")
            .exchange()
            .expectStatus().isOk()
            .expectBody(User.class)
            .value(User::getName, equalTo("John Doe"));
    }
}
​

性能优化

1. 背压处理

Flux<Data> stream = dataSource.getData()
    .onBackpressureBuffer(1000)  // 缓冲 1000 个元素
    .onBackpressureDrop()         // 丢弃无法处理的元素
    .doOnBackpressureBuffer(dropped -> 
        System.out.println("Dropped: " + dropped)
    );
​

2. 调度器

// 在 I/O 调度器上执行
Flux.interval(Duration.ofMillis(100))
    .subscribeOn(Schedulers.boundedElastic());

// 在并行调度器上执行
Flux.range(1, 1000)
    .parallel()
    .runOn(Schedulers.parallel())
    .subscribe();
​

3. 缓存

Flux<User> users = userRepository.findAll()
    .cache();  // 缓存结果

Mono<User> user = userRepository.findById(1)
    .cache(Duration.ofMinutes(10));  // 缓存 10 分钟
​

最佳实践

1. 避免阻塞：不要在响应式流中使用阻塞操作
2. 错误处理：正确处理错误和异常
3. 资源管理：使用 using 或 doFinally 清理资源
4. 背压处理：合理配置背压策略
5. 测试覆盖：编写全面的响应式测试

总结

Spring WebFlux 提供了强大的响应式编程能力：

1. 核心概念：理解 Flux/Mono 和操作符
2. 数据访问：使用 R2DBC 进行非阻塞数据库操作
3. 安全配置：集成 Spring Security
4. 函数式路由：使用 RouterFunction
5. 性能优化：背压、调度器、缓存

掌握响应式编程，构建高并发应用！