附录 D：从 CompletableFuture 到响应式流迁移指南

CompletableFuture<User> userFuture = fetchUser(id)
    .thenCompose(user -> fetchOrders(user.getAccountId()))
    .thenApply(orders -> calculateTotal(orders))
    .exceptionally(e -> {
        log.error("Failed", e);
        return User.empty();
    });

Mono<User> userMono = fetchUser(id)
    .flatMap(user -> fetchOrders(user.getAccountId()))
    .map(orders -> calculateTotal(orders))
    .onErrorResume(e -> {
        log.error("Failed", e);
        return Mono.just(User.empty());
    });

CompletableFuture<User> f1 = fetchUser(1);
CompletableFuture<User> f2 = fetchUser(2);
CompletableFuture<User> f3 = fetchUser(3);

CompletableFuture<List<User>> usersFuture = CompletableFuture.allOf(f1, f2, f3)
    .thenApply(v -> List.of(f1.join(), f2.join(), f3.join()));

Mono<List<User>> usersMono = Mono.zip(
    fetchUser(1), fetchUser(2), fetchUser(3),
    (u1, u2, u3) -> List.of(u1, u2, u3)
);

fetchUser(id).orTimeout(Duration.ofSeconds(5));

fetchUser(id).timeout(Duration.ofSeconds(5));

// 错误：阻塞了响应式线程
User user = userMono.block();  // 等同于 future.get()！

// 正确：在订阅中处理
userMono.subscribe(u -> process(u));

// 测试中用 StepVerifier
StepVerifier.create(userMono)
    .expectNext(expectedUser)
    .verifyComplete();

// CompletableFuture：异常可能被 CompletionException 包装
future.exceptionally(e -> {
    Throwable cause = (e instanceof CompletionException && e.getCause() != null)
        ? e.getCause()
        : e;
    return fallback(cause);
});

// Mono：onErrorResume 收到的是操作符抛出的原始异常
mono.onErrorResume(e -> Mono.just(fallback(e)));

// CompletableFuture：没有背压概念
for (int i = 0; i < 1000000; i++) {
    fetchUser(i);  // 100 万个请求同时发出！
}

// Mono：用 flatMap 的 concurrency 控制
Flux.range(0, 1000000)
    .flatMap(i -> fetchUser(i), 100)  // 最多 100 个并发
    .subscribe();

// CompletableFuture 默认用 ForkJoinPool
CompletableFuture.supplyAsync(() -> compute());

// Mono 默认在订阅线程执行，需要显式切换
Mono.fromCallable(() -> compute())
    .subscribeOn(Schedulers.boundedElastic());

// CompletableFuture → Mono
Mono<String> mono = Mono.fromFuture(completableFuture);

// Mono → CompletableFuture
CompletableFuture<String> future = mono.toFuture();

// Mono → Flux
Flux<String> flux = mono.flux();

// Flux → Mono（取第一个）
Mono<String> first = flux.next();

// 现有 CF 代码
public CompletableFuture<User> getUser(String id) {
    return userClient.get(id);
}

// 新代码使用 Mono，在边界处转换
public Mono<Order> getOrderWithUser(String orderId) {
    return orderRepository.findById(orderId)
        .flatMap(order -> Mono.fromFuture(getUser(order.getUserId()))
            .map(user -> {
                order.setUser(user);
                return order;
            }));
}

public Mono<PaymentResult> processPayment(PaymentRequest request) {
    return validate(request)
        .flatMap(valid -> Mono.fromFuture(
            legacyPaymentService.charge(request)  // CF
        ))
        .timeout(Duration.ofSeconds(10))
        .onErrorResume(e -> Mono.just(PaymentResult.failure(e)));
}

// Spring MVC（同步）
@GetMapping("/user/{id}")
public User getUser(@PathVariable String id) {
    return userService.findById(id);  // 阻塞！
}

// Spring WebFlux（异步）
@GetMapping("/user/{id}")
public Mono<User> getUser(@PathVariable String id) {
    return userService.findById(id);  // 非阻塞
}

// Spring Data JPA（同步）
public interface UserRepository extends JpaRepository<User, String> {
    User findByEmail(String email);
}

// Spring Data R2DBC（异步）
public interface UserRepository extends ReactiveCrudRepository<User, String> {
    Mono<User> findByEmail(String email);
}