当工作线程发生异常时关闭 ThreadPoolExecutor

Question

工作线程发出正常关闭信号的最佳方式是什么？

我有一个固定大小的线程池，它通过一组连续的任务工作，每个任务持续不超过几秒钟。在正常操作期间，它运行良好，并且随着工作量的增加而突飞猛进。

我遇到的问题是在其中一个线程中引发异常时。如果发生这种情况，我想把整个事情搞砸，但一直无法正常工作。

目前的做法

我一直在使用的天真的方法是在“Supervisor”类中有一个静态方法，它使用标准的shutdown() 和awaitTermination() 方法关闭线程池。如果遇到问题，任何“Worker”类都会调用它。这样做不是传播异常，因为execute() 需要Runnable 并且run() 方法不能抛出异常。

这是一些伪代码：

// Finds work to do and passes them on to workers
class Supervisor {

  ThreadPoolExecutor exec;

  static main() {
    exec = new FixedThreadPool(...);
    forever {
      exec.execute(new Worker(next available task));
     }
  }

  static stopThreadPool() {
    exec.shutdown();
    if(!exec.awaitTermination(timeout_value)) {
      print "Timed out waiting on terminate"
    }
  }
}


class Worker {
  run() {
    try {
      // Work goes here
    } catch () {
      Supervisor.stopThreadPool()
    }
  }
}

我看到的效果是线程确实暂停了一段时间，但随后我看到了超时消息，它们都恢复了处理。这种模式一直持续到我手动关闭它。如果我在跳出循环后在 main 方法中调用stopThreadPool()，则关机会按预期正确发生。

这种方法显然是错误的，因为它不起作用，但它也感觉设计不正确。

重申这个问题：工作线程发出正常关闭信号的最佳方式是什么？

其他信息

我在 SO 上看到的问题有两种：

• “如何杀死线程池中的线程？”
• “我如何知道我的所有线程都已完成？”

那不是我想要的。他们似乎也只谈论一组有限的任务，而我正在处理连续的提要。

我已经阅读了一种使用exec.submit() 和Futures 的替代方法，它让主管类有责任检查一切是否正常，但我对它的了解还不够，无法知道它是否是一个更好的设计。例外情况是，嗯......例外，所以我不想在正常情况下不必要地增加工作/复杂性。

（小注：这是一个工作项目，还有其他人参与。为简单起见，我在问题中说“我”。）

Answer 1

您离正确的解决方案并不远，问题是您需要正确处理由shutdown调用引起的中断。所以你的线程的 run 方法应该是这样的：

run () {
  try {
    while (Thread.interrupted() == false) {
      doSomeWork();
    }
  } catch (Exception e) { 
    myExecutor.shutdown();
  }
}

请注意，我明确使用了 shutdown() 而没有 awaitTermination()，因为否则等待线程是阻止 Executor 正确终止的线程，因为一个线程仍在等待。完美的单线程死锁。 ;)

顺便说一句，中断检查是关于如何优雅地终止线程的提示：通过将 running 布尔值设置为 false 或通过中断来结束 run 方法，线程将在稍后死亡。

要检查您的所有线程是否都已终止（= 即将结束其运行方法），您可以将 CountDownLatch 用于简单情况，或将 CyclicBarrier/Phaser 类用于更复杂情况。

Answer 2

这里有两个问题：

1. 如果您打算仅对工作程序中的任何异常强制关闭，那么您是否使用 shutdown() 并等待对应项。只需使用shutdownNow强制它，你应该会很好。 shutdown 执行正常关机。

2. 当这样的事情发生时，试着打破你的 for 循环。最好的方法是在执行调用的 for 循环中尝试捕获。当工作人员发生异常时，抛出未经检查的异常并在 for 循环中捕获它。终止 for 循环并调用您的方法以强制关闭执行程序。这是一种更清洁的方法。或者，您也可以考虑在执行程序中使用考虑处理程序来执行此操作。