等待FutureTask上的取消（）

Question

我想取消从ThreadPoolExecutor获得的FutureTask，但我想确保线程池上的Callable已停止它的工作。

如果我调用FutureTask＃cancel（false）然后get（）（阻塞直到完成），我得到一个CancelledException。这个异常是立即抛出还是在任务停止执行之后抛出？

Answer 1

这个答案通过检查修复了阿列克谢的竞争条件和FooJBar代码该任务已被取消。 （在FutureTask.run检查状态并运行可调用期间，在取消和getWithJoin期间都可以成功完成调用，但仍然可以调用）但是，可调用仍将运行。）

我也决定不覆盖原始的取消，因为新的取消需要申报InterruptedException。新的取消摆脱其无用的返回值（因为true可能意味着“任务尚未开始”，“任务已经开始并且已经完成其大部分损害”，“任务已经开始并最终完成”）中的任何一个。 super.cancel的返回值的检查也已经过去了，所以如果新的cancel被多次从不同的线程调用，它们都将等待任务完成。

import java.util.concurrent.Callable; 
import java.util.concurrent.CancellationException; 
import java.util.concurrent.ExecutionException; 
import java.util.concurrent.Executors; 
import java.util.concurrent.FutureTask; 
import java.util.concurrent.Semaphore; 
import java.util.concurrent.TimeUnit; 
import java.util.concurrent.TimeoutException; 

/** 
* Based on: http://stackoverflow.com/questions/6040962/wait-for-cancel-on-futuretask 
* 
* @author Aleksandr Dubinsky 
*/ 
public class FixedFutureTask<T> extends FutureTask<T> { 

    /** 
     * Creates a {@code FutureTask} that will, upon running, execute the given {@code Runnable}, 
     * and arrange that {@code get} will return the given result on successful completion. 
     * 
     * @param runnable the runnable task 
     * @param result the result to return on successful completion. 
     *    If you don't need a particular result, consider using constructions of the form: 
     *    {@code Future<?> f = new FutureTask<Void>(runnable, null)} 
     * @throws NullPointerException if the runnable is null 
     */ 
     public 
    FixedFutureTask (Runnable runnable, T result) { 
      this (Executors.callable (runnable, result)); 
     } 

    /** 
     * Creates a {@code FutureTask} that will, upon running, execute the given {@code Callable}. 
     * 
     * @param callable the callable task 
     * @throws NullPointerException if the callable is null 
     */ 
     public 
    FixedFutureTask (Callable<T> callable) { 
      this (new MyCallable (callable)); 
     } 

     /** Some ugly code to work around the compiler's limitations on constructors */ 
     private 
    FixedFutureTask (MyCallable<T> myCallable) { 
      super (myCallable); 
      myCallable.task = this; 
     } 

    private final Semaphore semaphore = new Semaphore(1); 

    private static class MyCallable<T> implements Callable<T> 
    { 
     MyCallable (Callable<T> callable) { 
       this.callable = callable; 
      } 

     final Callable<T> callable; 
     FixedFutureTask<T> task; 

      @Override public T 
     call() throws Exception { 

       task.semaphore.acquire(); 
       try 
       { 
        if (task.isCancelled()) 
         return null; 

        return callable.call(); 
       } 
       finally 
       { 
        task.semaphore.release(); 
       } 
      } 
    } 

    /** 
     * Waits if necessary for the computation to complete or finish cancelling, and then retrieves its result, if available. 
     * 
     * @return the computed result 
     * @throws CancellationException if the computation was cancelled 
     * @throws ExecutionException if the computation threw an exception 
     * @throws InterruptedException if the current thread was interrupted while waiting 
     */ 
     @Override public T 
    get() throws InterruptedException, ExecutionException, CancellationException { 

      try 
      { 
       return super.get(); 
      } 
      catch (CancellationException e) 
      { 
       semaphore.acquire(); 
       semaphore.release(); 
       throw e; 
      } 
     } 

