Java ExecutorService多线程任务未全部执行问题求助

算力阿峰

2026-7-16

多线程环境下ExecutorService任务执行计数异常问题解决

问题描述

通过Java的ExecutorService执行多组列表的迭代任务，总计需调用1890万次calculateRecords方法。测试发现：

单线程运行时，int类型变量count的统计值符合预期；
多线程运行时，count始终无法达到预期值，任务看似提前终止。

相关代码如下：

package main;

import java.text.DecimalFormat;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class Test
{
    public static int count = 0;
    private static final DecimalFormat decimalFormat = new DecimalFormat("#.###");
    
    public static void main(String[] args)
    {
        int aStart = 140;
        int aEnd = 400;
        int aIncrement = 20;
        
        double bStart = 9;
        double bEnd = 27;
        double bIncrement = 0.75;
        
        int cStart = 14;
        int cEnd = 18;
        int cIncrement = 1;
        
        int dStart = 90;
        int dEnd = 180;
        int dIncrement = 30;
        
        int eStart = 60;
        int eEnd = 120;
        int eIncrement = 30;
        
        int fStart = 130;
        int fEnd = 250;
        int fIncrement = 30;
        
        double gStart = 5.5;
        double gEnd = 6.7;
        double gIncrement = 0.3;
        
        double hStart = 2.8;
        double hEnd = 4.2;
        double hIncrement = 0.2;
        
        double iStart = 2.6;
        double iEnd = 4;
        double iIncrement = 0.2;
        
        List<Integer> listA = list(aStart, aEnd, aIncrement);
        List<Double>  listB = list(bStart, bEnd, bIncrement);
        List<Integer> listC = list(cStart, cEnd, cIncrement);
        List<Integer> listD = list(dStart, dEnd, dIncrement);
        List<Integer> listE = list(eStart, eEnd, eIncrement);
        List<Integer> listF = list(fStart, fEnd, fIncrement);
        List<Double>  listG = list(gStart, gEnd, gIncrement);
        List<Double>  listH = list(hStart, hEnd, hIncrement);
        List<Double>  listI = list(iStart, iEnd, iIncrement);
        
        final int MAX_THREADS = 2;
        ExecutorService executor = Executors.newFixedThreadPool(MAX_THREADS);
        
        List<String> row = new ArrayList<>();
        
        for (int a : listA)
        {
            for (double b : listB)
            {
                for (int c : listC)
                {
                    for (int d : listD)
                    {
                        for (int e : listE)
                        {
                            for (int f : listF)
                            {
                                for (double g : listG)
                                {
                                    for (double h : listH)
                                    {
                                        for (double i : listI)
                                        {
                                            if (i < h) 
                                            {
                                                executor.submit(() -> row.addAll(calculateRecords("C://CSV", a, b, c, d, e, f, g, h, i)));
                                            }
                                        }
                                    }
                                }
                            }
                        }
                    }
                }
            }
        }
        
        executor.shutdown();
        try
        {
            while (!executor.isTerminated())
            {
                // Wait for all tasks to complete
            }
            System.out.println("TEST COMPLETE");
        } catch (Exception e)
        {
            System.out.println("Thread execution interrupted.");
            Thread.currentThread().interrupt();
        }

        System.out.println(count);
    }
    
    public static ArrayList<Integer> list(int startPoint, int endPoint, int incrementPoint)
    {
        ArrayList<Integer> a = new ArrayList<>();
        for (int queryPoint = startPoint; queryPoint <= endPoint; queryPoint += incrementPoint)
        {
            a.add(queryPoint);
        }
        
        return a;
    }
    
    public static ArrayList<Double> list(double startPoint, double endPoint, double incrementPoint)
    {
        ArrayList<Double> a = new ArrayList<>();
        double tolerance = incrementPoint / 2.0;
        for (double queryPoint = startPoint; queryPoint <= endPoint + tolerance; queryPoint += incrementPoint)
        {
            double formatPoint = Double.parseDouble(decimalFormat.format(queryPoint));
            a.add(formatPoint);
        }
        return a;
    }
    
    public static List<String> calculateRecords(String directory, int a, double b, int c, int d, int e, int f, double g, double h, double i)
    {
        List<String> listC = new ArrayList<>();
        count++;
        return listC;
    }
}

问题根源

非线程安全的计数器：count是普通int类型，count++操作并非原子性（包含读取、加1、写入三个步骤），多线程并发执行时会出现竞态条件，导致计数丢失，最终统计值低于预期。
非线程安全的集合操作：row是ArrayList，不支持并发修改。多个线程同时调用row.addAll()会触发ConcurrentModificationException，异常可能导致任务提前终止，进一步影响计数准确性。

解决方案

1. 替换为线程安全的计数器

使用java.util.concurrent.atomic.AtomicInteger替代int类型的count，它提供原子性的自增操作，避免多线程下的计数丢失。

2. 使用线程安全的集合

将ArrayList<String> row替换为线程安全的集合，比如CopyOnWriteArrayList<String>，或者在主线程汇总任务结果（更高效），避免并发修改异常。

3. 优化等待任务完成的逻辑

使用executor.awaitTermination()替代空循环等待，更优雅且节省资源。

修改后的代码示例

