ExecutorService 停止处理两个线程中的一个

Question

1. 我有一个需要在 for 循环中处理的 40000 条记录的列表。因为我有两个处理器系统。我创建了一个这样的固定线程池：

   int threads = Runtime.getRuntime().availableProcessors(); ExecutorService service = Executors.newFixedThreadPool(threads);

2. 并将我的ArrayList 分成两个子列表。对于每个子列表，我创建了一个Callable，它执行相同的功能（涉及迭代子列表并进行一些处理）并返回一个Future 对象。

3. 我使用executorServiceObject.submit(callable) 提交了这两个Callable，并将返回的Future 对象添加到我的Future 对象列表中

这是我的问题：

我写了System.Out.printLn("Processed Item" +item.id) // consider item as the name of reference variable for current iteration

一段时间以来一切都很好，我可以看到两个线程同时工作。但一段时间后，其中一个线程已停止处理。只有一个线程在运行。 （我知道这一点是因为我可以在控制台上看到不再打印给线程 2 的 id）。

有人知道这是怎么回事吗？我的意思是为什么 ExecutorService 停止运行第二个线程。

提前感谢您的帮助。

按照我之前应该做的那样添加示例代码：

public List<Output> processInputs(List<Input> inputs)
        throws InterruptedException, ExecutionException {

    int threads = Runtime.getRuntime().availableProcessors();
    ExecutorService service = Executors.newFixedThreadPool(threads);

    List<Future<Output>> futures = new ArrayList<Future<Output>>();
    for (final Input input : inputs) {
        Callable<Output> callable = new Callable<Output>() {
            public Output call() throws Exception {
                Output output = new Output();
                // process your input here and compute the output
                return output;
            }
        };
        futures.add(service.submit(callable));
    }

    service.shutdown();

    List<Output> outputs = new ArrayList<Output>();
    for (Future<Output> future : futures) {
        outputs.add(future.get());
    }
    return outputs;

Answer 1

一段时间以来一切都很好，我可以看到两个线程同时工作。但一段时间后，其中一个线程已停止处理。只有一个线程在运行。 （我知道这一点是因为我可以在控制台上看到分配给线程 2 的 id 不再被打印）。

我怀疑您的处理线程抛出了异常。 Future.get() 方法可以抛出 ExecutionException "if the computation threw an exception"。

// the following might throw an exception if the background job threw
outputs.add(future.get());

如果您的“处理您的输入”代码引发了 NPE、IOException 等，则该异常由 Callable 引发并存储在 Future 中，因此它可以由 get() 引发方法，但包装在 ExecutionException 中。这很有用，因此等待的线程可以获取并处理（记录等）后台线程抛出的异常。

除了让您的 processInputs(...) 方法将异常抛出给可能会丢失的调用者之外，我会在您的 while 循环中执行以下操作：

try {
   outputs.add(future.get());
} catch (InterruptedException ie) {
   // always a good pattern if the thread that is waiting was interrupted
   Thread.currentThread().interrupt();
   return;
} catch (ExecutionException ee) {
   // somehow log the error
   logger.error("Computation failed to process", ee);
   // now continue and get the next future in the list
}

如果您没有捕获并正确处理该ExecutionException，那么处理异常也会终止调用processInputs(...) 的线程。