Spring WebFlux 核心操作符之map、flatMap 与 Mono 常用方法详解

// 基本数据转换
Flux<Integer> numbers = Flux.just(1, 2, 3, 4, 5);
Flux<Integer> squared = numbers.map(n -> n * n);
// 输出: 1, 4, 9, 16, 25
// WebFlux 中的实体转换
public Mono<UserDTO> getUserById(Long id) {
    return userRepository.findById(id)
           .map(user -> {
               // 同步转换 Entity 到 DTO
               UserDTO dto = new UserDTO();
               dto.setId(user.getId());
               dto.setName(user.getName());
               dto.setEmail(user.getEmail());
               return dto;
           });
}

// 异步数据转换
Flux<Integer> numbers = Flux.just(1, 2, 3, 4, 5);
Flux<Integer> result = numbers.flatMap(n -> 
    Mono.just(n * n).delayElement(Duration.ofMillis(100))
);
// WebFlux 中的复杂业务处理
public Mono<OrderWithDetailsDTO> getOrderWithDetails(Long orderId) {
    return orderRepository.findById(orderId)
           .flatMap(order -> {
               // 异步查询关联数据
               return productService.getProduct(order.getProductId())
                      .flatMap(product -> 
                         userService.getUser(order.getUserId())
                               .map(user -> {
                                   OrderWithDetailsDTO dto = new OrderWithDetailsDTO();
                                   dto.setOrder(order);
                                   dto.setProduct(product);
                                   dto.setUser(user);
                                   return dto;
                               })
                      );
           });
}

// 基础创建
Mono<String> mono1 = Mono.just("Hello");
Mono<String> mono2 = Mono.justOrEmpty(null); // 空 Mono
Mono<String> mono3 = Mono.justOrEmpty(Optional.of("value"));
Mono<String> emptyMono = Mono.empty();
Mono<String> errorMono = Mono.error(new RuntimeException("Error"));
// 延迟创建
Mono<String> deferredMono = Mono.defer(() -> 
    Mono.just("Value created at subscription time: " + System.currentTimeMillis())
);
// 从其他类型创建
Mono<String> fromCallable = Mono.fromCallable(() -> 
    expensiveOperation()
);
Mono<String> fromFuture = Mono.fromFuture(
    CompletableFuture.supplyAsync(() -> "Future result")
);

Mono<String> original = Mono.just("hello");
// 转换操作
Mono<String> upperCase = original.map(String::toUpperCase);
Mono<Integer> length = original.map(String::length);
// 异步转换
Mono<String> processed = original.flatMap(str -> 
    processStringAsync(str)
);
// 过滤操作
Mono<String> filtered = original.filter(str -> str.length() > 3);
Mono<String> defaultIfEmpty = Mono.<String>empty()
    .defaultIfEmpty("Default Value");
// 类型转换
Mono<Object> objectMono = Mono.just("hello");
Mono<String> casted = objectMono.cast(String.class);

Mono<String> unreliableMono = createUnreliableMono();
// 基础错误处理
Mono<String> safeMono = unreliableMono
    .onErrorReturn("Fallback Value")
    .onErrorResume(TimeoutException.class, ex -> 
        Mono.just("Timeout Fallback")
    )
    .onErrorResume(ex -> 
        backupService.getData().onErrorReturn("Final Fallback")
    );
// 错误转换
Mono<String> mappedError = unreliableMono
    .onErrorMap(IOException.class, ex -> 
        new BusinessException("Data access failed", ex)
    );