    /** 
     * Waits if necessary for at most the given time for the computation to complete or finish cancelling, and then retrieves its result, if available. 
     * 
     * @param timeout the maximum time to wait 
     * @param unit the time unit of the timeout argument 
     * @return the computed result 
     * @throws CancellationException if the computation was cancelled 
     * @throws ExecutionException if the computation threw an exception 
     * @throws InterruptedException if the current thread was interrupted while waiting 
     * @throws CancellationException 
     * @throws TimeoutException if the wait timed out 
     */ 
     @Override public T 
    get (long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, CancellationException, TimeoutException { 

      try 
      { 
       return super.get (timeout, unit); 
      } 
      catch (CancellationException e) 
      { 
       semaphore.acquire(); 
       semaphore.release(); 
       throw e; 
      } 
     } 

    /** 
     * Attempts to cancel execution of this task and waits for the task to complete if it has been started. 
     * If the task has not started when {@code cancelWithJoin} is called, this task should never run. 
     * If the task has already started, then the {@code mayInterruptIfRunning} parameter determines 
     * whether the thread executing this task should be interrupted in an attempt to stop the task. 
     * 
     * <p>After this method returns, subsequent calls to {@link #isDone} will 
     * always return {@code true}. Subsequent calls to {@link #isCancelled} 
     * will always return {@code true} if this method returned {@code true}. 
     * 
     * @param mayInterruptIfRunning {@code true} if the thread executing this task should be interrupted; 
     *        otherwise, in-progress tasks are allowed to complete 
     * @throws InterruptedException if the thread is interrupted 
     */ 
     public void 
    cancelAndWait (boolean mayInterruptIfRunning) throws InterruptedException { 

      super.cancel (mayInterruptIfRunning); 

      semaphore.acquire(); 
      semaphore.release(); 
     } 
} 

Answer 2

它一被取消就抛出。

有没有简单的方法知道它已经开始并完成。你可以创建一个可运行的包装来检查它的状态。

final AtomicInteger state = new AtomicInteger(); 
// in the runnable 
state.incrementAndGet(); 
try { 
    // do work 
} finally { 
    state.decrementAdnGet(); 
} 

Answer 3

是的，CancellationException立即抛出。您可以扩展FutureTask以添加get()方法的版本，等待Callable的线程完成。

public class ThreadWaitingFutureTask<T> extends FutureTask<T> { 

    private final Semaphore semaphore; 

    public ThreadWaitingFutureTask(Callable<T> callable) { 
     this(callable, new Semaphore(1)); 
    } 

    public T getWithJoin() throws InterruptedException, ExecutionException { 
     try { 
      return super.get(); 
     } 
     catch (CancellationException e) { 
      semaphore.acquire(); 
      semaphore.release(); 
      throw e; 
     } 
    } 

    private ThreadWaitingFutureTask(final Callable<T> callable, 
       final Semaphore semaphore) { 
     super(new Callable<T>() { 
      public T call() throws Exception { 
       semaphore.acquire(); 
       try { 
        return callable.call(); 
       } 
       finally { 
        semaphore.release(); 
       } 
      } 
     }); 
     this.semaphore = semaphore; 
    } 
} 

Answer 4

阿列克谢的例子效果很好。我写了一个变型与构造以一个Runnable（将返回null），并展示了如何直接阻断（加入）的取消（）：

public class FutureTaskCancelWaits<T> extends FutureTask<T> { 

    private final Semaphore semaphore; 

    public FutureTaskCancelWaits(Runnable runnable) { 
     this(Executors.callable(runnable, (T) null)); 
    } 

    public FutureTaskCancelWaits(Callable<T> callable) { 
     this(callable, new Semaphore(1)); 
    } 

    @Override 
    public boolean cancel(boolean mayInterruptIfRunning) { 
     // If the task was successfully cancelled, block here until call() returns 
     if (super.cancel(mayInterruptIfRunning)) { 
      try { 
       semaphore.acquire(); 
       // All is well 
       return true; 
      } catch (InterruptedException e) { 
       // Interrupted while waiting... 
      } finally { 
       semaphore.release(); 
      } 
     } 
     return false; 
    } 

    private FutureTaskCancelWaits(final Callable<T> callable, final Semaphore semaphore) { 
     super(new Callable<T>() { 
      public T call() throws Exception { 
       semaphore.acquire(); 
       try { 
        return callable.call(); 
       } finally { 
        semaphore.release(); 
       } 
      } 
     }); 
     this.semaphore = semaphore; 
    } 
}