Time limit on individual threads with ExecutorService

I have an ExecutorService managing a number of Callables. The tasks that the Callables run are mostly black box transformations and number crunching. Under certain conditions, the data being transformed will oscillate and the thread will take over an hour to finish. For comparison, most threads are completed within a minute.

It's been deteremined that the dat开发者_C百科a from the long-running threads is not relevent. I would like to interrupt any thread that runs longer than a certain amount of time. What would the best way to do this?

Use a ScheduleExecutorService to schedule a task to taskFuture.cancel(true) the long running task when the timeout is reached. If the task finishes before then it won't be cancelled.

ExecutorService service = Executors.newFixedThreadPool(N);
ScheduledExecutorService canceller = Executors.newSingleThreadScheduledExecutor();

public <T> Future<T> executeTask(Callable<T> c, long timeoutMS){
   final Future<T> future = service.submit(c);
   canceller.schedule(new Callable<Void>(){
       public Void call(){
          future.cancel(true);
          return null;
       }
    }, timeoutMS, TimeUnit.MILLI_SECONDS);
   return future;
}

You could cancel the future etc as in the other answers, but you need to make sure that your threads which are "number crunching" can handle the interrupt and terminate gracefully. You say that this a black box operation - how certain are you that the interrupted status of the thread is being checked actively within the black box? If it isn't, you can't cancel it with an interrupt. The black box needs to be written with interruption in mind.

The best way for you to do this would be to introduce one more Executor. You can use a ScheduledExecutorService to cancel all long running tasks for example:

ExecutorService service = Executors.newFixedThreadPool(N);

ScheduledExecutorService canceller = Executors.newScheduledThreadPool(1);

public void executeTask(Callable<?> c){
   final Future<?> future = service.submit(c);
   canceller.schedule(new Runnable(){
       public void run(){
          future.cancel(true);
       }
    }, SECONDS_UNTIL_TIMEOUT, TimeUnit.SECONDS);
}

You could get a list of your corresponding Futures (that are created when you submit a Callable) together with its startup time.

Another task could then check every minute if there are some task running for more than a defined time and if so, invoke cancel(true) in the Future. Done futures would be removed off the list.

You can use this method

<T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks,
                              long timeout,
                              TimeUnit unit)
                          throws InterruptedException

and set the maximum timeout to one minute. If your thread takes more than that it is just aborted.

The problem is you can not kill/stop/suspend the java thread. All are deprecated.

future.cancel will only considered before worker thread

So if the thread is stuck with any code execution and doesn't have the logic to consider the thread interruption, then execution will still continue.

This is even true for executorService.shutdown();.

The documentation says.

There are no guarantees beyond best-effort attempts to stop processing actively executing tasks. For example, typical implementations will cancel via Thread. interrupt, so any task that fails to respond to interrupts may never terminate.

So the only solution is to log such a situation and fix the inherent issue.

    Future<?> future = executor.submit(task);
    try {
        future.get(15, TimeUnit.MINUTES);
    } catch (InterruptedException e) {
        // retry waiting. iterative approach not shown here
    } catch (ExecutionException e) {
        // your task exploded
    } catch (TimeoutException e) {
        // The task is running longer than usual
        // Log it for future analysis

        //If the interrupt signal is handled 
        //OR the task is stuck in the queue itself for a long time
        //then worth calling below 
        future.cancel(true);
    }