java - ThreadPoolExecutor 忽略错误 (OutOfMemoryError)

Question

我们正在编写一个应用程序，它使用ThreadPoolExecutor的自定义扩展来处理一堆Runnable。在开发应用程序时，我们遇到了 Runnable 类中发生的 OutOfMemoryError 错误。但是，ThreadPoolExecutor 并没有像我们期望的那样调用 afterExecute()，而是继续运行。

我已将代码缩减为一个小型的、可重复的应用程序。

ThreadPoolTest.java:

package org.codechimp.threadpool;

import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class ThreadPoolTest {

    /**
     * @param args
     */
    public static void main(String[] args) {
        LinkedBlockingQueue<Runnable> threadQueue = new LinkedBlockingQueue<Runnable>();
        ThreadPoolExecutor executor = new MyThreadPoolExecutor(10, Integer.MAX_VALUE, 2000, TimeUnit.MILLISECONDS, threadQueue);

        // Create a bunch of Runnables now...
        for (int i = 0; i < 10000; i++) {
            executor.execute(new MyRunnable(i));
            if (i % 100 == 0) {
                System.out.println("Queued " + i + " Runnables");
            }
            if (i % 1000 == 0) {
                try {
                    Thread.sleep(1);
                } catch (InterruptedException ignored) { }
            }
        }
        System.out.println("Done queing the Runnables.");

        while (!executor.isTerminated() && !executor.isTerminating() && executor.getQueue().size() > 0) {
            System.out.println(executor.getQueue().size() + " Runnables in the queue.");
            try {
                Thread.sleep(1000);
            } catch (InterruptedException ignored) { }
        }
        System.out.println("Runnable queue has stopped processing.");

        executor.shutdown();
        try {
            executor.awaitTermination(10, TimeUnit.MINUTES);
        } catch (InterruptedException ignored) { }
        System.out.println("Shutdown completed...exiting");
    }
}

MyThreadPoolExecutor.java:

package org.codechimp.threadpool;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class MyThreadPoolExecutor extends ThreadPoolExecutor {
    public MyThreadPoolExecutor(int corePoolSize, int maximumPoolSize,
            long keepAliveTime, TimeUnit unit,
            BlockingQueue<Runnable> workQueue,
            RejectedExecutionHandler handler) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, handler);
    }

    public MyThreadPoolExecutor(int corePoolSize, int maximumPoolSize,
            long keepAliveTime, TimeUnit unit,
            BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory,
            RejectedExecutionHandler handler) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
                threadFactory, handler);
    }

    public MyThreadPoolExecutor(int corePoolSize, int maximumPoolSize,
            long keepAliveTime, TimeUnit unit,
            BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
                threadFactory);
    }

    public MyThreadPoolExecutor(int corePoolSize, int maximumPoolSize,
            long keepAliveTime, TimeUnit unit,
            BlockingQueue<Runnable> workQueue) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
    }

    @Override
    protected void afterExecute(Runnable r, Throwable t) {
        if (t != null) {
            System.out.println("We got an error: " + t);
            int remaining = this.shutdownNow().size();
            System.out.println(remaining + " Runnables left in the queue");
        }
    }
}

MyRunnable.java

package org.codechimp.threadpool;

import org.apache.commons.lang.math.RandomUtils;

public class MyRunnable implements Runnable {
    private int runnableNumber;
    private int counter = 0;

    public MyRunnable(int number) {
        this.runnableNumber = number;
    }

    /**
     * Simple runnable that generates an OutOfMemoryError after the 1000th loop
     */
    public void run() {
        while (counter < 1000) {
            counter++;
            if (counter % 100 == 0) {
                System.out.println("\tRunnable " + this.runnableNumber + " reached " + this.counter + ".");
            }

            if (this.runnableNumber == 15 && this.counter % 200 == 0) {
                throw new OutOfMemoryError("This is a test!");
            }

            int wait = RandomUtils.nextInt(100);
            if (wait > 0) {
                try {
                    //System.out.println("\tRunnable " + this.runnableNumber + " waiting " + wait + ".");
                    Thread.sleep(wait);
                } catch (InterruptedException e) {
                    throw new RuntimeException("Thread.sleep() failed", e);
                }
            }
        }
    }
}

这是一个 super 简单的示例，它将在 MyThreadPoolExectuor 中创建 10k 个 MyRunnable，并随着计数器的增加打印出一些状态消息。第 16 个可运行对象(编号 15，从 0 开始计数)将在第 200 个增量时抛出 OutOfMemoryError。如果 MyThreadPoolExecutor 的 afterExecute() 获得 Throwable，它将打印一条消息。当我在 Java 6 和 Java 7 下运行它时，它永远不会打印此消息。

我们如何使应用程序在所有 Throwable 上保释？我们真的想在那时终止事情。

更新/编辑:

我正在更新此内容，因为我的要求似乎有些困惑。 我知道错误被打印出来。问题不在于 ThreadPoolExecutor 没有打印 OutOfMemoryError，问题在于，如标题和我提出的问题中所述最后，afterExecute() 不会因 Error 被调用。由于 OutOfMemoryError 是 Error 的子类，这意味着当发生错误时我无法停止代码。

再次，请阅读代码试图执行的操作。它肯定是在尝试“处理”该错误。事实上，它试图通过调用 shutdownNow() 来停止 ThreadPoolExecutor。但是，由于正在生成的 Error 被以某种方式抑制，因此该代码不会被执行。结果，应用程序只是继续运行，忽略了它到处喷出 OOME 的事实。

再问一个问题:

How do I detect a Runnable has recieved an Error (OutOfMemoryError or other) and terminate the ThreadPoolExecutor so the app will simply stop in it's tracks?

Answer 1

我通过代码简单说明一下，如果查看runWorker()实现，发现错误仍然是在afterExecute()中发送的。因此，如果使用 afterExecute 钩子(Hook)，则无法跳过错误。请引用https://stackoverflow.com/a/50480190/5620851 .

final void runWorker(Worker arg0) {
    Thread arg1 = Thread.currentThread();
    Runnable arg2 = arg0.firstTask;
    arg0.firstTask = null;
    arg0.unlock();
    boolean arg3 = true;

    try {
        while (arg2 != null || (arg2 = this.getTask()) != null) {
            arg0.lock();
            ..
            try {
                this.beforeExecute(arg1, arg2);
                Object arg4 = null;

                try {
                    arg2.run();
                } catch (RuntimeException arg27) {
                    arg4 = arg27;
                    throw arg27;
                } catch (Error arg28) {
                    arg4 = arg28;
                    throw arg28;
                } catch (Throwable arg29) {
                    arg4 = arg29;
                    throw new Error(arg29);
                } finally {
                    this.afterExecute(arg2, (Throwable) arg4);
                }
            } finally {
                arg2 = null;
                ++arg0.completedTasks;
                arg0.unlock();
            }
        }

        arg3 = false;
    } finally {
        this.processWorkerExit(arg0, arg3);
    }

}