如何防止 CompletableFuture#whenComplete 在上下文线程中执行

Question

我有以下代码：

ConcurrentHashMap taskMap= new ConcurrentHashMap();
....
taskMap.compute(key, (k, queue) -> {
        CompletableFuture<Void> future = (queue == null)
                ? CompletableFuture.runAsync(myTask, poolExecutor)
                : queue.whenCompleteAsync((r, e) -> myTask.run(), poolExecutor);
        //to prevent OutOfMemoryError in case if we will have too much keys
        future.whenComplete((r, e) -> taskMap.remove(key, future));            
        return future;
    });

如果future已经完成的whenComplete函数参数在与调用相同的线程中compute调用，则此代码的问题。在此方法的主体中，我们从地图中删除条目。但是计算方法文档禁止这样做并且应用程序冻结。

我该如何解决这个问题？

Answer 1

最明显的解决方案是使用whenCompleteAsync代替whenComplete，因为前者保证使用提供的Executor而不是调用线程来执行操作。可以用

Executor ex = r -> { System.out.println("job scheduled"); new Thread(r).start(); };
for(int run = 0; run<2; run++) {
    boolean completed = run==0;
    System.out.println("*** "+(completed? "with already completed": "with async"));
    CompletableFuture<String> source = completed?
        CompletableFuture.completedFuture("created   in "+Thread.currentThread()):
        CompletableFuture.supplyAsync(() -> {
            LockSupport.parkNanos(TimeUnit.SECONDS.toNanos(1));
            return "created   in "+Thread.currentThread();
        }, ex);

    source.thenApplyAsync(s -> s+"\nprocessed in "+Thread.currentThread(), ex)
          .whenCompleteAsync((s,t) -> {
                if(t!=null) t.printStackTrace(); else System.out.println(s);
                System.out.println("consumed  in "+Thread.currentThread());
            }, ex)
          .join();
}

这将打印出类似的东西

Executor ex = r -> { System.out.println("job scheduled"); new Thread(r).start(); };
for(int run = 0; run<2; run++) {
    boolean completed = run==0;
    System.out.println("*** "+(completed? "with already completed": "with async"));
    CompletableFuture<String> source = completed?
        CompletableFuture.completedFuture("created   in "+Thread.currentThread()):
        CompletableFuture.supplyAsync(() -> {
            LockSupport.parkNanos(TimeUnit.SECONDS.toNanos(1));
            return "created   in "+Thread.currentThread();
        }, ex);

    source.thenApplyAsync(s -> s+"\nprocessed in "+Thread.currentThread(), ex)
          .whenCompleteAsync((s,t) -> {
                if(t!=null) t.printStackTrace(); else System.out.println(s);
                System.out.println("consumed  in "+Thread.currentThread());
            }, ex)
          .join();
}

所以你可以使用

taskMap.compute(key, (k, queue) -> {
        CompletableFuture<Void> future = (queue == null)
                ? CompletableFuture.runAsync(myTask, poolExecutor)
                : queue.whenCompleteAsync((r, e) -> myTask.run(), poolExecutor);
        //to prevent OutOfMemoryError in case if we will have too much keys
        future.whenCompleteAsync((r, e) -> taskMap.remove(key, future), poolExecutor);
        return future;
    });

如果提前完成的可能性很大，则可以使用以下方法减少开销

taskMap.compute(key, (k, queue) -> {
        CompletableFuture<Void> future = (queue == null)
                ? CompletableFuture.runAsync(myTask, poolExecutor)
                : queue.whenCompleteAsync((r, e) -> myTask.run(), poolExecutor);
        //to prevent OutOfMemoryError in case if we will have too much keys
        if(future.isDone()) future = null;
        else future.whenCompleteAsync((r, e) -> taskMap.remove(key, future), poolExecutor);
        return future;
    });

也许，你没有想到这个明显的解决方案，因为你不喜欢依赖动作总是被安排为池的新任务，即使完成已经发生在不同的任务中。您可以使用专门的执行程序解决此问题，该执行程序只会在必要时重新安排任务：

Executor inPlace = Runnable::run;
Thread forbidden = Thread.currentThread();
Executor forceBackground
       = r -> (Thread.currentThread()==forbidden? poolExecutor: inPlace).execute(r);

…

future.whenCompleteAsync((r, e) -> taskMap.remove(key, future), forceBackground);

但是您可能会重新考虑是否真的需要这种复杂的每个映射清理逻辑。这不仅很复杂，而且可能会产生显着的开销，可能会安排大量的清理操作，这些操作在执行时已经过时时并不是真正需要的。

执行起来可能更简单，甚至更高效

taskMap.values().removeIf(CompletableFuture::isDone);

不时清理整个地图。