带有 Java Streams 的 ExecutorService 给出了这个错误“java.util.concurrent.ExecutionException: java.util.ConcurrentModificationException”答案

【问题标题】：ExecutorService with Java Streams gives this error " java.util.concurrent.ExecutionException: java.util.ConcurrentModificationException"带有 Java Streams 的 ExecutorService 给出了这个错误“java.util.concurrent.ExecutionException: java.util.ConcurrentModificationException”
【发布时间】：2021-02-16 21:49:29
【问题描述】：

我的代码需要从 8 个“玩家”的 ArrayLists 中找到 N 个最佳组合，其中恰好有 1 个“玩家”。每个 ArrayList 介于 20 到 40 之间。这会导致大量迭代的运行时间很长。在尝试优化运行时间之前，代码可以正常运行，只是运行时间不理想。我认为最好的方法是使用 Executor Services 和 Java Streams，尽管我对这两者都是新手。这是我下面代码的最小化版本：

public static void main(String [] args){
//Lineup is an object which I store the combination of players in
ArrayList <Lineup> currBoard = new ArrayList<Lineup>();
final ExecutorService executorService = Executors.newFixedThreadPool(nThreads);
    String threadId = Thread.currentThread().getName();
    int tid = Integer.parseInt(threadId.substring(threadId.length() - 1));
    int threadWorkLoad = newPG.size() / 4;
    //int threadStart = (int)(tid - 1) * threadWorkLoad;
    final List<Future<?>> futures = new ArrayList<>();
    System.out.println(newPG.size());
    for(int i=0; i < nThreads; i++){        
        Future<?> future = executorService.submit(() -> {
                    //System.out.println(pg.getName());
                    
                    currBoard.addAll(parallelFunction(nThreads, amounts, newPG, newSG, newSF, newPF, newC, newG, newF, newL));
                    
                
                
        });
        futures.add(future);
    }
    executorService.shutdown();
    try{
        executorService.awaitTermination(5000, TimeUnit.MILLISECONDS);
    }catch(InterruptedException e){
        System.out.println("oof");
    }
    try {
        for (Future<?> future : futures) {
            future.get(); // do anything you need, e.g. isDone(), ...
        }
    } catch (InterruptedException | ExecutionException e) {
        e.printStackTrace();
    }
}

这是我遍历可能的阵容的函数

