java.util.concurrent.RejectedExecutionException：

Question

我试图做小 poc 来了解 Futures.addCallback(future, callback) 和 Futures.addCallback(future, callback, executor1) 之间的区别。根据我的理解，在后一种情况下，回调应该由函数调用中指定的 executor1 运行。 但是，在运行以下代码 sn-p 时，我无法找出上述异常的原因：

  class MyTest implements Callable<Integer>{

    int i;
    int j;

    public MyTest(int k, int l) {
        i = k;
        j = l;
    }

    public int getI() {
        return i;
    }

    public void setI(int i) {
        this.i = i;
    }

    public int getJ() {
        return j;
    }

    public void setJ(int j) {
        this.j = j;
    }

    @Override
    public Integer call() throws Exception {
         System.out.println(Thread.currentThread().getName());  
        System.out.println("Computing.........");
        Thread.currentThread().sleep(5000);
        return i + j;
    }

}


Callable<Integer> call = new MyTest(1,2);
        ListeningExecutorService executor = MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(1));
        ListeningExecutorService executor1 = MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(1));
        //ExecutorService executor = Executors.newFixedThreadPool(1);
        ListenableFuture<Integer> future = executor.submit(call);
        FutureCallback<Integer> callback = new FutureCallback<Integer>() {

            @Override
            public void onSuccess(Integer result) {
                System.out.println(Thread.currentThread().getName());   
                System.out.println("Completed Successfully and result is :" + result);

            }

            @Override
            public void onFailure(Throwable t) {
                System.out.println("Failed");

            }
        };

        Futures.addCallback(future, callback, executor1);
        executor.shutdown();
        executor1.shutdown();

异常日志

 ????? ??, ???? ?:??:?? ??????? com.google.common.util.concurrent.ExecutionList executeListener
    SEVERE: RuntimeException while executing runnable com.google.common.util.concurrent.Futures$5@37e5f7d7 with executor com.google.common.util.concurrent.MoreExecutors$ListeningDecorator@3d1d9ca7
    java.util.concurrent.RejectedExecutionException: Task com.google.common.util.concurrent.Futures$5@37e5f7d7 rejected from java.util.concurrent.ThreadPoolExecutor@39220f5b[Terminated, pool size = 0, active threads = 0, queued tasks = 0, completed tasks = 0]
        at java.util.concurrent.ThreadPoolExecutor$AbortPolicy.rejectedExecution(ThreadPoolExecutor.java:2047)
        at java.util.concurrent.ThreadPoolExecutor.reject(ThreadPoolExecutor.java:823)
        at java.util.concurrent.ThreadPoolExecutor.execute(ThreadPoolExecutor.java:1369)
        at com.google.common.util.concurrent.MoreExecutors$ListeningDecorator.execute(MoreExecutors.java:484)
        at com.google.common.util.concurrent.ExecutionList.executeListener(ExecutionList.java:156)
        at com.google.common.util.concurrent.ExecutionList.execute(ExecutionList.java:145)
        at com.google.common.util.concurrent.ListenableFutureTask.done(ListenableFutureTask.java:91)
        at java.util.concurrent.FutureTask.finishCompletion(FutureTask.java:384)
        at java.util.concurrent.FutureTask.set(FutureTask.java:233)
        at java.util.concurrent.FutureTask.run(FutureTask.java:274)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
        at java.lang.Thread.run(Thread.java:745)

Answer 1

抛出异常是因为在你的代码中 sn-p executor1 被调用 executor1.shutdown(); 停止尝试注释掉这一行并且异常消失了。

如果你想在callback 完成后关闭执行器，在回调代码中执行关闭

  FutureCallback<Integer> callback = new FutureCallback<Integer>() {

        @Override
        public void onSuccess(Integer result) {
            System.out.println(Thread.currentThread().getName());   
            System.out.println("Completed Successfully and result is :" + result);
            executor1.shutdown(); // shutdown it here! 
        }

        @Override
        public void onFailure(Throwable t) {
            System.out.println("Failed");
            executor1.shutdown(); // shutdown it here! 
        }
    };
    Futures.addCallback(future, callback, executor1);
    executor.shutdown();
    // executor1.shutdown(); // do not stop it here!