Java响应式编程Mono接口详解

// 1. 从确定值创建
Mono<String> mono1 = Mono.just("Hello, World!");
// 2. 创建空 Mono
Mono<String> mono2 = Mono.empty();
// 3. 从可能为null的值创建（如果value为null，则返回空Mono）
Mono<String> mono3 = Mono.justOrNull(someNullableValue);
// 4. 从可调用对象创建（延迟计算）
Mono<String> mono4 = Mono.fromCallable(() -> {
    // 可能抛出异常的计算
    return expensiveComputation();
});
// 5. 从 Future 创建
CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> "Result");
Mono<String> mono5 = Mono.fromFuture(future);
// 6. 从 Supplier 创建
Mono<String> mono6 = Mono.fromSupplier(() -> generateValue());
// 7. 创建错误 Mono
Mono<String> mono7 = Mono.error(new RuntimeException("Something went wrong"));
// 8. 延迟创建（直到有订阅者时才执行）
Mono<String> mono8 = Mono.defer(() -> {
    return Mono.just(System.currentTimeMillis() + " - generated");
});

Mono<User> userMono = getUserById(userId)
    .filter(user -> user.isActive())  // 过滤非活跃用户
    .map(user -> {                    // 转换数据
        UserDTO dto = new UserDTO();
        dto.setName(user.getName());
        dto.setEmail(user.getEmail());
        return dto;
    })
    .flatMap(dto -> {                 // 异步转换到另一个Mono
        return sendNotification(dto.getEmail())
            .map(success -> dto);
    })
    .doOnNext(dto -> {                // 副作用操作
        log.info("Processed user: {}", dto.getName());
    })
    .doOnError(error -> {             // 错误时的副作用
        log.error("Failed to process user", error);
    })
    .timeout(Duration.ofSeconds(5))   // 设置超时
    .retry(3);                        // 失败时重试3次

Mono<String> safeMono = riskyOperation()
    .onErrorResume(error -> {          // 发生错误时提供备用Mono
        log.warn("Operation failed, using fallback", error);
        return Mono.just("Fallback Value");
    })
    .onErrorReturn("Default Value")    // 发生错误时返回默认值
    .onErrorMap(error -> {             // 转换错误类型
        return new BusinessException("Operation failed", error);
    })
    .doFinally(signal -> {             // 最终清理（无论成功或失败）
        cleanupResources();
    });

// 1. 最简单的订阅（触发执行但忽略结果）
mono.subscribe();
// 2. 带成功回调的订阅
mono.subscribe(
    value -> System.out.println("Received: " + value)
);
// 3. 带成功和错误回调的订阅
mono.subscribe(
    value -> System.out.println("Received: " + value),
    error -> System.err.println("Error: " + error.getMessage())
);
// 4. 完整的订阅（成功、错误、完成）
mono.subscribe(
    value -> System.out.println("Received: " + value),
    error -> System.err.println("Error: " + error.getMessage()),
    () -> System.out.println("Completed successfully")
);
// 5. 带订阅回调的订阅
mono.subscribe(
    new Subscriber<String>() {
        @Override
        public void onSubscribe(Subscription s) {
            s.request(1);  // 请求数据
        }
        @Override
        public void onNext(String value) {
            System.out.println("Received: " + value);
        }
        @Override
        public void onError(Throwable t) {
            System.err.println("Error: " + t.getMessage());
        }
        @Override
        public void onComplete() {
            System.out.println("Completed");
        }
    }
);

Mono<String> debugMono = someOperation()
    .log("my.mono")  // 添加日志
    .checkpoint("debug point")  // 添加检查点
    .doOnSubscribe(s -> log.debug("Subscribed"))
    .doOnNext(v -> log.debug("Next: {}", v))
    .doOnError(e -> log.error("Error: {}", e.getMessage()))
    .doOnCancel(() -> log.debug("Cancelled"))
    .doOnSuccess(v -> log.debug("Completed with: {}", v));