java - 如何管理 ThreadPoolExecutor 中的池线程终止？

Question

引用实践中的并发:

To set an UncaughtExceptionHandler for pool threads, provide a ThreadFactory to the ThreadPoolExecutor constructor. (As with all thread manipulation, only the thread's owner should change its UncaughtExceptionHandler.) The standard thread pools allow an uncaught task exception to terminate the pool thread, but use a try-finally block to be notified when this happens so the thread can be replaced. Without an uncaught exception handler or other failure notification mechanism, tasks can appear to fail silently, which can be very confusing. If you want to be notified when a task fails due to an exception so that you can take some task-specific recovery action, either wrap the task with a Runnable or Callable that catches the exception or override the afterExecute hook in THReadPoolExecutor.

本书没有提供任何如何实现这一目标的示例。

你能展示一下这个技巧吗？

附注

我尝试编写代码示例:

ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(2, 4, 1, TimeUnit.MINUTES, new LinkedBlockingQueue<>(), new ThreadFactory() {
        @Override
        public Thread newThread(Runnable r) {
            Thread thread = new Thread();
            thread.setUncaughtExceptionHandler(new Thread.UncaughtExceptionHandler() {
                @Override
                public void uncaughtException(Thread t, Throwable e) {
                    e.printStackTrace();
                }
            });
            System.out.println("created thread with id " + Thread.currentThread().getId());
            return thread;
        }
    });
    Runnable runnable = new Runnable() {
        @Override
        public void run() {
            System.out.println(Thread.currentThread().getId() + " started");
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println(Thread.currentThread().getId() + " termination");
            throw new RuntimeException();

        }
    };
    threadPoolExecutor.submit(runnable);
    threadPoolExecutor.submit(runnable);
    threadPoolExecutor.submit(runnable);
    threadPoolExecutor.submit(runnable);
}

但是这个输出

created thread with id 1
created thread with id 1

永远。

Answer 1

您的代码中有几个错误。首先，您必须通过Runnable传递给您的ThreadFactory到创建的线程，否则，您将留下一个损坏的线程，不执行任何任务。其次，您正在打印 Thread.currentThread() 的 id在你的工厂中，这显然不是新创建的线程。这就是为什么它打印 id 1两次。

尽管如此，修复这些错误后，您将不会看到未捕获的异常。原因是，虽然线程池执行器的行为如所述，submit方法包装你的 Runnable进入FutureTask它自己捕获所有异常，以便在调用get()时可以报告它们.

此时，我们要记住，有一个任意Runnable的队列s，不一定是 FutureTask实例。因此，未捕获的异常仍然是可能的，例如当我们直接通过 execute 对可运行对象进行排队时.

所以固定的例子:

public class Test {
    public static void main(String[] args) throws InterruptedException {
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(2, 4, 1,
                TimeUnit.MINUTES, new LinkedBlockingQueue<>(), r -> {
                    Thread thread = new Thread(r);
                    thread.setUncaughtExceptionHandler((t, e) -> {
                        synchronized(System.out) {
                            System.out.println("Uncaught exception in "+t.getId());
                            e.printStackTrace(System.out);
                        }
                    });
                    System.out.println("created thread with id " + thread.getId());
                    return thread;
        });
        Runnable runnable = () -> {
            System.out.println(Thread.currentThread().getId() + " started");
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println(Thread.currentThread().getId() + " termination");
            throw new RuntimeException();
            };
        threadPoolExecutor.execute(runnable);
        threadPoolExecutor.execute(runnable);
        threadPoolExecutor.execute(runnable);
        threadPoolExecutor.execute(runnable);
        threadPoolExecutor.shutdown();
    }
}

将打印

created thread with id 11
created thread with id 12
11 started
12 started
11 termination
12 termination
created thread with id 14
created thread with id 15
Uncaught exception in 11
java.lang.RuntimeException
    at smp.Test.lambda$main$2(Test.java:28)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
    at java.lang.Thread.run(Thread.java:745)
15 started
14 started
Uncaught exception in 12
java.lang.RuntimeException
    at smp.Test.lambda$main$2(Test.java:28)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
    at java.lang.Thread.run(Thread.java:745)
14 termination
15 termination
Uncaught exception in 14
java.lang.RuntimeException
    at smp.Test.lambda$main$2(Test.java:28)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
    at java.lang.Thread.run(Thread.java:745)
Uncaught exception in 15
java.lang.RuntimeException
    at smp.Test.lambda$main$2(Test.java:28)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
    at java.lang.Thread.run(Thread.java:745)

当然，消息的数量和顺序可能会有所不同。

如引用中所述，您还可以通过覆盖 afterExecute 了解异常。 :

ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(2, 4, 1,
    TimeUnit.MINUTES, new LinkedBlockingQueue<>()) {
        @Override protected void afterExecute(Runnable r, Throwable t) {
            if(t!=null) synchronized(System.out) {
               System.out.println("Uncaught exception in "+Thread.currentThread().getId());
               t.printStackTrace(System.out);
            }
        }
    };

尽管在我的测试中，默认的未捕获异常处理程序也打印了堆栈跟踪......