在Spring Boot响应式架构中高效整合多外部API数据

import reactor.core.publisher.Mono;
import java.util.Map;
import java.util.HashMap;
import java.time.Duration;

// 1. 外部API服务通用接口
public interface ExternalApiService {
    /**
     * 异步获取外部API数据
     * @return 包含API数据的Mono，如果失败则返回包含错误信息的Mono
     */
    Mono<Map<String, Object>> fetchData();
}

// 2. 某个具体外部API的服务实现
// 假设这是用于获取用户信息的API
public class UserApiService implements ExternalApiService {

    private final String apiUrl;
    private final String apiKey;
    private final Duration timeout;

    public UserApiService(String apiUrl, String apiKey, Duration timeout) {
        this.apiUrl = apiUrl;
        this.apiKey = apiKey;
        this.timeout = timeout;
    }

    @Override
    public Mono<Map<String, Object>> fetchData() {
        // 模拟调用外部API的逻辑，例如使用WebClient
        // 实际应用中会使用Spring WebClient进行HTTP请求
        return Mono.delay(Duration.ofMillis(500)) // 模拟网络延迟
                   .map(l -> {
                       Map<String, Object> data = new HashMap<>();
                       data.put("user_id", "123");
                       data.put("user_name", "John Doe");
                       data.put("source", "UserApi");
                       return data;
                   })
                   .timeout(timeout) // 设置API调用超时
                   .onErrorResume(e -> {
                       // 定义错误处理和默认值
                       Map<String, Object> errorData = new HashMap<>();
                       errorData.put("user_status", "error");
                       errorData.put("user_message", "Failed to fetch user data: " + e.getMessage());
                       errorData.put("source", "UserApi");
                       return Mono.just(errorData); // 失败时返回默认/错误数据
                   });
    }
}

// 3. 另一个外部API的服务实现
// 假设这是用于获取订单信息的API
public class OrderApiService implements ExternalApiService {

    private final String apiUrl;
    private final String authHeader; // 假设使用不同的认证方式
    private final Duration timeout;

    public OrderApiService(String apiUrl, String authHeader, Duration timeout) {
        this.apiUrl = apiUrl;
        this.authHeader = authHeader;
        this.timeout = timeout;
    }

    @Override
    public Mono<Map<String, Object>> fetchData() {
        return Mono.delay(Duration.ofMillis(700)) // 模拟网络延迟
                   .map(l -> {
                       Map<String, Object> data = new HashMap<>();
                       data.put("order_id", "ORD-001");
                       data.put("order_amount", 199.99);
                       data.put("source", "OrderApi");
                       return data;
                   })
                   .timeout(timeout)
                   .onErrorResume(e -> {
                       Map<String, Object> errorData = new HashMap<>();
                       errorData.put("order_status", "error");
                       errorData.put("order_message", "Failed to fetch order data: " + e.getMessage());
                       errorData.put("source", "OrderApi");
                       return Mono.just(errorData);
                   });
    }
}

import org.springframework.stereotype.Service;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import java.util.List;
import java.util.Map;
import java.util.HashMap;
import java.util.stream.Collectors;

@Service
public class DataAggregationService {

    private final List<ExternalApiService> apiServices; // 注入所有外部API服务

    // 通过构造器注入所有 ExternalApiService 实现
    public DataAggregationService(List<ExternalApiService> apiServices) {
        this.apiServices = apiServices;
    }

    /**
     * 聚合所有外部API的数据并返回单个JSON
     * @return 包含所有聚合数据的Mono
     */
    public Mono<Map<String, Object>> aggregateAllData() {
        // 将所有服务的 fetchData() 调用转换为 Mono 列表
        List<Mono<Map<String, Object>>> monos = apiServices.stream()
                                                            .map(ExternalApiService::fetchData)
                                                            .collect(Collectors.toList());

        // 使用 Flux.merge 来并发执行所有 Mono，并收集它们的结果
        // Flux.merge 不保证结果的顺序，但会尽快发出每个 Mono 的结果
        return Flux.merge(monos)
                   .reduce(new HashMap<>(), (aggregatedMap, currentMap) -> {
                       // 将每个API返回的Map合并到总的聚合Map中
                       aggregatedMap.putAll(currentMap);
                       return aggregatedMap;
                   })
                   .map(finalMap -> {
                       // 可以添加一些聚合后的元数据
                       finalMap.put("aggregation_timestamp", System.currentTimeMillis());
                       return finalMap;
                   });
    }

    // 对于少量固定数量的API，也可以使用 Mono.zip
    public Mono<Map<String, Object>> aggregateSpecificData(UserApiService userApi, OrderApiService orderApi) {
        Mono<Map<String, Object>> userMono = userApi.fetchData();
        Mono<Map<String, Object>> orderMono = orderApi.fetchData();

        return Mono.zip(userMono, orderMono)
                   .map(tuple -> {
                       Map<String, Object> result = new HashMap<>();
                       result.putAll(tuple.getT1()); // 用户数据
                       result.putAll(tuple.getT2()); // 订单数据
                       result.put("aggregated_timestamp", System.currentTimeMillis());
                       return result;
                   });
    }
}