响应式服务中多外部API异步集成策略与实践

import reactor.core.publisher.Mono;
import org.springframework.stereotype.Service;
import org.springframework.web.reactive.function.client.WebClient;

// 定义通用外部API服务接口
public interface ExternalApiService<T> {
    Mono<T> fetchData(); // 异步获取数据
    String getServiceName(); // 获取服务名称
}

// 具体的API实现类A
@Service
public class ApiServiceA implements ExternalApiService<DataA> {
    private final WebClient webClient;

    public ApiServiceA(WebClient.Builder webClientBuilder) {
        this.webClient = webClientBuilder.baseUrl("http://api.external.com/serviceA").build();
    }

    @Override
    public Mono<DataA> fetchData() {
        // 调用外部API A的逻辑，可能包含限流、重试等
        return webClient.get().uri("/data")
                        .retrieve()
                        .bodyToMono(DataA.class)
                        .onErrorResume(e -> {
                            // 错误处理，例如记录日志，并返回一个默认值或空Mono
                            System.err.println("Error calling API A: " + e.getMessage());
                            return Mono.just(new DataA("default_value_A")); // 错误时返回默认值
                        });
    }

    @Override
    public String getServiceName() {
        return "API_A";
    }
}

// 具体的API实现类B (结构类似，针对不同API的URL、数据结构和错误处理)
@Service
public class ApiServiceB implements ExternalApiService<DataB> {
    private final WebClient webClient;

    public ApiServiceB(WebClient.Builder webClientBuilder) {
        this.webClient = webClientBuilder.baseUrl("http://api.external.com/serviceB").build();
    }

    @Override
    public Mono<DataB> fetchData() {
        return webClient.get().uri("/info")
                        .retrieve()
                        .bodyToMono(DataB.class)
                        .onErrorResume(e -> {
                            System.err.println("Error calling API B: " + e.getMessage());
                            return Mono.just(new DataB("default_value_B"));
                        });
    }

    @Override
    public String getServiceName() {
        return "API_B";
    }
}

// 假设的数据模型
class DataA {
    private String value;
    public DataA(String value) { this.value = value; }
    public String getValue() { return value; }
    // getters, setters
}

class DataB {
    private String info;
    public DataB(String info) { this.info = info; }
    public String getInfo() { return info; }
    // getters, setters
}

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

@Service
public class DataAggregatorService {

    private final ApiServiceA apiServiceA;
    private final ApiServiceB apiServiceB;
    // ... 注入所有外部API服务 (例如通过构造器注入列表或单独注入)

    public DataAggregatorService(ApiServiceA apiServiceA, ApiServiceB apiServiceB /* ... */) {
        this.apiServiceA = apiServiceA;
        this.apiServiceB = apiServiceB;
        // ...
    }

    /**
     * 获取聚合后的数据
     * @return 包含所有外部API数据的Mono<AggregatedResponse>
     */
    public Mono<AggregatedResponse> getAggregatedData() {
        Mono<DataA> monoA = apiServiceA.fetchData();
        Mono<DataB> monoB = apiServiceB.fetchData();
        // ... 其他API的Mono，例如 Mono<DataC> monoC = apiServiceC.fetchData();

        // 使用Mono.zip组合多个Mono的结果
        // Mono.zip最多支持8个Mono，如果超过20个，可以考虑链式zip或使用Flux.zip
        return Mono.zip(monoA, monoB, (dataA, dataB) -> {
            // 在这里将所有数据聚合成一个AggregatedResponse对象
            return new AggregatedResponse(dataA.getValue(), dataB.getInfo(), /* ... */);
        }).onErrorResume(e -> {
            // 聚合层面的错误处理，例如当所有API都失败，或聚合逻辑出现问题时
            System.err.println("Error during data aggregation: " + e.getMessage());
            return Mono.just(new AggregatedResponse("aggregation_error", "aggregation_error")); // 返回聚合默认值
        });
    }
}

// 假设的聚合响应类，用于封装所有API返回的数据
class AggregatedResponse {
    private String valueA;
    private String infoB;
    // ... 其他API的数据字段

    public AggregatedResponse(String valueA, String infoB) {
        this.valueA = valueA;
        this.infoB = infoB;
    }
    // getters, setters
}