擴展FutureTask，如何處理取消

Question

我已經從java.util.concurrent擴展FutureTask來提供回調以跟蹤提交給ExecutorService的任務的執行情況。現在

public class StatusTask<V> extends FutureTask<V> { 

    private final ITaskStatusHandler<V> statusHandler; 

    public StatusTask(Callable<V> callable, ITaskStatusHandler<V> statusHandler){ 
     super(callable); 
     if (statusHandler == null) 
      throw new NullPointerException("statusHandler cannot be null"); 
     this.statusHandler = statusHandler; 
     statusHandler.TaskCreated(this); 
    } 

    @Override 
    public void run() { 
     statusHandler.TaskRunning(this); 
     super.run(); 
    } 

    @Override 
    protected void done() { 
     super.done(); 
     statusHandler.TaskCompleted(this); 
    } 

} 

，我所看到的是，如果該任務被提交，但最終排隊，我cancel(true);的任務 - run()方法仍然被調用 - 和FutureTask.run()（可能）會檢查任務被取消，並沒有按不打電話給包裝的可回收物品。

我應該如

@Override 
public void run() { 
    if(!isCancelled()) { 
    statusHandler.TaskRunning(this); 
    super.run(); 
    } 
} 

還是應該叫super.run()？這兩種方法似乎都容易受到檢查取消和做某件事情之間的競爭條件的影響..任何想法都會受到讚賞。

Answer 1

你說得對，有一個比賽中出現。 FutureTask#done()最多會被稱爲一次，因此如果任務在通過RunnableFuture#run()運行之前已被取消，您將錯過FutureTask#done()的呼叫。

你有沒有考慮過一種更簡單的方法，總是發出一組對稱呼叫到ITaskStatusHandler#taskRunning()和ITaskStatusHandler#taskCompleted()，像這樣？

@Override 
public void run() { 
    statusHandler.TaskRunning(this); 
    try { 
    super.run(); 
    finally { 
    statusHandler.TaskCompleted(this); 
    } 
} 

一旦RunnableFuture#run()被調用，這是真的，你在運行的任務，或者至少試圖運行。一旦FutureTask#run()完成，您的任務不再運行。恰巧在取消的情況下，轉換是（幾乎）立即的。

試圖避免調用ITaskStatusHandler#taskRunning()如果內部Callable或Runnable永遠不會被調用FutureTask#run()將要求您建立Callable或Runnable和FutureTask派生類型本身之間的一些共享的結構，這樣，當第一次叫你的內在功能您設置了一些標誌，表明外部FutureTask衍生類型可以作爲鎖存器觀察到，表明是的，在取消之前，函數確實開始運行。但是，到那時，你必須承諾致電ITaskStatusHandler#taskRunning()，所以區別不是很有用。

我一直在掙扎了類似的設計問題最近，彼時我重寫FutureTask#run()方法操作後定居在對稱和之前。

Answer 2

你的問題是你的未來任務仍然被執行後，你已經調用取消他們，對吧？

將任務提交給執行程序服務後，應該由執行程序管理。 （如果你喜歡，你仍然可以取消一個任務。）你應該用executor shutdownNow method取消執行。 （這將調用所有提交任務的取消方法。）關機仍然會執行所有提交的任務。

執行者不會「知道」任務被取消。它將獨立於未來任務的內部狀態調用該方法。

最簡單的方法是按照原樣使用Executor框架並編寫Callable裝飾器。

class CallableDecorator{ 

    CallableDecorator(Decorated decorated){ 
    ... 
    } 

    setTask(FutureTask task){ 
    statusHandler.taskCreated(task); 
    } 

    void call(){ 
    try{ 
     statusHandler.taskRunning(task); 
     decorated.call(); 
    }finally{ 
     statusHandler.taskCompleted(task); 
    } 
    } 
} 

唯一的問題是，任務不能在裝飾器的構造函數中。 （這是未來任務構造函數的一個參數。）要打破這個循環，您必須使用setter或代理工作來處理構造函數注入。也許這根本不需要回調，你可以說：statusHandler.callableStarted(decorated)。

根據您的要求，您可能必須發出異常和中斷信號。

基本實現：

class CallableDecorator<T> implements Callable<T> { 

    private final Callable<T> decorated; 
    CallableDecorator(Callable<T> decorated){ 
     this.decorated = decorated; 
    } 

    @Override public T call() throws Exception { 
     out.println("before " + currentThread()); 
     try { 
      return decorated.call(); 
     }catch(InterruptedException e){ 
      out.println("interupted " + currentThread()); 
      throw e; 
     } 
     finally { 
      out.println("after " + currentThread()); 
     } 
    } 
} 

ExecutorService executor = newFixedThreadPool(1); 
Future<Long> normal = executor.submit(new CallableDecorator<Long>(
     new Callable<Long>() { 
      @Override 
      public Long call() throws Exception { 
       return System.currentTimeMillis(); 
      } 
     })); 
out.println(normal.get()); 

Future<Long> blocking = executor.submit(new CallableDecorator<Long>(
     new Callable<Long>() { 
      @Override 
      public Long call() throws Exception { 
       sleep(MINUTES.toMillis(2)); // blocking call 
       return null; 
      } 
     })); 

sleep(SECONDS.toMillis(1)); 
blocking.cancel(true); // or executor.shutdownNow(); 

輸出：

before Thread[pool-1-thread-1,5,main] 
after Thread[pool-1-thread-1,5,main] 
1259347519104 
before Thread[pool-1-thread-1,5,main] 
interupted Thread[pool-1-thread-1,5,main] 
after Thread[pool-1-thread-1,5,main]