java - 使用 ThreadPoolExecutor 使用 CompletableFuture 时在单个线程上设置超时

Question

我有一个使用 ThreadPoolExecutor 异步执行任务的程序。我使用 Java 8 中的 CompletableFutures 来安排这些任务，然后让它们由线程池中可用的线程执行。

我的代码是这样的:

public class ThreadTest {

    public void print(String m) {
        System.out.println(m);
    }

    public class One implements Callable<Integer> {

        public Integer call() throws Exception {
            print("One...");
            Thread.sleep(6000);
            print("One!!");
            return 100;
        }
    }

    public class Two implements Callable<String> {

        public String call() throws Exception {
            print("Two...");
            Thread.sleep(1000);
            print("Two!!");
            return "Done";
        }
    }

    @Test
    public void poolRun() throws InterruptedException, ExecutionException {
        int n = 3;
        // Build a fixed number of thread pool
        ExecutorService pool = Executors.newFixedThreadPool(n);

        CompletableFuture futureOne = CompletableFuture.runAsync(() -> new One());
        // Wait until One finishes it's task.
        CompletableFuture futureTwo = CompletableFuture.runAsync(() -> new One());
        // Wait until Two finishes it's task.
        CompletableFuture futureTwo = CompletableFuture.runAsync(() -> new Two());

        CompletableFuture.allOf(new CompletableFuture[]{futureOne, futureTwo, futureThree}).get();
        pool.shutdown();
    }
}

我需要为每个单独的线程设置超时，例如在 10 分钟时超时。我研究了 CompletableFuture 的 .get(TimeUnit timeUnit) 方法，但我不确定这是否会在线程池或单个线程本身上设置超时。

https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/CompletableFuture.html#get-long-java.util.concurrent.TimeUnit-

或者我应该改变我使用执行器服务来设置单个线程超时的方式吗？

谢谢!

Answer 1

CompletableFuture.get 不会停止运行您的任务的线程。只要您为结果指定，调用线程就会等待，如果超时，它将抛出异常。但是运行任务的线程将继续执行直到完成。这是潜在的现实:Java 不允许您随时任意终止任务。曾经有一段时间这是 API 的一部分，Thread 类上有 Thread.suspend/resume/stop 方法。这些已被弃用，因为无法知道挂起或停止的线程是否持有可能阻止其他线程执行的锁。因此，在任意时间和地点停止线程本质上是不安全的。您最终会在程序中陷入僵局。

参见此处:https://docs.oracle.com/javase/8/docs/technotes/guides/concurrency/threadPrimitiveDeprecation.html

相同的论点适用于您在并发包中找到的任何池和执行程序以及其他类。您不能随意停止线程或任务。

您必须将停止和完成的逻辑放入任务本身，或者您必须等到它完成。在这种情况下，您有一个运行一秒钟，一个运行 6 秒。您可以使用互斥锁和信号量之类的东西，可以使用并发包和 concurrent.locks 包中的许多东西，所有这些都有助于协调线程和传递有关它们所在位置的信息。

但是你不会在任何地方找到允许你在任何时间点杀死任意线程的方法，除了上面列出的那些之前被弃用的方法，以及那些你被鼓励远离的方法。

Future.cancel 将阻止任务启动，它会尝试中断正在运行的任务，但它所做的只是停止线程(通过使其抛出 InterruptedException)当前被可中断方法调用阻塞像 Thread.sleep()、Object.wait() 或 Condition.await()。如果您的任务正在执行任何其他操作，它将在完成或调用可中断方法调用之前不会停止。

这将适用于上面的代码，因为您正在调用 Thread.sleep。但是一旦您的任务开始工作，它就会像我描述的那样工作。