public ArrayList<Lineup> parallelFunction(int nThreads, int amounts, ArrayList<Player> newPG, ArrayList<Player> newSG, ArrayList<Player> newSF,
    ArrayList<Player> newPF, ArrayList<Player> newC, ArrayList<Player> newG, ArrayList<Player> newF,
        ArrayList<Player> newL) {

    ArrayList<Player> temp = new ArrayList<Player>();
    
    ArrayList<Lineup> threadBoard = new ArrayList<Lineup>();

    String threadId = Thread.currentThread().getName();
    int tid = Integer.parseInt(threadId.substring(threadId.length() - 1));
    int threadWorkLoad = (newPG.size() + nThreads - 1) / nThreads;
    int threadStart = (int)(tid - 1) * threadWorkLoad;
    int threadStop = Math.min(threadStart + threadWorkLoad, newPG.size());
    //ReentrantLock lock = new ReentrantLock();
    
    
    ArrayList<Player> tempPG = new ArrayList<Player>();

    for(int i = threadStart; i <= threadStop; i++){
        tempPG.add(newPG.get(i));
    }

    Stream<Player> pgStream = StreamSupport.stream(tempPG.spliterator(), true);
    pgStream.forEach(pg -> {

        System.out.println("TID: " + tid + ", PG: " + pg.getName() + ", iterations: " + counting);
        Stream<Player> sgStream = StreamSupport.stream(newSG.spliterator(), true);
        sgStream.forEach(sg -> {
        //System.out.println(tid + ", " + sg.getName());
        Stream<Player> sfStream = StreamSupport.stream(newSF.spliterator(), true);
        sfStream.forEach(sf -> {
            Stream<Player> pfStream = StreamSupport.stream(newPF.spliterator(), true);
            pfStream.forEach(pf -> {
                Stream<Player> cStream = StreamSupport.stream(newC.spliterator(), true);
                cStream.forEach(c -> {
                    Stream<Player> gStream = StreamSupport.stream(newG.spliterator(), true);
                    gStream.forEach(g -> {
                        if ((!(g.getName().equals(sg.getName()) || g.getName().equals(pg.getName())))
                                && ((pg.getSalary() + sg.getSalary() + sf.getSalary() + pf.getSalary()
                                        + c.getSalary() + 3000 + 3000 + 3000) < 50000)) {
                                Stream<Player> fStream = StreamSupport.stream(newF.spliterator(), true);
                                fStream.forEach(f -> {
                                if ((!(f.getName().equals(sf.getName()) || f.getName().equals(pf.getName())))
                                        && ((g.getSalary() + pg.getSalary() + sg.getSalary() + sf.getSalary()
                                                + pf.getSalary() + c.getSalary() + 3000 + 3000) < 50000)) {
                                        Stream<Player> pStream = StreamSupport.stream(newL.spliterator(), true);
                                        pStream.forEach(p -> {
                                        if (!(p.getName().equals(pg.getName()) || p.getName().equals(sg.getName())
                                                || p.getName().equals(sf.getName())
                                                || p.getName().equals(pf.getName())
                                                || p.getName().equals(c.getName())
                                                || p.getName().equals(g.getName())
                                                || p.getName().equals(f.getName()))) {
                                            double currScore = 0;
                                            double totalSal = 0;
                                            totalSal = pg.getSalary() + sg.getSalary() + sf.getSalary()
                                                    + pf.getSalary() + c.getSalary() + f.getSalary() + g.getSalary()
                                                    + p.getSalary();
                                            currScore = pg.getProjection() + sg.getProjection() + sf.getProjection()
                                                    + pf.getProjection() + c.getProjection() + f.getProjection()
                                                    + g.getProjection() + p.getProjection();
                                            if (totalSal <= 50000.0) {
                                                counting += 1;
                                                temp.clear();
                                                temp.add(pg);
                                                temp.add(sg);
                                                temp.add(sf);
                                                temp.add(pf);
                                                temp.add(c);
                                                temp.add(g);
                                                temp.add(f);
                                                temp.add(p);
                                                ArrayList<Lineup> tempBoard = new ArrayList<Lineup>();

                                                Lineup aLine = new Lineup(temp, currScore);
                                                
                                                if (threadBoard.size() < amounts) {
                                                    if (!alreadyIn(threadBoard, temp)) {
                                                        
                                                        threadBoard.add(aLine);
                                                    }


                                                } else if (currScore > threadBoard.get(amounts - 1).totalScore) {
                                                    // Collections.sort(currBoard, Lineup.TotComp);
                                                    if (!alreadyIn(threadBoard, temp)) {
                                                        for (int i = 0; i < threadBoard.size() - 1; i++) {
                                                            tempBoard.add(threadBoard.get(i));
                                                        }
                                                        threadBoard.clear();
                                                        threadBoard.addAll(tempBoard);
                                                        tempBoard.clear();
                                                        threadBoard.add(aLine);
                                                    }
                                                }
                                                Collections.sort(threadBoard, Lineup.TotComp);
                                            }

                                            // temp.clear();
                                        } // p if
                                    });
                                    pStream.close();
                                } // f if
                            });
                            fStream.close();
                        } // g if
                    });
                    gStream.close();
                });
                cStream.close();
            });
            pfStream.close();
        });
        sfStream.close();
    });
    sgStream.close();
});
pgStream.close();







return threadBoard;
}

这会在大量迭代后导致此错误，java.util.concurrent.ExecutionException: java.util.ConcurrentModificationException: java.util.ConcurrentModificationException

是我在 executorService 或流上做错了什么导致了这种情况。

或者有没有人有任何建议来提高这段代码的速度，因为它在崩溃之前仍然不能令人满意。

编辑

主要：

ArrayList<Lineup> currBoard = new ArrayList<Lineup>();
    int nThreads = Runtime.getRuntime().availableProcessors();

    final ExecutorService executorService = Executors.newFixedThreadPool(nThreads);
    String threadId = Thread.currentThread().getName();
    int tid = Integer.parseInt(threadId.substring(threadId.length() - 1));
    int threadWorkLoad = newPG.size() / 4;
    //int threadStart = (int)(tid - 1) * threadWorkLoad;
    final List<Future<ThreadBoard>> futures = new ArrayList<Future<ThreadBoard>>();
    
    for(int i=0; i < nThreads; i++){
        Future<ThreadBoard> future = executorService.submit(() -> {
                    //System.out.println(pg.getName());
                return parallelFunction(nThreads, amounts, newPG, newSG, newSF, newPF, newC, newG, newF, newL); 
        });
        futures.add(future);
    }
    executorService.shutdown();
    try{
        executorService.awaitTermination(5000, TimeUnit.MILLISECONDS);
    }catch(InterruptedException e){
        System.out.println("oof");
    }
    try {
        for (Future<ThreadBoard> future : futures) {
            currBoard.addAll(future.get().getCurrentLineup()); // do anything you need, e.g. isDone(), ...
        }
    } catch (InterruptedException | ExecutionException e) {
        e.printStackTrace();
    }

并行函数：

