CompletableFuture.allOf()未等待全部任务完成问题求助

你的代码出现任务未完成就继续执行、列表未完全更新的核心原因有两个：

1. 非线程安全集合导致数据丢失 #

ArrayList不是线程安全集合，多个异步任务同时调用add()方法时，会因内部数组扩容、索引更新的竞态条件导致元素丢失或数据不一致，这就是valueList1和valueList2经常没更全的直接原因。

2. 共享对象的线程安全问题 #

outputObject.setSomeField()的逻辑存在竞态条件：多个线程可能同时通过outputObject.getSomeField()==null的判断，导致重复赋值或者预期外的结果。

另外，exceptionally中的异常处理存在问题：你直接抛出SomeCustomException，如果这是受检异常会编译报错；如果是未受检异常，会导致对应的CompletableFuture进入异常状态，但allOf仍会等待所有任务完成，不过这种抛出方式可能掩盖真实异常信息。

修复后的代码示例 #

推荐采用无副作用的结果收集方式（避免异步任务直接修改共享变量），同时使用线程安全的处理逻辑：

import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.stream.Collectors;
import java.util.concurrent.CompletableFuture;

// 假设的类型，根据实际情况替换
class MyData {}
class OutputObject {
    private Object someField;
    private List<AnotherValue> valueList1;
    private List<Value> valueList2;

    public Object getSomeField() { return someField; }
    public void setSomeField(Object someField) { this.someField = someField; }
    public void setValueList1(List<AnotherValue> valueList1) { this.valueList1 = valueList1; }
    public void setValueList2(List<Value> valueList2) { this.valueList2 = valueList2; }
}
class AnotherValue {}
class Value {}
class OutputFromMyMethod {
    private Object field;
    private AnotherValue value1;
    private Value value2;

    public Object getField() { return field; }
    public AnotherValue getValue1() { return value1; }
    public Value getValue2() { return value2; }
}

public class AsyncTaskDemo {
    public void processTasks() {
        List<MyData> dataList = getFromAMethod();
        OutputObject outputObject = new OutputObject();

        // 让每个异步任务返回封装好的结果，避免直接修改共享集合
        List<CompletableFuture<ResultWrapper>> futures = new ArrayList<>();

        for (MyData data : dataList) {
            CompletableFuture<ResultWrapper> future = CompletableFuture.supplyAsync(() -> {
                OutputFromMyMethod output = executeMyMethod(data);
                return new ResultWrapper(
                    output.getField(),
                    output.getValue1(),
                    output.getValue2()
                );
            }).exceptionally(throwable -> {
                // 统一处理异常，保留原始异常栈信息
                throw new RuntimeException("执行异步任务失败", throwable);
            });
            futures.add(future);
        }

        // 等待所有任务完成，并收集结果
        CompletableFuture<Void> allOf = CompletableFuture.allOf(
            futures.toArray(new CompletableFuture[0])
        );

        // 收集所有正常完成的结果（allOf已保证任务全部完成，join不会阻塞）
        List<ResultWrapper> results = allOf.thenApply(v -> {
            return futures.stream()
                .map(CompletableFuture::join)
                .collect(Collectors.toList());
        }).join();

        // 初始化someField：取第一个非null的field值，避免竞态条件
        results.stream()
            .map(ResultWrapper::getField)
            .filter(Objects::nonNull)
            .findFirst()
            .ifPresent(outputObject::setSomeField);

        // 批量收集valueList1和valueList2
        List<AnotherValue> valueList1 = results.stream()
            .map(ResultWrapper::getValue1)
            .collect(Collectors.toList());
        List<Value> valueList2 = results.stream()
            .map(ResultWrapper::getValue2)
            .collect(Collectors.toList());

        outputObject.setValueList1(valueList1);
        outputObject.setValueList2(valueList2);

        // 后续同步代码正常执行
        // taskOutput.setDisplayOutputs(ffxFeeDisplayOutputList);
    }

    // 模拟方法，根据实际情况实现
    private List<MyData> getFromAMethod() { return new ArrayList<>(); }
    private OutputFromMyMethod executeMyMethod(MyData data) { return new OutputFromMyMethod(); }

    // 辅助类：封装单个异步任务的结果
    static class ResultWrapper {
        private final Object field;
        private final AnotherValue value1;
        private final Value value2;

        public ResultWrapper(Object field, AnotherValue value1, Value value2) {
            this.field = field;
            this.value1 = value1;
            this.value2 = value2;
        }

        public Object getField() { return field; }
        public AnotherValue getValue1() { return value1; }
        public Value getValue2() { return value2; }
    }
}

关键优化点说明 #

• 避免共享状态修改：每个异步任务只返回结果，不直接修改外部集合或对象，从根源避免线程安全问题。
• 安全收集结果：通过allOf等待所有任务完成后，再统一收集结果，确保数据完整。
• 异常处理更清晰：exceptionally中用RuntimeException包装原始异常，既保留异常栈信息，又不会导致编译问题。
• 原子化初始化field：通过stream的findFirst确保someField只会被赋值一次，避免竞态条件。

如果一定要保留原有的“任务中更新集合”的方式，也可以将ArrayList替换为CopyOnWriteArrayList（适合读多写少场景），或者在add方法外层加synchronized块，但这种方式不如无副作用的结果收集优雅。

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