SwingWorker的doInBackground性能优化：实现Market对象并行创建

我有一个继承SwingWorker的类，在doInBackground方法里遍历对象列表创建新的Market对象。每个Market实例化耗时约1秒（因构造函数会调用initialize方法），目前整体为顺序执行，速度太慢，希望实现循环和对象创建的并行执行。

原顺序执行代码 #

@Override
protected Void doInBackground() throws Exception {
    System.out.println("Market loader running task with ticker size; " + this.ticker.size());
    synchronized (markets) {
        for (int i = 0; i < ticker.size(); i++) {
            TickerStatistics t = this.ticker.get(i);
            markets.add(new Market(t.getSymbol(), this.interval));  //每次执行约耗时1秒，导致整体顺序执行
            setProgress(100 * i / ticker.size());
        }
    }
    return null;
}

错误的并行尝试（仍顺序执行） #

我尝试了以下写法，但速度依然很慢，本质还是顺序执行：

protected Void doInBackground() throws Exception {
    System.out.println("Market loader running task with ticker size; " + this.ticker.size());
    List<CompletableFuture> futureList = new ArrayList<>();
    for (int i = 0; i < ticker.size(); i++) {
        TickerStatistics t = this.ticker.get(i);
        int count = i+1;
        futureList.add(CompletableFuture.supplyAsync(() -> markets.add(new Market(t.getSymbol(), interval))).thenAccept(progress -> setProgress(100 * count / ticker.size())));
        futureList.forEach(CompletableFuture::join);
    }
    return null;
}

问题分析与优化方案 #

错误尝试的核心问题 #

你在循环内部每次添加Future后立刻调用futureList.forEach(CompletableFuture::join)，这会强制等待当前任务执行完毕才进入下一次循环，完全没有利用到并行执行的优势，和顺序执行没区别。

正确的并行实现 #

1. 先批量启动所有异步任务，再统一等待完成
   将join操作移到循环外，让所有Future同时启动执行，最后再等待全部任务完成。
2. 避免异步任务直接操作共享集合
   原代码中直接在异步任务里调用markets.add()，如果markets不是线程安全集合，会引发并发问题。建议先在异步任务中创建Market对象，最后统一批量添加到共享集合。
3. 线程安全地更新进度
   使用原子计数器跟踪已完成任务数，确保进度更新的线程安全性，避免多任务同时更新进度导致的计算错误。

优化后的代码示例 #

import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;

@Override
protected Void doInBackground() throws Exception {
    int totalTasks = ticker.size();
    System.out.println("Market loader running task with ticker size; " + totalTasks);
    
    // 原子计数器，线程安全地统计已完成任务数
    AtomicInteger completedTasks = new AtomicInteger(0);

    // 批量创建异步任务，仅负责Market对象的初始化
    List<CompletableFuture<Market>> futureList = ticker.stream()
            .map(tickerStat -> CompletableFuture.supplyAsync(() -> new Market(tickerStat.getSymbol(), this.interval))
                    .whenComplete((market, throwable) -> {
                        // 任务完成后更新计数器和进度
                        int completed = completedTasks.incrementAndGet();
                        setProgress(100 * completed / totalTasks);
                    }))
            .collect(Collectors.toList());

    // 等待所有任务完成，收集所有创建好的Market对象
    List<Market> newMarkets = futureList.stream()
            .map(CompletableFuture::join)
            .collect(Collectors.toList());

    // 最后统一同步添加到共享集合，减少锁的持有时间
    synchronized (markets) {
        markets.addAll(newMarkets);
    }

    return null;
}

额外优化建议 #

• 自定义线程池控制并发数：默认的supplyAsync会使用ForkJoinPool，若任务数量过多可能导致资源耗尽。可以自定义线程池限制并发数：

  // 根据CPU核心数设置线程池大小，避免过度并发
  ExecutorService customExecutor = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());
  // 创建任务时指定自定义线程池
  CompletableFuture.supplyAsync(() -> new Market(...), customExecutor);
  // 任务全部提交后关闭线程池
  customExecutor.shutdown();
  

• 拆分耗时初始化操作：如果Market的initialize方法内有可并行的子操作（比如IO请求、数据计算），可以在该方法内部进一步做并行优化。

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