万字长文：Spring WebFlux + Project Reactor 全栈指南——构建响应式Java云原生应用

// 创建一个 Flux
Flux<String> flux = Flux.just("A", "B", "C");

// 创建一个 Mono
Mono<String> mono = Mono.just("Hello");

Flux<Integer> numbers = Flux.range(1, 5)
    .map(i -> i * 2) // 声明一个操作
    .filter(i -> i > 5); // 声明另一个操作

// 此时没有任何计算发生

numbers.subscribe(System.out::println); // 订阅后，流水线才开始工作
// 输出: 6, 8, 10

Flux.just(1, 2, 3).map(i -> i * i).subscribe(System.out::println); // 1, 4, 9

Flux.just(1, 2, 3, 4).filter(i -> i % 2 == 0).subscribe(System.out::println); // 2, 4

// 模拟一个异步服务调用
Mono<String> getUserById(Long id) {
    return Mono.just("User-" + id);
}

Flux<Long> userIds = Flux.just(1L, 2L, 3L);
// 对每个ID发起异步调用，并合并结果
userIds.flatMap(this::getUserById).subscribe(System.out::println);
// 输出: User-1, User-2, User-3 (顺序可能不固定)

Mono.error(new RuntimeException("Oops!"))
    .onErrorReturn("Fallback Value")
    .subscribe(System.out::println); // 输出: Fallback Value

Mono.error(new RuntimeException("Primary failed!"))
    .onErrorResume(e -> Mono.just("Recovered from: " + e.getMessage()))
    .subscribe(System.out::println);

@RestController
@RequestMapping("/api/users")
public class UserController {

    private final UserService userService; // 假设这是一个响应式服务

    public UserController(UserService userService) {
        this.userService = userService;
    }

    @GetMapping("/{id}")
    public Mono<User> getUser(@PathVariable Long id) {
        return userService.findById(id);
    }

    @GetMapping
    public Flux<User> getAllUsers() {
        return userService.findAll();
    }

    @PostMapping
    public Mono<User> createUser(@RequestBody Mono<User> userMono) {
        return userMono.flatMap(userService::save);
    }
}

@Configuration
public class UserRoutes {

    @Bean
    public RouterFunction<ServerResponse> userRoutes(UserHandler userHandler) {
        return route(GET("/api/users/{id}"), userHandler::getUser)
                .andRoute(GET("/api/users"), userHandler::getAllUsers)
                .andRoute(POST("/api/users"), userHandler::createUser);
    }
}

@Component
public class UserHandler {

    private final UserService userService;

    public UserHandler(UserService userService) {
        this.userService = userService;
    }

    public Mono<ServerResponse> getUser(ServerRequest request) {
        Long id = Long.valueOf(request.pathVariable("id"));
        return userService.findById(id)
                .flatMap(user -> ServerResponse.ok().bodyValue(user))
                .switchIfEmpty(ServerResponse.notFound().build());
    }

    public Mono<ServerResponse> getAllUsers(ServerRequest request) {
        return ServerResponse.ok().body(userService.findAll(), User.class);
    }

    public Mono<ServerResponse> createUser(ServerRequest request) {
        Mono<User> user = request.bodyToMono(User.class);
        return user.flatMap(u -> ServerResponse.ok().body(userService.save(u)))
                   .switchIfEmpty(ServerResponse.badRequest().build());
    }
}

<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-webflux</artifactId>
    </dependency>
    <!-- 使用响应式 MongoDB -->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-data-mongodb-reactive</artifactId>
    </dependency>
</dependencies>

@Document
public class Post {
    @Id
    private String id;
    private String title;
    private String content;
    // constructors, getters, setters
}

public interface PostRepository extends ReactiveCrudRepository<Post, String> {
    Flux<Post> findByTitleContaining(String title);
}

@Service
public class PostService {
    private final PostRepository postRepository;

    public PostService(PostRepository postRepository) {
        this.postRepository = postRepository;
    }

    public Flux<Post> findAll() {
        return postRepository.findAll();
    }

    public Mono<Post> findById(String id) {
        return postRepository.findById(id);
    }

    public Mono<Post> save(Post post) {
        return postRepository.save(post);
    }

    public Flux<Post> searchByTitle(String keyword) {
        return postRepository.findByTitleContaining(keyword);
    }
}

@RestController
@RequestMapping("/posts")
public class PostController {
    private final PostService postService;

    public PostController(PostService postService) {
        this.postService = postService;
    }

    @GetMapping
    public Flux<Post> getAllPosts() {
        return postService.findAll();
    }

    @GetMapping("/{id}")
    public Mono<ResponseEntity<Post>> getPost(@PathVariable String id) {
        return postService.findById(id)
                .map(ResponseEntity::ok)
                .defaultIfEmpty(ResponseEntity.notFound().build());
    }

    @PostMapping
    public Mono<Post> createPost(@RequestBody Post post) {
        return postService.save(post);
    }

    @GetMapping("/search")
    public Flux<Post> searchPosts(@RequestParam String q) {
        return postService.searchByTitle(q);
    }
}

Flux.fromIterable(blockingService.getAllIds()) // 阻塞调用
    .publishOn(Schedulers.boundedElastic()) // 切换到弹性线程池
    .map(id -> blockingService.findById(id)) // 在新线程池中执行
    .subscribeOn(Schedulers.parallel()); // （可选）指定订阅发生的线程

@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
class PostControllerTest {

    @Autowired
    private WebTestClient webTestClient;

    @Test
    void shouldGetAllPosts() {
        webTestClient.get().uri("/posts")
            .exchange()
            .expectStatus().isOk()
            .expectBodyList(Post.class)
            .hasSize(2);
    }
}