如何让可调用任务提交到 ExecutorService 超时

Question

我将 Callable 任务（使用 submit()）提交给 ExecutionService 的实现。偶尔我似乎遇到了死锁，但无法在发生死锁的位置或原因工作，所以我想为任务设置超时，我不清楚该怎么做？

我应该

1. 在提交任务并设置超时时，在 ExecutionService 上使用 invokeAny() 而不是 submit()。我使用 submit() 一次提交许多任务，我可以像这样使用 invokeAny() 吗，我很谨慎，因为我不明白为什么没有需要超时的 submit() 方法。
2. 在我的 ExecutorService 的构造函数中修改 keepAliveTime（但我认为这是在做其他事情
3. 修改我的实际 Callable 实现，但如果它已死锁，则无法自行解除死锁。

选项 1 似乎是唯一可行的解​​决方案，但它是吗？

更多详情

我认为可能值得更详细地解释该过程的工作原理，以防它有助于解决问题。

可调用任务 P1 启动并在文件夹上工作，其中的所有文件和文件夹开始将歌曲分组，它在 ExecutorService ES1 中运行，并且只有一个 P1 实例提交给 ES1。

我们还有其他三个 Callable 类：P2、P3 和 P4 - 每一个都有自己关联的 Executor Service，ES2、ES3、Es4）。一旦 P1 创建了一个组，它就会向关联的 ES 提交一个任务，该组作为数据传递，即它可以将 P2 的实例提交给 E2、P3 或 P3 或 P4 到 E4，它选择哪一个取决于分组，P2、P3 和 P4 都做不同的事情。

假设它已经提交了一个 P2 的实例，P2 将通过将 P3 提交给 E3 或将 P4 提交给 E4 来完成处理。它是一种单向管道 P3 只能提交给 P4，一旦所有任务都提交给 P4 并且 P4 完成了所有处理完成的任务。

我们通过构造 ES1、ES2、ES3 和 ES4 来完成处理，向 P1 提交任务，然后在每个 ExecutorService 上依次调用 shutdown()，这样，直到 P1 完成提交所有组后，shutdown()才会返回，然后调用ES2 上的 shutdown() 在 ES2 清除 P2 任务等队列之前不会返回。

偶尔一切都会停止我假设某个进程正在阻止其他进程继续，所以在这一点上，我想要一种取消需要太长时间的进程以便其他进程可以继续的方法，这比它无限期挂起要好得多

答案更新

我尝试按照建议使用 invokeAny()，它有点工作。 如果 P1 向 E2 提交 P2 的一个实例，然后它会在完成之前等待，这没关系，因为当使用 submit() 时，它只是返回任何方式，它不会进一步处理，但有两个问题：

1. 每个 ExecutorService 使用一个 500 的有界队列，其想法是如果 P2 比 P1 慢得多，我们就不会继续在 ES2 上堆叠东西并最终耗尽内存。所以现在 P1 在他们调用的任务完成之前不会完成队列实际上更小，因为它们不仅包含等待 ES2 上的插槽完成的任务，而且它们包含已经提交给 ES2 但正在等待它的任务完成。

2. 管道是链式的，因此如果我们对从 P1 提交的任务以及从 P2、P3 和 P4 提交的任务使用 invokeAny，那么当一个任务从 P1 提交到 P2 时，它不会返回，直到从 E4 完成后续处理！

Answer 1

我做了一些修改后，@PhilipWhitehouse 的评论就起作用了。

总之，创建一个封装了 ScheduledExecutorPool 的自定义 ThreadPool，以便在提交任务时设置超时。

这里有完整的解决方案：

How can I make shutdown work properly with this custom ExecutorService?

Answer 2

解决死锁情况总是可取的。但要做出替代方案，您可以使用可调用的未来对象来设置超时持续时间。

查看单个可调用实例的以下解决方案。同样，您可以使用未来调用列表来实现可调用的数量。它使用 future.get(...) 方法，我们在其中设置超时。如果可调用对象在设置的超时之前未完成执行，则线程将完成其执行。

import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

public class ThreadTimeOut{
   public static void main(String[] args) throws Exception {
    ExecutorService executor = Executors.newSingleThreadExecutor();
    Future<String> future = executor.submit(new Task());

    try {
        System.out.println("Started..");
        System.out.println(future.get(3, TimeUnit.SECONDS));
        System.out.println("Finished!");
    } catch (TimeoutException e) {
        System.out.println("Terminated!");
    }

    executor.shutdownNow();
    }
}

class Task implements Callable<String> {
   @Override
   public String call() throws Exception {
       Thread.sleep(4000); // Just to demo a long running task of 4 seconds.
       return "Ready!";
   }
}

Answer 3

不确定这是否会对您有所帮助，但我之前通过这个秘诀设法快速轻松地找到了死锁：

• 在 Eclipse 中调试
• 重现“挂起”
• 选择服务器实例并单击 Eclipse 调试器中的“暂停”按钮。这将暂停所有线程。
• 现在向下滚动线程列表。死锁的线程标记为红色。每个线程显示它持有的锁和它正在等待的锁。
• 利润！

Answer 4

你可以使用番石榴的MoreExecutorsListeningExecutorService。它不会神奇地解决您的问题，但可以提供一些帮助：

1) 您可以为通过invokeAll 调用的每个Callable 设置超时。如果一个 callable 没有在给定的时间内完成，它应该被杀死。

2) 您可以创建所有ListenableFutures 的全局映射，其中每个人都会在创建时注册一个标志并在完成时清除该标志。这样您就可以知道哪些未来没有帮助缩小问题范围。

Answer 5

我认为最好的方法是找到并修复死锁。你不能只是杀死线程。您应该在任务中实现某种任务取消，并要求此任务取消它正在执行的操作。但如果它陷入僵局，你将无能为力。You can use jsconsole to detect a deadlock

使用带有超时的 invokeAny 会阻塞线程，直到其中一项提交的任务成功完成或超时到期。如果超时到期，您将收到 TimeoutException，但您的任务将运行。 ExecutorService 将通过 Future.cancel(true) 要求他们取消。在内部它会中断线程，将任务的线程 isInterrupted 标志设置为 true。如果您在任务中使用阻塞方法来响应中断，它们将抛出 Interrupted 异常。否则，您应该检查任务中的中断状态，如果它返回 true，则相应地响应它。如果没有阻塞方法或检查中断状态，则本次取消无效。