使用 ThreadPoolExecutor 进行状态异常处理

Question

基本上，当您使用 ThreadPoolExecutor 时，有几种不同的异常处理策略：

Thread.setUncaughtExceptionHandler()（和Thread.getDefaultUncaughtExceptionHandler()）

Exception 被包装在 Future 中，所以 UncaughtExceptionHandler 永远不会被调用，所以不能使用它。
设置ThreadFactory 唯一相关的部分是 Thread.setUncaughtExceptionHandler() on newley created thread。但这不会有任何效果，请参阅第 1 页）。

覆盖 ThreadPoolExecutor.afterExecute()

protected void afterExecute(Runnable r, Throwable t) {
    super.afterExecute(r, t);
    if (t == null && r instanceof Future<?>) {
        try {
              Object result = ((Future<?>) r).get();
            } catch (CancellationException ce) {
              t = ce;
            } catch (ExecutionException ee) {
              t = ee.getCause();
            } catch (InterruptedException ie) {
              Thread.currentThread().interrupt(); // ignore/reset
            }
    }
    if (t != null){
        logger.error("ThreadPoolExecutor.afterExecute", t);
    }
}

这种方法几乎行得通。如果您的异常处理是无状态的，即您不需要访问原始 Runnable/Callable 任务的状态，这没关系。在有状态的情况下，您无法访问原始任务（即使反射也无济于事，因为上面的 Runnable 不会保留原始任务）。

当我想要访问原始任务的状态时，如何处理异常？

Answer 1

首先看 Handling Exceptions for ThreadPoolExecutor 有关 afterExecute() 方法问题的更多背景信息。

ThreadPoolExecutor 有

protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) ;

和

protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value);

when callable, runnable 是您可以装饰的原始任务。这是基本策略。下面是使用 Spring 的工作代码（为了清楚起见，我删除了注释）：

package org.springframework.scheduling.concurrent;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.RunnableFuture;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import org.springframework.core.task.TaskDecorator;
import org.springframework.scheduling.concurrent.ConcurrentTaskExecutor;
import org.springframework.scheduling.concurrent.ExecutorConfigurationSupport;
import org.springframework.scheduling.concurrent.ThreadPoolExecutorFactoryBean;
import org.springframework.util.Assert;

public class ThreadPoolTaskExecutor extends ExecutorConfigurationSupport implements AsyncListenableTaskExecutor, SchedulingTaskExecutor {

    private final Object poolSizeMonitor = new Object();

    private int corePoolSize = 1;

    private int maxPoolSize = Integer.MAX_VALUE;

    private int keepAliveSeconds = 60;

    private int queueCapacity = Integer.MAX_VALUE;

    private boolean allowCoreThreadTimeOut = false;
    //fix
    private CallableTransform callableTransform;

    private ThreadPoolExecutor threadPoolExecutor;


    public void setCorePoolSize(int corePoolSize) {
        synchronized (this.poolSizeMonitor) {
            this.corePoolSize = corePoolSize;
            if (this.threadPoolExecutor != null) {
                this.threadPoolExecutor.setCorePoolSize(corePoolSize);
            }
        }
    }

    public int getCorePoolSize() {
        synchronized (this.poolSizeMonitor) {
            return this.corePoolSize;
        }
    }

    public void setMaxPoolSize(int maxPoolSize) {
        synchronized (this.poolSizeMonitor) {
            this.maxPoolSize = maxPoolSize;
            if (this.threadPoolExecutor != null) {
                this.threadPoolExecutor.setMaximumPoolSize(maxPoolSize);
            }
        }
    }

    public int getMaxPoolSize() {
        synchronized (this.poolSizeMonitor) {
            return this.maxPoolSize;
        }
    }

    public void setKeepAliveSeconds(int keepAliveSeconds) {
        synchronized (this.poolSizeMonitor) {
            this.keepAliveSeconds = keepAliveSeconds;
            if (this.threadPoolExecutor != null) {
                this.threadPoolExecutor.setKeepAliveTime(keepAliveSeconds, TimeUnit.SECONDS);
            }
        }
    }

    public int getKeepAliveSeconds() {
        synchronized (this.poolSizeMonitor) {
            return this.keepAliveSeconds;
        }
    }

    public void setQueueCapacity(int queueCapacity) {
        this.queueCapacity = queueCapacity;
    }

    public void setAllowCoreThreadTimeOut(boolean allowCoreThreadTimeOut) {
        this.allowCoreThreadTimeOut = allowCoreThreadTimeOut;
    }


    //fix
    public void setCallableDecorator(CallableDecorator callableDecorator) {
        Assert.isNull(this.callableTransform, "You can' call setCallableDecorator() and setTaskDecorator() more than once");
        this.callableTransform = new CallableTransform(){

            @Override
            public Callable<?> decorate(Object originalTask) {
                Callable<?> ret = callableDecorator.decorate((Callable<?>)originalTask);
                return ret;
            }

            @Override
            public boolean isCallable(){
                return true;
            }
    };
}

    //fix
    public void setTaskDecorator(TaskDecorator taskDecorator) {
        Assert.isNull(this.callableTransform, "You can' call setCallableDecorator() and setTaskDecorator() more than once");
        this.callableTransform =  new CallableTransform(){

            @Override
            public Callable<?> decorate(Object originalTask) {
                Callable<?> ret= Executors.callable(taskDecorator.decorate((Runnable)originalTask));
                return ret;
            }

            @Override
            public boolean isCallable(){
                return false;
            }
        };
    }