// 重试机制
Mono<String> withRetry = unreliableMono
    .retry(3) // 简单重试3次
    .retryWhen(Retry.backoff(3, Duration.ofSeconds(1)) // 指数退避重试
    .timeout(Duration.ofSeconds(5)); // 超时控制

Mono<String> userMono = getUser();
Mono<String> profileMono = getProfile();
Mono<Integer> scoreMono = getScore();
// zip - 并行执行并组合结果
Mono<Tuple3<String, String, Integer>> zipped = 
    Mono.zip(userMono, profileMono, scoreMono);
Mono<String> combined = Mono.zip(userMono, profileMono)
    .map(tuple -> tuple.getT1() + " - " + tuple.getT2());
// zipWith - 链式组合
Mono<String> userWithProfile = userMono
    .zipWith(profileMono, (user, profile) -> user + " : " + profile);
// then - 顺序执行（忽略前一个结果）
Mono<Void> sequence = userMono
    .then(profileMono)
    .then(cleanupOperation());
// when - 等待多个操作完成
Mono<Void> allCompleted = Mono.when(userMono, profileMono, scoreMono);

Mono<String> businessMono = getBusinessData();
Mono<String> withLogging = businessMono
    .doOnSubscribe(subscription -> 
        log.info("Starting business operation")
    )
    .doOnNext(value -> 
        log.info("Processing value: {}", value)
    )
    .doOnSuccess(value -> 
        log.info("Operation completed successfully: {}", value)
    )
    .doOnError(error -> 
        log.error("Operation failed", error)
    )
    .doOnCancel(() -> 
        log.warn("Operation cancelled")
    )
    .doOnTerminate(() -> 
        log.info("Operation terminated")
    );

Mono<String> dataMono = getData();
// 缓存
Mono<String> cached = dataMono.cache(Duration.ofMinutes(10));
// 延迟
Mono<String> delayed = dataMono.delayElement(Duration.ofSeconds(1));
// 超时控制
Mono<String> withTimeout = dataMono.timeout(Duration.ofSeconds(5));
// 重复（转换为 Flux）
Flux<String> repeated = dataMono.repeat(3);
// 日志调试
Mono<String> withLog = dataMono.log("data.flow");

public Mono<OrderResult> processUserOrder(OrderRequest request) {
    return validateRequest(request)
        .flatMap(validated -> 
            inventoryService.checkStock(validated.getProductId(), validated.getQuantity())
        )
        .flatMap(stockAvailable -> {
            if (!stockAvailable) {
                return Mono.error(new InsufficientStockException());
            }
            return processPayment(request);
        })
        .flatMap(paymentResult -> {
            if (paymentResult.isSuccess()) {
                return createOrder(request)
                    .flatMap(order -> 
                        updateInventory(order)
                            .then(sendConfirmationEmail(order))
                            .then(Mono.just(OrderResult.success(order)))
                    );
            } else {
                return Mono.just(OrderResult.failed("Payment failed: " + paymentResult.getMessage()));
            }
        })
        .timeout(Duration.ofSeconds(30))
        .retryWhen(Retry.backoff(3, Duration.ofSeconds(1))
        .doOnSuccess(result -> 
            metricsService.recordOrderSuccess(result.getOrderId())
        )
        .doOnError(error -> {
            log.error("Order processing failed for request: {}", request, error);
            metricsService.recordOrderFailure();
        })
        .onErrorResume(ex -> 
            handleOrderFailure(request, ex)
        );
}
private Mono<OrderResult> handleOrderFailure(OrderRequest request, Throwable ex) {
    if (ex instanceof TimeoutException) {
        return Mono.just(OrderResult.failed("Order timeout, please try again"));
    } else if (ex instanceof InsufficientStockException) {
        return Mono.just(OrderResult.failed("Insufficient stock"));
    } else {
        return compensationService.compensateOrder(request)
            .then(Mono.just(OrderResult.failed("System error, order cancelled")));
    }
}

public Flux<ProcessedItem> processBatch(Flux<InputItem> items) {
    return items
        .window(100) // 每100个元素为一组
        .flatMap(window -> 
            window.flatMap(this::validateItem)
                  .collectList()
                  .flatMap(validatedItems -> 
                      processBatchAsync(validatedItems)
                          .timeout(Duration.ofMinutes(5))
                          .retry(2)
                  )
                  .flatMapIterable(ProcessedBatch::getItems)
        )
        .doOnNext(processed -> 
            log.debug("Processed item: {}", processed.getId())
        )
        .doOnComplete(() -> 
            log.info("Batch processing completed")
        );
}

// 良好的错误处理模式
public Mono<ApiResponse> robustApiCall() {
    return externalService.call()
        .timeout(Duration.ofSeconds(10))
        .retryWhen(Retry.backoff(3, Duration.ofSeconds(1)))
        .onErrorResume(TimeoutException.class, ex -> 
            fallbackService.getData()
        )
        .onErrorReturn(ApiResponse.error("Service unavailable"))
        .doOnError(ex -> 
            metrics.increment("api.call.failed")
        );
}

// 添加详细的监控点
Mono<String> monitoredOperation = dataSource.getData()
    .name("database.query")
    .metrics()
    .doOnSubscribe(s -> 
        tracer.startSpan("business-operation")
    )
    .doOnNext(value -> 
        log.debug("Intermediate value: {}", value)
    )
    .doOnTerminate(() -> 
        tracer.finishSpan()
    );