2

我正在研究 CompletionService 类，我发现将提交队列与完成队列分离非常有用。

但我也错过了一种轮询/取消任务的方法（这可以被认为以某种方式完成）。它可以以某种方式轻松完成吗？

Future<String> task1Future =  completionService.submit(myCallableTask1);
Future<String> task2Future =  completionService.submit(myCallableTask2);
task2Future.cancel();
Future<String> lastComplTaskFuture = completionService.take(); 
                        //Seems to return only the completed or 
                        //interrupted tasks, not those cancelled (task2)

编辑：在检查了一些答案后，我意识到发生了什么。CompletitionService 以与提交的作业相同的顺序返回。如果您运行作业 a、b 和 c；在 a 工作时取消 b 和 c；最后轮询completionService，直到a taks 终止才会通知b 和c 的取消。另外，我意识到，如果您关闭执行程序而不是取消单个任务，那么仍在队列中的那些任务不会到达完成服务，即使它们的取消在活动任务尚未结束时完成。

EDIT2：好的，我添加了一个完整的测试用例

import java.util.concurrent.*;

public class CompletionTest {

    public static void main(String[] args){
        CompletionService<String> completion = 
                new ExecutorCompletionService<String>(Executors.newSingleThreadExecutor()); 
        Future<String> aFuture =
                 completion.submit(() -> {Thread.sleep(10000); return "A";});
        completion.submit(() -> {return "B";}).cancel(true);
        completion.submit(() -> {return "C";}).cancel(true);
        completion.submit(() -> {return "D";}).cancel(true);

        long arbitraryTime = System.currentTimeMillis();
        while (true){
            String result = getNextResult(completion);
            System.out.println(result);

            if (System.currentTimeMillis() > arbitraryTime+5000){
                aFuture.cancel(true);
            }
         }
    }

    public static String getNextResult(CompletionService<String> service){
        try {
            String taskId = service.take().get();
            return "'"+taskId+"' job completed successfully.";
        } 
        catch (InterruptedException e ) { return "Unknown job was interrupted/stopped while waiting (no thread.wait() or infinite loop in the job so I guess this should not be possible)."; } 
        catch (CancellationException e) { return "Unknown job was cancelled."; }  
        catch (ExecutionException e) { 
            return "Unknown job returned with the following internal exception while running: " + e.getCause().getMessage();
        }
    }
}

我期望输出如下：

Unknown job was cancelled.
Unknown job was cancelled.
Unknown job was cancelled.
'A' job completed successfully.

但取而代之的是：

'A' job completed successfully.
Unknown job was cancelled.
Unknown job was cancelled.
Unknown job was cancelled.

我什至尝试使用 PriorityBlockingQueue 作为 CompletionService 的队列，Comparator.<Future<String>,Boolean>comparing(Future::isCancelled).reversed()但它也没有工作（我猜如果元素改变状态它不会求助）

4

1 回答 1

1

您将任务提交到具有单个线程的线程池。

您提交的第一个任务将消耗该线程（短时间休眠）。

其他人将在ExecutorService. 取消相应的期货不会从该队列中删除相应的任务。它只是改变了Future.

任务将保留在队列中，直到有线程空闲来处理它们。线程会注意到相应Future的被取消并取消整个任务。就在那时CompletionService，他们很快就知道他们已经“完成”了。

考虑增加线程池的大小。

于 2015-12-08T05:09:16.713 回答