如何在CompletableFuture的thenApply中归约列表生成SmsDetails对象

直接在thenApply方法内完成结果归约即可，核心逻辑是遍历所有异步任务返回的MibsReturn对象提取目标msisdn字段，再组装为符合要求的SmsDetails对象。

核心逻辑说明 #

• 修正原生CompletableFuture的泛型声明，指定后续返回类型为SmsDetails，避免不必要的强制类型转换
• 遍历所有MibsReturn结果提取msisdn：如果业务上只需要统计发送成功的号码，增加成功状态过滤即可；如果需要全量号码直接提取对应字段
• 组装返回SmsDetails时保留原入参的公共属性（短信内容、发送方标识等），仅替换归约得到的msisdns列表字段

完整可运行代码 #

@Override
public SmsDetails send(SmsDetails smsDetails) {
    List<CompletableFuture<MibsReturn>> completableFutures =
            smsDetails.getDestinations().stream()
                    .map(msisdn -> getMibsReturns(msisdn, smsDetails.getMessage(), smsDetails.getSource()))
                    .collect(Collectors.toList());

    // 所有异步任务的汇总等待对象，传空数组即可由JVM自动适配长度，无需手动计算size
    CompletableFuture<Void> allFutures = CompletableFuture
            .allOf(completableFutures.toArray(new CompletableFuture[0]));

    CompletableFuture<List<MibsReturn>> allCompletableFuture = allFutures.thenApply(v ->
            completableFutures.stream()
                    .map(CompletableFuture::join)
                    .collect(Collectors.toList())
    );

    // 补全泛型，编写归约逻辑
    CompletableFuture<SmsDetails> completableFuture = allCompletableFuture.thenApply(mibsReturns -> {
        // 提取msisdn列表，以下方法名请和你实际业务类的方法定义保持一致
        List<String> msisdns = mibsReturns.stream()
                // 若只保留发送成功的号码，放开下面这行注释即可
                // .filter(mibsReturn -> mibsReturn.isSuccess())
                .map(MibsReturn::getMsisdn)
                .collect(Collectors.toList());

        // 组装返回结果，如果你的SmsDetails用Builder、全参构造而非setter，对应调整写法即可
        SmsDetails returnDetails = new SmsDetails();
        returnDetails.setMessage(smsDetails.getMessage());
        returnDetails.setSource(smsDetails.getSource());
        returnDetails.setMsisdns(msisdns);
        // 如需附加发送状态码、失败原因等字段，在此处补充赋值逻辑即可

        return returnDetails;
    });

    try {
        return completableFuture.get();
    } catch (InterruptedException | ExecutionException e) {
        // 恢复线程中断状态，按需包装为业务自定义异常抛出
        Thread.currentThread().interrupt();
        throw new RuntimeException("短信发送异步任务执行失败", e);
    }
}

private CompletableFuture<MibsReturn> getMibsReturns(String msisdn, String message, String senderId) {
    return CompletableFuture.supplyAsync(() -> sendViaMibs(msisdn, message, senderId));
}

注意事项 #

• 代码中MibsReturn#getMsisdn、MibsReturn#isSuccess以及SmsDetails的setter方法名需要和你实际项目中的类定义对齐，不要直接复制使用
• 原代码中CompletableFuture#get()方法会抛出受检异常，必须做对应处理，不要直接上抛导致编译失败
• 如果异步任务需要自定义线程池，不要用CompletableFuture.supplyAsync默认的ForkJoinPool，在方法入参里传入自定义线程池实例即可

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