public ThreadBoard parallelFunction(int nThreads, int amounts, ArrayList<Player> newPG, ArrayList<Player> newSG, ArrayList<Player> newSF,
    ArrayList<Player> newPF, ArrayList<Player> newC, ArrayList<Player> newG, ArrayList<Player> newF,
        ArrayList<Player> newL) {

    
    ThreadBoard threadBoard = new ThreadBoard();

    String threadId = Thread.currentThread().getName();
    int tid = Integer.parseInt(threadId.substring(threadId.length() - 1));
    int threadWorkLoad = (newPG.size() + nThreads - 1) / nThreads;
    int threadStart = (int)(tid - 1) * threadWorkLoad;
    int threadStop = Math.min(threadStart + threadWorkLoad, newPG.size());
    //ReentrantLock lock = new ReentrantLock();
    
    
    
    ArrayList<Player> tempPG = new ArrayList<Player>();

    for(int i = threadStart; i <= threadStop; i++){
        tempPG.add(newPG.get(i));
    }

    Stream<Player> pgStream = StreamSupport.stream(tempPG.spliterator(), false);
    pgStream.forEach(pg -> {

        System.out.println("TID: " + tid + ", PG: " + pg.getName() + ", iterations: " + counting);
        Stream<Player> sgStream = StreamSupport.stream(newSG.spliterator(), false);
        sgStream.forEach(sg -> {
        //System.out.println(tid + ", " + sg.getName());
        Stream<Player> sfStream = StreamSupport.stream(newSF.spliterator(), false);
        sfStream.forEach(sf -> {
            Stream<Player> pfStream = StreamSupport.stream(newPF.spliterator(), false);
            pfStream.forEach(pf -> {
                Stream<Player> cStream = StreamSupport.stream(newC.spliterator(), false);
                cStream.forEach(c -> {
                    Stream<Player> gStream = StreamSupport.stream(newG.spliterator(), false);
                    gStream.forEach(g -> {
                        if ((!(g.getName().equals(sg.getName()) || g.getName().equals(pg.getName())))
                                && ((pg.getSalary() + sg.getSalary() + sf.getSalary() + pf.getSalary()
                                        + c.getSalary() + 3000 + 3000 + 3000) < 50000)) {
                                Stream<Player> fStream = StreamSupport.stream(newF.spliterator(), false);
                                fStream.forEach(f -> {
                                if ((!(f.getName().equals(sf.getName()) || f.getName().equals(pf.getName())))
                                        && ((g.getSalary() + pg.getSalary() + sg.getSalary() + sf.getSalary()
                                                + pf.getSalary() + c.getSalary() + 3000 + 3000) < 50000)) {
                                        Stream<Player> pStream = StreamSupport.stream(newL.spliterator(), false);
                                        pStream.forEach(p -> {
                                        if (!(p.getName().equals(pg.getName()) || p.getName().equals(sg.getName())
                                                || p.getName().equals(sf.getName())
                                                || p.getName().equals(pf.getName())
                                                || p.getName().equals(c.getName())
                                                || p.getName().equals(g.getName())
                                                || p.getName().equals(f.getName()))) {
                                            double currScore = 0;
                                            double totalSal = 0;
                                            totalSal = pg.getSalary() + sg.getSalary() + sf.getSalary()
                                                    + pf.getSalary() + c.getSalary() + f.getSalary() + g.getSalary()
                                                    + p.getSalary();
                                            currScore = pg.getProjection() + sg.getProjection() + sf.getProjection()
                                                    + pf.getProjection() + c.getProjection() + f.getProjection()
                                                    + g.getProjection() + p.getProjection();
                                            if (totalSal <= 50000.0) {
                                                counting += 1;
                                                ArrayList<Player> temp = new ArrayList<Player>();
                                                temp.clear();
                                                temp.add(pg);
                                                temp.add(sg);
                                                temp.add(sf);
                                                temp.add(pf);
                                                temp.add(c);
                                                temp.add(g);
                                                temp.add(f);
                                                temp.add(p);

                                                Lineup aLine = new Lineup(temp, currScore);
                                                threadBoard.addLineup(aLine, currScore, amounts, temp);
                                                
                                                // if (threadBoard.size() < amounts) {
                                                //  //if (!alreadyIn(threadBoard, temp)) {
                                                        
                                                //      threadBoard.add(aLine);
                                                //  //}


                                                // } else if (currScore > threadBoard.get(amounts - 1).totalScore) {
                                                //  // Collections.sort(currBoard, Lineup.TotComp);
                                                //  //if (!alreadyIn(threadBoard, temp)) {
                                                //      for (int i = 0; i < threadBoard.size() - 1; i++) {
                                                //          tempBoard.add(threadBoard.get(i));
                                                //      }
                                                //      threadBoard.clear();
                                                //      threadBoard.addAll(tempBoard);
                                                //      tempBoard.clear();
                                                //      threadBoard.add(aLine);
                                                //  //}
                                                // }
                                                //Collections.sort(threadBoard.getCurrentLineup(), Lineup.TotComp);
                                            }

                                            //temp.clear();
                                        } // p if
                                    });
                                    pStream.close();
                                } // f if
                            });
                            fStream.close();
                        } // g if
                    });
                    gStream.close();
                });
                cStream.close();
            });
            pfStream.close();
        });
        sfStream.close();
    });
    sgStream.close();
});
pgStream.close();







return threadBoard;
}

