Java多线程测试任务结果聚合方案优化咨询

我开发了一个运行数千项测试的应用，原本的多线程架构中，每个线程独立完成读输入、生成结果、对比结果及输出的全流程，但部分步骤耗时可达数分钟。因此我计划拆分任务：

• WorkerA线程负责将输入转换为拆分后的分步结果
• WorkerB线程负责结果对比
• WorkerC线程负责聚合所有结果生成统计数据并输出

目前我遇到的问题是WorkerC的实现：我采用以job id为键、对应任务的所有分步结果列表为值的Map来聚合结果，以此判断任务的所有步骤是否完成。我编写了模拟该流程的程序，可正常运行，但不确定是否存在更优的实现方式，希望获得相关建议。

Main类 #

public class Main {
    public static void main(String[] args) {
        try {
            new Test().run();
        } catch (Exception e) {
            System.out.println(e);
        }
    }
}

核心Test类 #

package org.example.worker;

import java.util.*;
import java.util.concurrent.*;

public class Test {
    public void run() throws Exception {
        //setup queues
        PriorityBlockingQueue<Result> resultQueue = new PriorityBlockingQueue<>();
        PriorityBlockingQueue<Job> jobQueue = new PriorityBlockingQueue<>();

        //setup thread count and random jobs
        int threadCnt = 4;
        Random rnd = new Random(threadCnt);

        for(int i=0;i<20;i++) {
            int num = (rnd.nextInt(0, 100) / 17) + 1;
            jobQueue.add(new Job(i, num));
        }

        //start workers
        WorkerB workerB;

        try(ExecutorService es = Executors.newFixedThreadPool(threadCnt)) {
            List<Future<?>> futures = new ArrayList<>();

            workerB = new WorkerB(0, resultQueue);
            futures.addFirst(es.submit(workerB));

            for(int i=1;i<threadCnt;i++) {
                futures.addFirst(es.submit(new WorkerA(i, jobQueue, resultQueue)));
            }

            //wait for threads to finish, be sure that the result worker is the last to finish
            for(int i=0; i<futures.size(); i++) {
                if(i == futures.size() - 1) {
                    workerB.stop();
                }

                futures.get(i).get();
            }
        }
    }

    class WorkerB implements Runnable {
        private final PriorityBlockingQueue<Result> resultQueue;
        private final int id;
        private boolean stop;

        public WorkerB(int id, PriorityBlockingQueue<Result> resultQueue) {
            this.id = id;
            this.resultQueue = resultQueue;
        }

        public void stop() {
            stop = true;
        }

        private void process(Result result) {
            System.out.println(id + ":: working on: " + result);
        }

        private void check(Map<Integer, List<Result>> results, int id) {
            List<Result> tmp = results.get(id);

            if(tmp == null || tmp.isEmpty()) {
                return;
            }

            if(tmp.getFirst().total() == tmp.size()) {
                tmp.stream().sorted().forEach(this::process);

                results.remove(id);
            }
        }

        private void addOrProcess(Map<Integer, List<Result>> results, Result result) {
            if(result.total() < 2) {
                process(result);

                return;
            }

            if(!results.containsKey(result.id())) {
                results.put(result.id(), new ArrayList<>());
            }

            results.get(result.id()).add(result);
        }

        @Override
        public void run() {
            try {
                Map<Integer, List<Result>> results = new HashMap<>();
                Result result = resultQueue.poll(100, TimeUnit.MILLISECONDS);

                while (!stop || result != null) {
                    if(result != null) {
                        addOrProcess(results, result);
                        check(results, result.id());
                    }

                    result = resultQueue.poll(100, TimeUnit.MILLISECONDS);
                }

                results.keySet().forEach(k -> check(results, k));
            } catch (Exception e) {
                System.out.println(id + ":: " + e);
            }

            System.out.println(id + ":: shutting down");
        }
    }

    class WorkerA implements Runnable {
        private final PriorityBlockingQueue<Job> jobQueue;
        private final PriorityBlockingQueue<Result> resultQueue;
        private final int id;

