大规模对象转POJO及多分类供应商API调用的可扩展性实现

一、用策略注册表解决请求体映射的扩展性问题 #

针对数百个分类的请求体差异，不用硬编码大量多态子类，而是用策略模式+注册表的方式，让每个分类对应独立的映射逻辑，通过注册表动态匹配：

1. 定义通用映射接口

// 通用的供应商请求体映射接口
public interface SupplierRequestMapper {
    // 返回该策略对应的分类标识
    String getCategory();
    // 将原始供应商JSON（用JsonNode/Map接收）转为API请求体
    Object mapToRequestBody(JsonNode supplierData);
}

2. 为每个分类实现映射策略
   比如"电子类"的实现：

@Component
public class ElectronicSupplierMapper implements SupplierRequestMapper {
    @Override
    public String getCategory() {
        return "electronic";
    }

    @Override
    public Object mapToRequestBody(JsonNode supplierData) {
        // 按该分类的API要求构造请求体
        ElectronicRequestDTO dto = new ElectronicRequestDTO();
        dto.setSupplierId(supplierData.get("id").asText());
        dto.setProductScore(supplierData.get("topScore").asDouble());
        // 其他字段映射逻辑...
        return dto;
    }
}

3. 构建策略注册表
   用Spring自动收集所有Mapper实例，手动实现也可以初始化一个Map：

@Component
public class RequestMapperRegistry {
    private final Map<String, SupplierRequestMapper> mapperMap;

    // Spring自动注入所有SupplierRequestMapper实现类
    public RequestMapperRegistry(List<SupplierRequestMapper> mappers) {
        this.mapperMap = mappers.stream()
                .collect(Collectors.toMap(SupplierRequestMapper::getCategory, Function.identity()));
    }

    // 根据分类获取对应的映射器
    public SupplierRequestMapper getMapper(String category) {
        SupplierRequestMapper mapper = mapperMap.get(category);
        if (mapper == null) {
            throw new IllegalArgumentException("无对应分类[" + category + "]的请求体映射器");
        }
        return mapper;
    }
}

二、优化流式处理的稳定性与可复用性 #

原来的并行流存在默认线程池占用、异常吞掉等问题，优化后封装成通用工具类：

1. 自定义业务线程池
   避免占用JVM默认ForkJoinPool，防止影响其他业务：

@Configuration
public class AsyncConfig {
    @Bean("supplierApiThreadPool")
    public Executor supplierApiThreadPool() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(10);
        executor.setMaxPoolSize(20);
        executor.setQueueCapacity(100);
        executor.setThreadNamePrefix("SupplierApi-");
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        executor.initialize();
        return executor;
    }
}

2. 封装通用API调用工具

@Component
public class SupplierApiClient {
    private static final Logger log = LoggerFactory.getLogger(SupplierApiClient.class);
    private final RequestMapperRegistry mapperRegistry;
    private final RestTemplate restTemplate;
    private final Executor apiThreadPool;

    @Autowired
    public SupplierApiClient(RequestMapperRegistry mapperRegistry, 
                             RestTemplate restTemplate,
                             @Qualifier("supplierApiThreadPool") Executor apiThreadPool) {
        this.mapperRegistry = mapperRegistry;
        this.restTemplate = restTemplate;
        this.apiThreadPool = apiThreadPool;
    }

    // 批量处理供应商数据的核心方法
    public void batchProcessSupplierData(Stream<JsonNode> supplierDataStream) {
        supplierDataStream
                // 按分类分组，确保每个分类只处理TOP10（如果原始数据未过滤）
                .collect(Collectors.groupingBy(node -> node.get("category").asText()))
                .entrySet().stream()
                .forEach(entry -> {
                    String category = entry.getKey();
                    List<JsonNode> top10Suppliers = entry.getValue().stream()
                            .sorted(Comparator.comparingDouble(n -> -n.get("score").asDouble()))
                            .limit(10)
                            .collect(Collectors.toList());

                    // 异步处理每个供应商的API调用
                    top10Suppliers.forEach(supplier -> {
                        apiThreadPool.execute(() -> {
                            try {
                                SupplierRequestMapper mapper = mapperRegistry.getMapper(category);
                                Object requestBody = mapper.mapToRequestBody(supplier);
                                String apiUrl = getApiUrlByCategory(category);
                                ResponseEntity<String> response = restTemplate.postForEntity(apiUrl, requestBody, String.class);
                                log.info("分类{}供应商{}调用成功，响应：{}", category, supplier.get("id").asText(), response.getBody());
                            } catch (Exception e) {
                                log.error("分类{}供应商{}调用失败", category, supplier.get("id").asText(), e);
                            }
                        });
                    });
                });
    }

    // 根据分类获取API地址，实际可抽离到配置文件
    private String getApiUrlByCategory(String category) {
        return "https://api.example.com/supplier/" + category + "/top";
    }
}

三、扩展性与维护性说明 #

• 新增分类时，只需实现SupplierRequestMapper接口并注册到容器，无需修改核心处理逻辑
• API地址、线程池参数等可抽离到配置文件，进一步提升灵活性
• 异常处理、响应日志统一封装，方便监控与问题排查

内容的提问来源于stack exchange，提问作者Joy