Runnable的ExecutorService，处理批量ArrayList未完成处理

Question

编辑：谢谢你，马克，对于那些有类似问题的人，我的问题是我首先创建了runnable类的线程实例，然后将线程提交给executorservice
它帮助我弄清楚，实际上，当我使用ExecutorService时，如果存在未捕获的异常；它不会通知您，它将取消该过程，而不会通知您。这就是为什么我得到了不完整的处理。

我有一个对象的ArrayList，我想对其进行多线程批量处理，但要限制在给定时间运行的线程数。我发现ExecutorService可以处理这个问题。但在测试它是否正在处理每个记录时，它似乎只处理我传递给它的对象的一小部分。

编辑：我已经删除了它的多线程部分，并且像正常一样处理对象而不使用执行器服务，在小批量（只有710）上，它工作正常；线程是否有可能完成得太快并且处理不正确？这意味着通常一次处理300k-800k记录；这就是为什么我想多线程它。

public void processContainerRecords(ArrayList<? extends ContainerRecord> records) {
    int cores = Runtime.getRuntime().availableProcessors();
    ExecutorService executor = Executors.newFixedThreadPool(cores);
    int batchSize = Settings.LOGIC_BATCH_SIZE;//100
    int batches = (int) Math.ceil((double) records.size() / (double) batchSize);

    ArrayList<Future<?>> threads = new ArrayList<Future<?>>();
    LogicProcessor newHandler = null;
    for (int startIndex = 0; startIndex < records.size(); startIndex += batchSize + 1) {
        if (records.size() < batchSize) {
            newHandler = new LogicProcessor(mainGUI, records.subList(startIndex, records.size()));
        } else {
            int bound = (startIndex + batchSize);
            if (bound > records.size()) {
                bound = records.size();
            }
            newHandler = new LogicProcessor(mainGUI, records.subList(startIndex, bound));
        }
        Thread newThread = new Thread(newHandler);
        Future<?> f = executor.submit(newThread);
        threads.add(f);
    }
    executor.shutdown();
    int completedThreads = 0;
    while (!executor.isTerminated()) {//monitors threads and waits until completion
        completedThreads = 0;
        for (Future<?> f : threads) {
            if (f.isDone()) {
                completedThreads++;
            }
        }
        //currentProgress = completedThreads;
    }

    for (ContainerRecord record : records) {//checks if each record has been processed
        System.out.println(record.getContainer() + ":" + record.isTouched());
    }
}

这是LogicProcessor类，它启动其线程实例

    private List<? extends ContainerRecord> archive;
private GUI mainGUI;

public LogicProcessor(GUI mainGUI, List<? extends ContainerRecord> records) {
    this.mainGUI = mainGUI;
    this.archive = records;
}

@Override
public void run() {
    handleLogic();
}

private void handleLogic() {
    Iterator iterator = archive.iterator();
    while (iterator.hasNext()) {
        ContainerRecord record = (ContainerRecord) iterator.next();
        record.touch();//sets a boolean in the object to validate if it has been processed yet.
    }
}

输出：在处理的710条记录（对象）中，有691条从未处理/触摸过，只有19条。

这是怎么回事？我尝试了很多方法，甚至制作了一个类LogicProcessor的数组，并将实例保留在数组中，以避免任何类型的GC删除实例。我不确定它为什么不处理这些记录。

Answer 1

我现在没有电脑来运行测试，但从查看您的代码来看，因此我的答案基于个人经验，不能将其视为代码审查，因为缺乏代码清晰度是错误的根源：）

>

不要将新线程提交到executor服务中。执行器的全部目的是向用户隐藏带有线程的单词。相反，您的逻辑处理器应该根据是否要返回值来实现可运行/可调用接口。

再次检查按批次划分的逻辑。如果您使用guava，它已经有了分区逻辑的实现。请参见本教程。我承认这更多是个人喜好，您的代码可能也不错，我还没有深入检查。

关闭方法和未来处理可能会简化。

调用shutdown方法会导致executor服务停止接受新的任务来执行，但它不会立即关闭服务，而是会等待它已经拥有的所有任务都将被执行。通常，这样的线程池是在应用程序生命周期的开始创建的，并且只要应用程序运行，线程池就一直处于活动状态。创建池非常昂贵，因为它分配线程。

如果希望保持池的打开状态，但确保所有任务都已完成，则可以使用循环来迭代未来。

所以我真的看不出有什么理由两者兼用。如果分配池只是为了提交一系列任务，那么调用shutdown就足够了。否则，您可以使用循环并将池视为全局对象，并在其他地方调用关闭，正如我上面解释的那样。