package main;

import java.text.DecimalFormat;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

public class Test
{
    // 替换为线程安全的原子计数器
    public static AtomicInteger count = new AtomicInteger(0);
    private static final DecimalFormat decimalFormat = new DecimalFormat("#.###");
    
    public static void main(String[] args)
    {
        int aStart = 140;
        int aEnd = 400;
        int aIncrement = 20;
        
        double bStart = 9;
        double bEnd = 27;
        double bIncrement = 0.75;
        
        int cStart = 14;
        int cEnd = 18;
        int cIncrement = 1;
        
        int dStart = 90;
        int dEnd = 180;
        int dIncrement = 30;
        
        int eStart = 60;
        int eEnd = 120;
        int eIncrement = 30;
        
        int fStart = 130;
        int fEnd = 250;
        int fIncrement = 30;
        
        double gStart = 5.5;
        double gEnd = 6.7;
        double gIncrement = 0.3;
        
        double hStart = 2.8;
        double hEnd = 4.2;
        double hIncrement = 0.2;
        
        double iStart = 2.6;
        double iEnd = 4;
        double iIncrement = 0.2;
        
        List<Integer> listA = list(aStart, aEnd, aIncrement);
        List<Double>  listB = list(bStart, bEnd, bIncrement);
        List<Integer> listC = list(cStart, cEnd, cIncrement);
        List<Integer> listD = list(dStart, dEnd, dIncrement);
        List<Integer> listE = list(eStart, eEnd, eIncrement);
        List<Integer> listF = list(fStart, fEnd, fIncrement);
        List<Double>  listG = list(gStart, gEnd, gIncrement);
        List<Double>  listH = list(hStart, hEnd, hIncrement);
        List<Double>  listI = list(iStart, iEnd, iIncrement);
        
        final int MAX_THREADS = 2;
        ExecutorService executor = Executors.newFixedThreadPool(MAX_THREADS);
        
        // 使用线程安全集合，或改为主线程汇总结果
        List<String> row = new java.util.concurrent.CopyOnWriteArrayList<>();
        
        for (int a : listA)
        {
            for (double b : listB)
            {
                for (int c : listC)
                {
                    for (int d : listD)
                    {
                        for (int e : listE)
                        {
                            for (int f : listF)
                            {
                                for (double g : listG)
                                {
                                    for (double h : listH)
                                    {
                                        for (double i : listI)
                                        {
                                            if (i < h) 
                                            {
                                                // 捕获任务中的异常，避免影响其他任务
                                                executor.submit(() -> {
                                                    try {
                                                        row.addAll(calculateRecords("C://CSV", a, b, c, d, e, f, g, h, i));
                                                    } catch (Exception ex) {
                                                        ex.printStackTrace();
                                                    }
                                                });
                                            }
                                        }
                                    }
                                }
                            }
                        }
                    }
                }
            }
        }
        
        executor.shutdown();
        try
        {
            // 等待所有任务完成，超时时间可根据实际情况调整
            if (!executor.awaitTermination(1, TimeUnit.HOURS)) {
                executor.shutdownNow();
            }
            System.out.println("TEST COMPLETE");
        } catch (InterruptedException e)
        {
            executor.shutdownNow();
            Thread.currentThread().interrupt();
            System.out.println("Thread execution interrupted.");
        }

        System.out.println(count.get());
    }
    
    public static ArrayList<Integer> list(int startPoint, int endPoint, int incrementPoint)
    {
        ArrayList<Integer> a = new ArrayList<>();
        for (int queryPoint = startPoint; queryPoint <= endPoint; queryPoint += incrementPoint)
        {
            a.add(queryPoint);
        }
        
        return a;
    }
    
    public static ArrayList<Double> list(double startPoint, double endPoint, double incrementPoint)
    {
        ArrayList<Double> a = new ArrayList<>();
        double tolerance = incrementPoint / 2.0;
        for (double queryPoint = startPoint; queryPoint <= endPoint + tolerance; queryPoint += incrementPoint)
        {
            double formatPoint = Double.parseDouble(decimalFormat.format(queryPoint));
            a.add(formatPoint);
        }
        return a;
    }
    
    public static List<String> calculateRecords(String directory, int a, double b, int c, int d, int e, int f, double g, double h, double i)
    {
        List<String> listC = new ArrayList<>();
        // 原子自增操作
        count.incrementAndGet();
        return listC;
    }
}

额外优化建议

批量提交任务：嵌套循环直接提交1890万次任务会占用大量内存，可将任务分批打包提交，减少内存消耗。
使用自定义线程池：Executors.newFixedThreadPool的队列是无界的，大量任务可能导致OOM，建议使用ThreadPoolExecutor自定义队列大小，配合拒绝策略。
避免外部变量捕获问题：原代码中lambda捕获的循环变量在Java 8+中是final或有效final的，但如果循环变量是可变类型需注意，当前代码中基本类型没问题。

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

Java ExecutorService多线程任务未全部执行问题求助

问题描述 #

问题根源 #

解决方案 #

1. 替换为线程安全的计数器 #

2. 使用线程安全的集合 #

3. 优化等待任务完成的逻辑 #

修改后的代码示例 #

额外优化建议 #