java - 使用 ThreadPoolExecutor 进行状态异常处理

Question

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

1. Thread.setUncaughtExceptionHandler()(和 Thread.getDefaultUncaughtExceptionHandler())

   Exception 被包装在 Future 中，所以 UncaughtExceptionHandler 永远不会被调用，所以不能使用它。

2. 设置线程工厂唯一相关的部分是新创建的线程上的 Thread.setUncaughtExceptionHandler()。但这不会有任何效果，请参阅第 1 页)。

3. 重写 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 不会保存原始任务)。

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

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);

当可调用时，可运行是您可以装饰的原始任务。这是基本策略。下面是使用 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));