Spring WebFlux 应用启动时如何优雅地处理响应式数据初始化

// 假设的Data类和R2dbcRepository接口
public class Data {
    private String id;
    private String value;

    public Data(String id, String value) {
        this.id = id;
        this.value = value;
    }

    public String getId() { return id; }
    public String getValue() { return value; }

    @Override
    public String toString() {
        return "Data{" + "id='" + id + '\'' + ", value='" + value + '\'' + '}';
    }
}

// 模拟的R2dbcRepository
interface R2dbcRepository {
    Mono<Data> findById(String id);
    Flux<Data> findAll();
    // ... 其他方法
}

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import reactor.core.publisher.Mono;
import reactor.core.publisher.Flux;

@Configuration
public class AppConfig {

    private final R2dbcRepository repo;

    // Spring会自动注入R2dbcRepository实例
    public AppConfig(R2dbcRepository repo) {
        this.repo = repo;
    }

    /**
     * 定义一个缓存的Mono<Data> Bean。
     * 实际的数据库查询只会在第一次订阅时执行，之后的结果会被缓存。
     */
    @Bean
    public Mono<Data> myCachedSingleDbData() {
        System.out.println("Defining myCachedSingleDbData Bean...");
        return repo.findById("config_key_1") // 假设查找特定配置
                   .map(it -> new Data(it.getId(), it.getValue().toUpperCase())) // 示例转换
                   .doOnSubscribe(s -> System.out.println("Subscribing to myCachedSingleDbData - DB call will happen now."))
                   .doOnNext(data -> System.out.println("Data fetched and cached: " + data))
                   .cache(); // 关键：缓存结果
    }

    /**
     * 定义一个缓存的Flux<Data> Bean，用于加载所有数据。
     */
    @Bean
    public Flux<Data> myCachedAllDbData() {
        System.out.println("Defining myCachedAllDbData Bean...");
        return repo.findAll()
                   .doOnSubscribe(s -> System.out.println("Subscribing to myCachedAllDbData - DB call will happen now."))
                   .doOnNext(data -> System.println("One item fetched and cached: " + data))
                   .cache(); // 关键：缓存结果
    }
}

import org.springframework.stereotype.Service;
import reactor.core.publisher.Mono;
import reactor.core.publisher.Flux;

@Service
public class MyBusinessService {

    private final Mono<Data> cachedSingleDbData;
    private final Flux<Data> cachedAllDbData;

    // Spring会自动注入上面定义的缓存Bean
    public MyBusinessService(Mono<Data> myCachedSingleDbData, Flux<Data> myCachedAllDbData) {
        this.cachedSingleDbData = myCachedSingleDbData;
        this.cachedAllDbData = myCachedAllDbData;
        System.out.println("MyBusinessService initialized.");
    }

    public Mono<String> processSingleData() {
        System.out.println("MyBusinessService: Processing single data...");
        return cachedSingleDbData
                .map(data -> "Processed: " + data.getValue())
                .doOnSuccess(s -> System.out.println("Single data processed: " + s));
    }

    public Flux<String> processAllData() {
        System.out.println("MyBusinessService: Processing all data...");
        return cachedAllDbData
                .map(data -> "Item: " + data.getId() + "-" + data.getValue())
                .doOnComplete(() -> System.out.println("All data processed."));
    }

    // 假设一个在应用启动后可能被调用的方法
    // 第一次调用时会触发数据库查询和缓存
    public Mono<Void> performInitialSetup() {
        return processSingleData()
                .then(processAllData())
                .then(); // 确保两个流都完成
    }
}