ThreadBoard 类：

public class ThreadBoard {
private List<Lineup> lineups;

public ThreadBoard(){
    this.lineups = new ArrayList<>();
}

public synchronized void addLineup(Lineup aLine, double currScore, int amounts, ArrayList<Player> temp) {
    if (lineups.size() < amounts) {
        if (!this.alreadyIn(lineups, temp)) {
            lineups.add(aLine);
        }
    } else if (currScore > lineups.get(amounts - 1).totalScore) {
        // Collections.sort(currBoard, Lineup.TotComp);
        if (!this.alreadyIn(lineups, temp)) {
            lineups.set(amounts - 1, aLine);
        }
    }
    Collections.sort(lineups, Lineup.TotComp);
}

public List<Lineup> getCurrentLineup() {
    return lineups;
}

public boolean alreadyIn(List<Lineup> board, ArrayList<Player> temp) {
    boolean found = false;
    for (int i = 0; i < board.size(); i++) {
        
        ArrayList <Player> check = board.get(i).getL();

        boolean pg = (check.contains(temp.get(0)));
        boolean sg = (check.contains(temp.get(1)));
        boolean pf = (check.contains(temp.get(2)));
        boolean sf = (check.contains(temp.get(3)));
        boolean c = (check.contains(temp.get(4)));
        boolean g = (check.contains(temp.get(5)));
        boolean f = (check.contains(temp.get(6)));
        boolean all = (check.contains(temp.get(7)));

        if (pg && sg && pf && sf && c && g && f && all) {
            found = true;
        }
    }
    

    return found;
}

}

【问题讨论】：

您不能多次使用Stream。根据Package java.util.stream Description：消耗品。流的元素在流的生命周期中只被访问一次。与迭代器一样，必须生成一个新流来重新访问源的相同元素。
@ThomasKläger 我在每次调用之前将 Stream 声明移到了右侧，它确实消除了错误。我之前在那里有过声明，不完全确定我为什么要移动它们
@ThomasKläger 但是我确实进行了编辑以描述代码存在的另一个问题
抱歉，您的代码太复杂了，我无法理解。但据我注意到您正在使用并行流来处理数据，这意味着来自parallelFunction() 的ArrayList<Player> temp 和ArrayList<Lineup> threadBoard 以及来自主要方法的ArrayList <Lineup> currBoard 会产生问题，因为简单的ArrayList 是不是线程安全的数据结构，但您正在从多个线程访问这些 ArrayList。
@ThomasKläger 感谢您的输入，据您所知，会在 parallelFunction() 帮助中创建 ArrayLists 的副本，还是会出现同样的线程安全问题？

标签： java optimization java-stream executorservice java.util.concurrent

【解决方案1】：

ArrayList<Lineup> threadBoard 的问题在于它用于收集、排序和限制来自不同并行流的结果。恕我直言，ArrayList 不是这些任务的正确数据结构。我会创建一个单独的类来执行这些步骤：

public class ThreadBoard {
    private List<Lineup> currentLineup = new ArrayList<>();
    
    public void addLineup(Lineup aLine, double currScore) {
        if (currentLineup.size() < amounts) {
            if (!alreadyIn(currentLineup, temp)) {
                currentLineup.add(aLine);
            }
        } else if (currScore > currentLineup.get(amounts - 1).totalScore) {
            // Collections.sort(currBoard, Lineup.TotComp);
            if (!alreadyIn(currentLineup, temp)) {
                currentLineup.set(amounts - 1, aLine);
                Collections.sort(currentLineup, Lineup.TotComp);
        }
    }

    public List<Lineup> getCurrentLineup() {
        return currentLineup;
    }
}

为了使这个线程安全，您可以将addLineup() 方法声明为同步。

但是我不确定所有这些并行流是否有助于提高性能（因为您已经将完整的作业分成并行处理的块）。

也许将完整的作业分成更多、更小的块，然后只使用顺序流来处理一个块会更有意义。最后，您的计算机只有有限数量的内核（可能是 4 个，可能是 16 个），并且将 CPU 绑定的作业分成更多的部分（并且通过使用嵌套的并行流，您正在这样做）比内核数量不多感觉。

【讨论】：

非常感谢您的持续帮助，但是我更改的代码似乎死锁/只是无法运行。没有出现错误。我对你的建议有什么不正确的地方吗？我在正在编辑的帖子中添加了更新的代码。