基本上，当您使用ThreadPoolExecutor时，有几种不同的策略与异常处理：
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不会保留原始任务）。当我想要原始任务的访问状态时，如何处理异常？
首先，请参阅Handling Exceptions for ThreadPoolExecutor，了解有关afterExecute（）方法问题的更多背景信息。
ThreadPoolExecutor有
protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) ;
和
protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value);
当可调用时，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));
基本上，当您使用ThreadPoolExecutor时，有几种不同的策略与异常处理：
虽然你可以覆盖ThreadPoolExecutor.beforeExecute(...)，通过反射挖出你在那里的runnable，设置一个ThreadLocal，然后在afterExecute(...)中使用它，这真的感觉像一个黑客，并且非常依赖于TPE实现。
我会在try / catch日志错误包装中包装你的Runnable或Callable方法。因此，您可以使用以下内容向线程池添加内容：
threadPool.submit(new RunnableWrapper(myRunnable));
// or
threadPool.submit(new CallableWrapper(myCallable));
这些将具有try / catch / log机制，并且还可以访问Runnable进行状态评估。对我来说，其他任何东西似乎都是黑客。
你当然可以覆盖submit(...)方法来自己包装作业。这似乎更清洁。