按照示例代码我正在注入一个睡眠100毫秒的双音素作为一组完成未来的完成动作.我通过给出一个单独的executorService来使用whenCompleteAsync方法. executorService是一个ThreadPoolExecutor,核心池大小为5,最大大小为5,队列长度为1.

public class CompleteTest {
    public static void main(String[] args) {
        ExecutorService executorService = new ThreadPoolExecutor(5, 5, 10,
                TimeUnit.SECONDS, new ArrayBlockingQueue<>(1));

        ArrayList<CompletableFuture<String>> list = new ArrayList<>();

        for (int i = 0; i <100; i++) {
            CompletableFuture<String> stringCompletableFuture = new CompletableFuture<>();
            stringCompletableFuture.whenCompleteAsync((e, a) -> {
                System.out.println("Complete " + e);
                try {
                    Thread.sleep(100);
                } catch (InterruptedException e1) {e1.printStackTrace();}
            }, executorService);

            list.add(stringCompletableFuture);
        }

        for (int i = 0; i < list.size(); i++) {
            list.get(i).complete(i + "");
        }
    }
}

当我运行代码时,即使我正在完成100个期货,也只会打印6个输出.这是5个核心线程和1个排队的线程.剩下的会怎么样？如果其他runnable由于队列已满而无法提交给执行程序服务,则不应该有异常.

产量

Complete 0
Complete 1
Complete 2
Complete 3
Complete 4
Complete 5

解决方法:

抛出异常,并且异常完成CompletableFuture,而不是您正在跟踪的任何一个.

您正在使用构造函数实例化和初始化ThreadPoolExecutor,该构造函数使用默认的RejectedExecutionHandler,它只是抛出异常.我们知道如果ExecutorService无法接受任务,则抛出RejectedExecutionException.那么任务添加到哪里以及抛出的异常在哪里？

就目前而言,所有链接都发生在完全同步时.当你调用它时,你将一个依赖项添加到接收者CompletableFuture,stringCompletableFuture.当stringCompletableFuture完成时(在这种情况下成功),它将创建一个新的CompletableFuture(它返回)并尝试在给定的ExecutorService上安排给定的BiConsumer.

由于ExecutorService的队列没有空格,因此它将调用RejectedExecutionHandler,它将抛出RejectedExecutionException.该异常在那时被捕获并且用于完成将返回的CompletableFuture.

换句话说,在for循环中,捕获whenCompleteAsync返回的CompletableFuture,存储它,并打印出它的状态.

ArrayList<CompletableFuture<String>> list = new ArrayList<>();
ArrayList<CompletableFuture<?>> dependents = new ArrayList<>();
for (int i = 0; i <100; i++) {
    CompletableFuture<String> stringCompletableFuture = new CompletableFuture<>();
    CompletableFuture<?> thisWillHaveException = stringCompletableFuture.whenCompleteAsync((e, a) -> {
        System.out.println("Complete " + e);
        try {
            Thread.sleep(100);
        } catch (InterruptedException e1) {e1.printStackTrace();}
    }, executorService);
    dependents.add(thisWillHaveException);
    list.add(stringCompletableFuture);
}

for (int i = 0; i < list.size(); i++) {
    list.get(i).complete(i + "");
}
Thread.sleep(2000);
dependents.forEach(cf -> {
    cf.whenComplete((r, e) -> {
        if (e != null)
            System.out.println(cf + " " + e.getMessage());
    });
});

您会注意到它们是全部(除了之前成功打印的6个)与RejectedExecutionException异常完成.

...
java.util.concurrent.CompletableFuture@2d8e6db6[Completed exceptionally] java.util.concurrent.RejectedExecutionException: Task java.util.concurrent.CompletableFuture$UniWhenComplete@3f91beef rejected from java.util.concurrent.ThreadPoolExecutor@4eec7777[Running, pool size = 5, active threads = 5, queued tasks = 1, completed tasks = 0]
java.util.concurrent.CompletableFuture@23ab930d[Completed exceptionally] java.util.concurrent.RejectedExecutionException: Task java.util.concurrent.CompletableFuture$UniWhenComplete@1a6c5a9e rejected from java.util.concurrent.ThreadPoolExecutor@4eec7777[Running, pool size = 5, active threads = 5, queued tasks = 1, completed tasks = 0]
java.util.concurrent.CompletableFuture@4534b60d[Completed exceptionally] java.util.concurrent.RejectedExecutionException: Task java.util.concurrent.CompletableFuture$UniWhenComplete@37bba400 rejected from java.util.concurrent.ThreadPoolExecutor@4eec7777[Running, pool size = 5, active threads = 5, queued tasks = 1, completed tasks = 0]