ExecutorService 在超时后中断任务
Posted
技术标签:
【中文标题】ExecutorService 在超时后中断任务【英文标题】：ExecutorService that interrupts tasks after a timeout 【发布时间】：2011-02-15 01:28:49 【问题描述】：我正在寻找可以提供超时的ExecutorService 实现。提交给 ExecutorService 的任务如果运行时间超过超时时间，就会被中断。实现这样的野兽并不是一项艰巨的任务，但我想知道是否有人知道现有的实现。
这是我根据下面的一些讨论得出的结论。有cmets吗？
import java.util.List;
import java.util.concurrent.*;
public class TimeoutThreadPoolExecutor extends ThreadPoolExecutor
private final long timeout;
private final TimeUnit timeoutUnit;
private final ScheduledExecutorService timeoutExecutor = Executors.newSingleThreadScheduledExecutor();
private final ConcurrentMap<Runnable, ScheduledFuture> runningTasks = new ConcurrentHashMap<Runnable, ScheduledFuture>();
public TimeoutThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, long timeout, TimeUnit timeoutUnit)
super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
this.timeout = timeout;
this.timeoutUnit = timeoutUnit;
public TimeoutThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, long timeout, TimeUnit timeoutUnit)
super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory);
this.timeout = timeout;
this.timeoutUnit = timeoutUnit;
public TimeoutThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, RejectedExecutionHandler handler, long timeout, TimeUnit timeoutUnit)
super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, handler);
this.timeout = timeout;
this.timeoutUnit = timeoutUnit;
public TimeoutThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler, long timeout, TimeUnit timeoutUnit)
super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler);
this.timeout = timeout;
this.timeoutUnit = timeoutUnit;
@Override
public void shutdown()
timeoutExecutor.shutdown();
super.shutdown();
@Override
public List<Runnable> shutdownNow()
timeoutExecutor.shutdownNow();
return super.shutdownNow();
@Override
protected void beforeExecute(Thread t, Runnable r)
if(timeout > 0)
final ScheduledFuture<?> scheduled = timeoutExecutor.schedule(new TimeoutTask(t), timeout, timeoutUnit);
runningTasks.put(r, scheduled);
@Override
protected void afterExecute(Runnable r, Throwable t)
ScheduledFuture timeoutTask = runningTasks.remove(r);
if(timeoutTask != null)
timeoutTask.cancel(false);
class TimeoutTask implements Runnable
private final Thread thread;
public TimeoutTask(Thread thread)
this.thread = thread;
@Override
public void run()
thread.interrupt();
【问题讨论】：
超时的“开始时间”是提交时间吗？或者任务开始执行的时间？ 好问题。当它开始执行时。大概是使用protected void beforeExecute(Thread t, Runnable r) 钩子。
@scompt.com 你还在使用这个解决方案还是已经被取代了
@PaulTaylor 我实施此解决方案的工作已被取代。 :-)
我确实需要这个，除了 a) 我需要我的主调度程序服务是一个具有单个服务线程的线程池，因为我的任务需要严格并发执行，并且 b) 我需要能够指定提交任务时每个任务的超时时间。我曾尝试以此为起点，但扩展了 ScheduledThreadPoolExecutor，但我看不到一种方法来获取指定的超时持续时间，该超时持续时间将在任务提交时指定到 beforeExecute 方法。任何建议都非常感谢！
【参考方案1】：
将任务包装在 FutureTask 中，您可以为 FutureTask 指定超时时间。看看我对这个问题的回答中的例子，
java native Process timeout
【讨论】：
我意识到使用java.util.concurrent 类有几种方法可以做到这一点，但我正在寻找ExecutorService 实现。
如果你说你希望你的 ExecutorService 隐藏从客户端代码中添加超时的事实，你可以实现你自己的 ExecutorService ，在执行它们之前用 FutureTask 包装交给它的每个可运行对象。 【参考方案2】：
您可以为此使用ScheduledExecutorService。首先，您只需提交一次即可立即开始并保留已创建的未来。之后，您可以提交一个新任务，该任务将在一段时间后取消保留的未来。
ScheduledExecutorService executor = Executors.newScheduledThreadPool(2);
final Future handler = executor.submit(new Callable() ... );
executor.schedule(new Runnable()
public void run()
handler.cancel();
, 10000, TimeUnit.MILLISECONDS);
这将执行您的处理程序（主要功能被中断）10 秒，然后将取消（即中断）该特定任务。
【讨论】：
有趣的想法，但是如果任务在超时之前完成（通常会这样）怎么办？我宁愿不要有大量的清理任务等待运行，只是为了发现他们分配的任务已经完成。当 Futures 完成删除清理任务时，需要有另一个线程来监控它们。 执行者只会安排一次取消。如果任务完成，那么取消是一个无操作并且工作继续不变。只需要一个额外的线程调度来取消任务和一个线程来运行它们。你可以有两个执行者，一个提交你的主要任务，一个取消它们。 确实如此，但如果超时时间为 5 小时并且在此期间执行了 10k 个任务怎么办。我想避免所有那些无操作占用内存并导致上下文切换。 @Scompt 不一定。将有 10k future.cancel() 调用，但是如果未来完成，那么取消将快速退出并且不会做任何不必要的工作。如果您不想要 10k 额外的取消调用，那么这可能行不通，但是任务完成后完成的工作量非常小。 @John W.：我刚刚意识到您的实施存在另一个问题。正如我之前评论的那样，我需要在任务开始执行时开始超时。我认为唯一的方法是使用beforeExecute 钩子。【参考方案3】：
不幸的是，该解决方案存在缺陷。 ScheduledThreadPoolExecutor 有一个 bug，在 this question 中也有报告：取消提交的任务不会完全释放与任务相关的内存资源；只有在任务到期时才会释放资源。
因此，如果您创建一个具有相当长的过期时间（典型用法）的TimeoutThreadPoolExecutor，并足够快地提交任务，您最终会填满内存 - 即使任务实际上已成功完成。
您可以通过以下（非常粗略的）测试程序看到问题：
public static void main(String[] args) throws InterruptedException
ExecutorService service = new TimeoutThreadPoolExecutor(1, 1, 10, TimeUnit.SECONDS,
new LinkedBlockingQueue<Runnable>(), 10, TimeUnit.MINUTES);
//ExecutorService service = Executors.newFixedThreadPool(1);
try
final AtomicInteger counter = new AtomicInteger();
for (long i = 0; i < 10000000; i++)
service.submit(new Runnable()
@Override
public void run()
counter.incrementAndGet();
);
if (i % 10000 == 0)
System.out.println(i + "/" + counter.get());
while (i > counter.get())
Thread.sleep(10);
finally
service.shutdown();
程序会耗尽可用内存，尽管它会等待生成的Runnables 完成。
我对此有过一段时间的思考，但不幸的是我无法想出一个好的解决方案。
编辑： 我发现这个问题被报告为JDK bug 6602600，并且似乎最近才得到修复。
【讨论】：
【参考方案4】：使用http://docs.oracle.com/javase/7/docs/api/java/util/concurrent/ExecutorService.html 中描述的ExecutorService.shutDownNow() 方法怎么样？这似乎是最简单的解决方案。
【讨论】：
因为它会停止所有计划任务，而不是问题要求的特定任务【参考方案5】：似乎问题不在于 JDK 错误 6602600（已于 2010 年 5 月 22 日解决），而在于 循环中的 sleep(10) 调用不正确。另外注意，主线程必须给出 通过调用 SLEEP(0) 在 外圈的每一个分支。 我认为最好使用 Thread.yield() 而不是 Thread.sleep(0)
之前问题代码的结果修正部分是这样的：
.......................
........................
Thread.yield();
if (i % 1000== 0)
System.out.println(i + "/" + counter.get()+ "/"+service.toString());
//
// while (i > counter.get())
// Thread.sleep(10);
//
它可以正常工作，外部计数器的数量高达 150 000 000 个测试圈。
【讨论】：
【参考方案6】：这个替代想法怎么样：
两个有两个执行者： 一个用于： 提交任务，不关心任务超时时间 将 Future 结果和应该结束的时间添加到内部结构中 用于执行内部作业，如果某些任务超时并且必须取消，则检查内部结构。小样本在这里：
public class AlternativeExecutorService
private final CopyOnWriteArrayList<ListenableFutureTask> futureQueue = new CopyOnWriteArrayList();
private final ScheduledThreadPoolExecutor scheduledExecutor = new ScheduledThreadPoolExecutor(1); // used for internal cleaning job
private final ListeningExecutorService threadExecutor = MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(5)); // used for
private ScheduledFuture scheduledFuture;
private static final long INTERNAL_JOB_CLEANUP_FREQUENCY = 1000L;
public AlternativeExecutorService()
scheduledFuture = scheduledExecutor.scheduleAtFixedRate(new TimeoutManagerJob(), 0, INTERNAL_JOB_CLEANUP_FREQUENCY, TimeUnit.MILLISECONDS);
public void pushTask(OwnTask task)
ListenableFuture<Void> future = threadExecutor.submit(task); // -> create your Callable
futureQueue.add(new ListenableFutureTask(future, task, getCurrentMillisecondsTime())); // -> store the time when the task should end
public void shutdownInternalScheduledExecutor()
scheduledFuture.cancel(true);
scheduledExecutor.shutdownNow();
long getCurrentMillisecondsTime()
return Calendar.getInstance().get(Calendar.MILLISECOND);
class ListenableFutureTask
private final ListenableFuture<Void> future;
private final OwnTask task;
private final long milliSecEndTime;
private ListenableFutureTask(ListenableFuture<Void> future, OwnTask task, long milliSecStartTime)
this.future = future;
this.task = task;
this.milliSecEndTime = milliSecStartTime + task.getTimeUnit().convert(task.getTimeoutDuration(), TimeUnit.MILLISECONDS);
ListenableFuture<Void> getFuture()
return future;
OwnTask getTask()
return task;
long getMilliSecEndTime()
return milliSecEndTime;
class TimeoutManagerJob implements Runnable
CopyOnWriteArrayList<ListenableFutureTask> getCopyOnWriteArrayList()
return futureQueue;
@Override
public void run()
long currentMileSecValue = getCurrentMillisecondsTime();
for (ListenableFutureTask futureTask : futureQueue)
consumeFuture(futureTask, currentMileSecValue);
private void consumeFuture(ListenableFutureTask futureTask, long currentMileSecValue)
ListenableFuture<Void> future = futureTask.getFuture();
boolean isTimeout = futureTask.getMilliSecEndTime() >= currentMileSecValue;
if (isTimeout)
if (!future.isDone())
future.cancel(true);
futureQueue.remove(futureTask);
class OwnTask implements Callable<Void>
private long timeoutDuration;
private TimeUnit timeUnit;
OwnTask(long timeoutDuration, TimeUnit timeUnit)
this.timeoutDuration = timeoutDuration;
this.timeUnit = timeUnit;
@Override
public Void call() throws Exception
// do logic
return null;
public long getTimeoutDuration()
return timeoutDuration;
public TimeUnit getTimeUnit()
return timeUnit;
【讨论】：
【参考方案7】：经过大量时间调查，
最后，我使用ExecutorService的invokeAll方法解决了这个问题。
这将在任务运行时严格中断任务。
这是示例
ExecutorService executorService = Executors.newCachedThreadPool();
try
List<Callable<Object>> callables = new ArrayList<>();
// Add your long time task (callable)
callables.add(new VaryLongTimeTask());
// Assign tasks for specific execution timeout (e.g. 2 sec)
List<Future<Object>> futures = executorService.invokeAll(callables, 2000, TimeUnit.MILLISECONDS);
for (Future<Object> future : futures)
// Getting result
catch (InterruptedException e)
e.printStackTrace();
executorService.shutdown();
优点是您也可以在同一ExecutorService 上提交ListenableFuture。
只需稍微修改第一行代码即可。
ListeningExecutorService executorService = MoreExecutors.listeningDecorator(Executors.newCachedThreadPool());
ListeningExecutorService 是 google guava 项目 (com.google.guava) 中 ExecutorService 的监听功能
【讨论】：
感谢您指出invokeAll。这很好用。对于任何考虑使用它的人，请注意：虽然invokeAll 返回一个Future 对象列表，但它实际上似乎是一个阻塞操作。
如果我们调用 future.get() 会不会阻塞。【参考方案8】：
使用 John W 的答案，我创建了一个实现，该实现在任务开始执行时正确开始超时。我什至为它写了一个单元测试:)
但是，它不适合我的需要，因为在调用Future.cancel() 时（即调用Thread.interrupt() 时），某些IO 操作不会中断。
Socket.connect 和 Socket.read 是调用 Thread.interrupt() 时可能不会中断的 IO 操作的一些示例（我怀疑大多数 IO 操作是在 java.io 中实现的）。当调用Thread.interrupt() 时，java.nio 中的所有 IO 操作都应该是可中断的。例如，SocketChannel.open 和 SocketChannel.read 就是这种情况。
无论如何，如果有人感兴趣，我为线程池执行器创建了一个要点，它允许任务超时（如果它们使用可中断操作...）：https://gist.github.com/amanteaux/64c54a913c1ae34ad7b86db109cbc0bf
【讨论】：
有趣的代码，我把它拉到我的系统中，想知道你是否有一些例子说明什么样的 IO 操作不会中断，所以我可以看看它是否会影响我的系统。谢谢！ @DuncanKrebs 我用不可中断 IO 的例子详细说明了我的答案：Socket.connect 和 Socket.read
myThread.interrupted() 不是正确的中断方法，因为它清除了中断标志。改用myThread.interrupt()，应该使用套接字
@DanielCuadra：谢谢，看来我打错了，因为Thread.interrupted() 无法中断线程。但是，Thread.interrupt() 不会中断 java.io 操作，它仅适用于 java.nio 操作。
我用interrupt()很多年了，它总是中断java.io操作（以及其他阻塞方法，如线程休眠、jdbc连接、blockingqueue take等）。也许你发现了一个错误的类或一些有错误的 JVM【参考方案9】：
检查这是否适合你，
public <T,S,K,V> ResponseObject<Collection<ResponseObject<T>>> runOnScheduler(ThreadPoolExecutor threadPoolExecutor,
int parallelismLevel, TimeUnit timeUnit, int timeToCompleteEachTask, Collection<S> collection,
Map<K,V> context, Task<T,S,K,V> someTask)
if(threadPoolExecutor==null)
return ResponseObject.<Collection<ResponseObject<T>>>builder().errorCode("500").errorMessage("threadPoolExecutor can not be null").build();
if(someTask==null)
return ResponseObject.<Collection<ResponseObject<T>>>builder().errorCode("500").errorMessage("Task can not be null").build();
if(CollectionUtils.isEmpty(collection))
return ResponseObject.<Collection<ResponseObject<T>>>builder().errorCode("500").errorMessage("input collection can not be empty").build();
LinkedBlockingQueue<Callable<T>> callableLinkedBlockingQueue = new LinkedBlockingQueue<>(collection.size());
collection.forEach(value ->
callableLinkedBlockingQueue.offer(()->someTask.perform(value,context)); //pass some values in callable. which can be anything.
);
LinkedBlockingQueue<Future<T>> futures = new LinkedBlockingQueue<>();
int count = 0;
while(count<parallelismLevel && count < callableLinkedBlockingQueue.size())
Future<T> f = threadPoolExecutor.submit(callableLinkedBlockingQueue.poll());
futures.offer(f);
count++;
Collection<ResponseObject<T>> responseCollection = new ArrayList<>();
while(futures.size()>0)
Future<T> future = futures.poll();
ResponseObject<T> responseObject = null;
try
T response = future.get(timeToCompleteEachTask, timeUnit);
responseObject = ResponseObject.<T>builder().data(response).build();
catch (InterruptedException e)
future.cancel(true);
catch (ExecutionException e)
future.cancel(true);
catch (TimeoutException e)
future.cancel(true);
finally
if (Objects.nonNull(responseObject))
responseCollection.add(responseObject);
futures.remove(future);//remove this
Callable<T> callable = getRemainingCallables(callableLinkedBlockingQueue);
if(null!=callable)
Future<T> f = threadPoolExecutor.submit(callable);
futures.add(f);
return ResponseObject.<Collection<ResponseObject<T>>>builder().data(responseCollection).build();
private <T> Callable<T> getRemainingCallables(LinkedBlockingQueue<Callable<T>> callableLinkedBlockingQueue)
if(callableLinkedBlockingQueue.size()>0)
return callableLinkedBlockingQueue.poll();
return null;
您可以限制调度程序使用的线程数量以及为任务设置超时。
【讨论】：
【参考方案10】：你可以使用 ExecutorService 提供的这个实现
invokeAll(Collection<? extends Callable<T>> tasks,long timeout, TimeUnit unit)
as
executor.invokeAll(Arrays.asList(task), 2 , TimeUnit.SECONDS);
但是，就我而言，我不能，因为 Arrays.asList 需要额外的 20 毫秒。
【讨论】：
【参考方案11】：这个怎么样？
final ExecutorService myExecutorService = ...;
// create CompletableFuture to get result/exception from runnable in specified timeout
final CompletableFuture<Object> timeoutFuture = new CompletableFuture<>();
// submit runnable and obtain cancellable Future from executor
final Future<?> cancellableFuture = myExecutorService.submit(() ->
try
Object result = myMethod(...);
timeoutFuture.complete(result);
catch (Exception e)
timeoutFuture.completeExceptionally(e);
);
// block the calling thread until "myMethod" will finish or time out (1 second)
try
Object result = timeoutFuture.get(1000, TimeUnit.MILLISECONDS);
// "myMethod" completed normally
catch (TimeoutException te)
// "myMethod" timed out
// ...
catch (ExecutionException ee)
// "myMethod" completed exceptionally - get cause
final Throwable cause = ee.getCause();
// ...
catch (InterruptedException ie)
// future interrupted
// ...
finally
// timeoutFuture.cancel(true); // CompletableFuture does not support cancellation
cancellableFuture.cancel(true); // Future supports cancellation
【讨论】：
以上是关于ExecutorService 在超时后中断任务的主要内容，如果未能解决你的问题，请参考以下文章
Java ExecutorService - 任务/可调用不取消/中断
ExecutorService 使用 invokeAll 和超时异常后可调用线程上的超时未终止
客户端发送请求, java后端有耗时任务，web服务器如nginx返回服务端超时信息，会中断后端请求处理吗
JDK 源码解析 —— Executors ExecutorService ThreadPoolExecutor 线程池