        public WorkerA(int id, PriorityBlockingQueue<Job> jobQueue, PriorityBlockingQueue<Result> resultQueue) {
            this.id = id;
            this.jobQueue = jobQueue;
            this.resultQueue = resultQueue;
        }

        @Override
        public void run() {
            try {
                Random rnd = new Random(id);
                Job job;

                while ((job = jobQueue.poll(100, TimeUnit.MILLISECONDS)) != null) {
                    int num = (rnd.nextInt(0, 100) / 30) + 1;

                    for(int i=0;i<num;i++) {
                        resultQueue.add(new Result(job.id(), i + 1, num));

                        Thread.sleep(100L * job.amt());
                    }
                }
            } catch (Exception e) {
                System.out.println(id + ":: " + e);
            }

            System.out.println(id + ":: shutting down");
        }
    }
}

Job和Result记录类 #

Job类 #

public record Job (
        int id,
        int amt
)  implements Comparable<Job> {
    @Override
    public int compareTo(Job job) {
        if(job == this) {
            return 0;
        }

        if(this.id < job.id) {
            return -1;
        } else if(this.id > job.id) {
            return 1;
        }

        return 0;
    }
}

Result类 #

public record Result(
        int id,
        int part,
        int total
) implements Comparable<Result> {
    @Override
    public int compareTo(Result result) {
        if(result == this) {
            return 0;
        }

        if(this.id < result.id) {
            return -1;
        } else if(this.id > result.id) {
            return 1;
        }

        return 0;
    }
}

1. 优化聚合容器与完成判断逻辑 #

当前用HashMap<Integer, List<Result>>的方式需要每次通过列表大小判断任务是否完成，效率和可靠性都有提升空间：

• 自定义JobInfo类封装任务状态：存储任务的预期总步数、已收集结果列表，避免重复从结果中获取total值，同时校验后续结果的total是否一致，防止数据异常：

  private static class JobInfo {
      final int expectedTotal;
      final List<Result> results = new ArrayList<>();
  
      JobInfo(int expectedTotal) {
          this.expectedTotal = expectedTotal;
      }
  }
  

  使用时，第一次收到某个job的结果就初始化JobInfo，后续添加结果后直接判断jobInfo.results.size() == jobInfo.expectedTotal即可。
• 可以搭配CountDownLatch跟踪进度：每个job对应一个CountDownLatch，每收到一个结果就countDown，当latch计数归0后自动触发结果处理，避免轮询判断。

2. 线程安全与性能优化 #

• 当前WorkerB是单线程处理，HashMap是安全的，但如果后续要扩展多WorkerB实例，必须替换为线程安全容器，比如ConcurrentHashMap，并使用computeIfAbsent方法避免竞态：

  results.computeIfAbsent(result.id(), k -> new JobInfo(result.total())).results.add(result);
  

• 单线程场景下，用ArrayDeque代替ArrayList存储结果，添加和遍历的效率更高。

3. 修复排序逻辑 #

当前Result的compareTo只比较job id，导致同一job的分步结果排序后可能乱序。修改compareTo方法，先比较id，再比较part，保证同一job的结果按分步顺序处理：

@Override
public int compareTo(Result result) {
    if (this == result) return 0;
    int idCompare = Integer.compare(this.id, result.id);
    if (idCompare != 0) return idCompare;
    return Integer.compare(this.part, result.part);
}

4. 优化停机逻辑 #

• WorkerB中的stop变量没有声明为volatile，主线程调用stop()后，WorkerB线程可能无法立刻感知到这个变化，导致停机延迟。需修改为：

  private volatile boolean stop;
  

• 处理剩余结果时，直接遍历Map的values集合比遍历keySet更高效：

  results.values().forEach(jobInfo -> {
      if (jobInfo.results.size() == jobInfo.expectedTotal) {
          jobInfo.results.stream().sorted().forEach(this::process);
      }
  });
  

5. 复杂场景下的框架选型 #

如果后续业务复杂度提升（比如需要失败重试、结果过滤、异步回调等），可以考虑使用响应式框架（如RxJava、Project Reactor），这类框架内置了groupBy、buffer等聚合操作符，能自动按job id分组并收集所有结果后触发处理，减少自定义聚合逻辑的工作量。

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