    @Override
    protected ExecutorService initializeExecutor(
            ThreadFactory threadFactory, RejectedExecutionHandler rejectedExecutionHandler) {

        BlockingQueue<Runnable> queue = createQueue(this.queueCapacity);

        ThreadPoolExecutor executor;

        //fix
        if (this.callableTransform != null) {
            executor = new ThreadPoolExecutor(
                    this.corePoolSize, this.maxPoolSize, this.keepAliveSeconds, TimeUnit.SECONDS,
                    queue, threadFactory, rejectedExecutionHandler) {

                @Override
                protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {
                    if(callableTransform==null){
                        return super.newTaskFor(callable);
                    }

                    Callable<?> wrapedCallable = null;

                    boolean isCallable = callableTransform.isCallable();
                    if(isCallable){
                        wrapedCallable = callableTransform.decorate(callable);
                    } else {
                        //callableTransform accepts Runnable, but we have Callable
                        throw new IllegalStateException("You use TaskDecorator, but submit Callable");
                    }

                    @SuppressWarnings("unchecked")
                    Callable<T> param = (Callable<T>)wrapedCallable;
                    return super.newTaskFor(param);
                }

                @Override
                protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) {
                    if(callableTransform==null){
                        return super.newTaskFor(runnable, value);
                    }

                    Callable<?> wrapedCallable = null;

                    boolean isRunnable = callableTransform.isRunnable();
                    if(isRunnable){
                        wrapedCallable = callableTransform.decorate(runnable);
                    } else {
                        //callableTransform accepts Callable, but we have Runnable
                        throw new IllegalStateException("You use CallableDecorator, but execute Runnable");
                    }

                    @SuppressWarnings("unchecked")
                    Callable<T> param = (Callable<T>)wrapedCallable;
                    return super.newTaskFor(param);
                }


            };

        } else {
            executor = new ThreadPoolExecutor(
                    this.corePoolSize, this.maxPoolSize, this.keepAliveSeconds, TimeUnit.SECONDS,
                    queue, threadFactory, rejectedExecutionHandler);

        }

        if (this.allowCoreThreadTimeOut) {
            executor.allowCoreThreadTimeOut(true);
        }

        this.threadPoolExecutor = executor;
        return executor;
    }


    protected BlockingQueue<Runnable> createQueue(int queueCapacity) {
        if (queueCapacity > 0) {
            return new LinkedBlockingQueue<>(queueCapacity);
        }
        else {
            return new SynchronousQueue<>();
        }
    }

    public ThreadPoolExecutor getThreadPoolExecutor() throws IllegalStateException {
        Assert.state(this.threadPoolExecutor != null, "ThreadPoolTaskExecutor not initialized");
        return this.threadPoolExecutor;
    }

    public int getPoolSize() {
        if (this.threadPoolExecutor == null) {
            // Not initialized yet: assume core pool size.
            return this.corePoolSize;
        }
        return this.threadPoolExecutor.getPoolSize();
    }

    public int getActiveCount() {
        if (this.threadPoolExecutor == null) {
            // Not initialized yet: assume no active threads.
            return 0;
        }
        return this.threadPoolExecutor.getActiveCount();
    }


    @FunctionalInterface
    public interface CallableDecorator {
        <V> Callable<V> decorate(Callable<V> task);
    }

    @FunctionalInterface
    static interface CallableTransform {
        Callable<?> decorate(Object originalTask);

        default boolean isCallable(){
            return true;
        }

        default boolean isRunnable(){
            return !isCallable();
        }
    }



    //rest of the code execute/submit override
    //...

    @Override
    public boolean prefersShortLivedTasks() {
        return true;
    }

}

用法示例如下：

    ThreadPoolTaskExecutor threadPoolFactory = new ThreadPoolTaskExecutor();
    threadPoolFactory.setCorePoolSize(4);
    threadPoolFactory.setMaxPoolSize(4);
    threadPoolFactory.setKeepAliveSeconds(0);


    CallableDecorator decorator = new CallableDecorator(){

        @Override
        public <T> Callable<T> decorate(Callable<T> task) {
            return () -> {
                try {
                    return task.call();
                }
                catch (Throwable e) {
                    synchronized (executor) {
                        if (!((MyRunnable) task).failSilent){   //note use of state of original Task
                            log.error("Execution Failure!", e);
                        }
                    }
                    throw e;
                }
            };
        }
    };
    threadPoolFactory.setCallableDecorator(decorator);

    threadPoolFactory.initialize();
    executor = threadPoolFactory.getThreadPoolExecutor();

还有：

    executor.submit(new MyCallable(true));

Answer 2

Basically, there is couple different strategies with exception handling when you use ThreadPoolExecutor:

尽管您可以覆盖 ThreadPoolExecutor.beforeExecute(...)，但通过反射挖掘出其中的可运行对象，设置 ThreadLocal 然后在 afterExecute(...) 中使用它，这真的感觉像是 hack 和非常依赖于 TPE 实现。

我会改为将您的 Runnable 或 Callable 方法包装在 try/catch 日志错误包装器中。这样您就可以使用以下内容将内容添加到线程池中：

threadPool.submit(new RunnableWrapper(myRunnable));
// or
threadPool.submit(new CallableWrapper(myCallable));

这些将具有 try/catch/log 机制，并且还可以访问 Runnable 进行状态评估。其他任何东西对我来说都像是黑客。

您当然可以重写 submit(...) 方法来自己包装作业。这看起来干净多了。

使用 ThreadPoolExecutor 进行状态异常处理

Stateful exception handling with ThreadPoolExecutor

java

multithreading

exception-handling

threadpool

executorservice