Java并发类API——ExecutorService

1.ExecutorService概述

ExecutorService 是 Java 并发库中一个非常重要的接口，它提供了一种管理和控制线程执行的方法。ExecutorService 是 Executor 接口的扩展，除了 Executor 提供的基础任务执行功能之外，ExecutorService 提供了更强大的功能，如任务提交、任务调度、以及线程池管理等。

1.1主要方法和功能

1. 任务提交:

• submit(Runnable task): 提交一个不带返回值的任务给线程池执行。

• submit(Callable<T> task): 提交一个带返回值的任务给线程池执行，并返回一个 Future<T> 对象，通过 Future 可以获取任务的执行结果。

2. 关闭和终止:

• shutdown(): 启动一个有序的关闭过程，在这个过程中，已提交的任务将被执行，但不会接受新任务。

• shutdownNow(): 尝试停止所有正在执行的任务，暂停处理正在等待的任务，并返回尚未执行的任务列表。

• isShutdown(): 判断线程池是否已经关闭。

• isTerminated(): 判断线程池是否已经终止，所有任务都已完成。

3. 任务管理:

• invokeAll(Collection<? extends Callable<T>> tasks): 执行给定的所有任务，并在所有任务完成后返回一个包含其状态和结果的 Future 列表。

• invokeAny(Collection<? extends Callable<T>> tasks): 执行给定的任务，返回第一个成功完成任务的结果（其他任务将被取消）。

4. 其他工具方法:

• awaitTermination(long timeout, TimeUnit unit): 阻塞直到所有任务在指定的时间内完成，或者线程池关闭，或者当前线程被中断。

1.2 ExecutorService 接口的核心方法

ExecutorService 接口的核心方法主要用于管理和控制线程池中的任务执行。以下是 ExecutorService 接口的核心方法：

任务提交方法

1. submit(Runnable task):

• 用于提交一个不带返回值的任务给线程池执行。返回一个 Future<?> 对象，通过它可以检查任务是否完成或取消。

2. submit(Callable<T> task):

• 用于提交一个带返回值的任务给线程池执行。返回一个 Future<T> 对象，通过它可以获取任务的执行结果。

3. submit(Runnable task, T result):

• 提交一个 Runnable 任务并提供一个默认的返回结果。在任务完成后，返回这个默认的结果。

批量任务执行方法

4. invokeAll(Collection<? extends Callable<T>> tasks):

• 执行给定的所有任务，当所有任务都完成后返回一个 Future<T> 列表。该方法会阻塞直到所有任务都完成。

5. invokeAll(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit):

• 类似于 invokeAll，但这个方法允许设定一个超时时间。如果在规定时间内有任务没有完成，它们将被取消。

6. invokeAny(Collection<? extends Callable<T>> tasks):

• 执行给定的任务，返回第一个成功完成任务的结果（其他任务将被取消）。如果执行中抛出异常，它会在取消所有任务后抛出异常。

方法invokeAny（）取得第一个完成任务的结果值，当第一个任务执行完成后，会调用interrupt（）方法将其他任务中断，所以在这些任务中可以结合if（Thread.currentThread（）.isInterrupted（）==true）代码来决定任务是否继续运行。

当第一个任务执行完成后，会调用interrupt（）方法将其他任务中断，所以在这些任务中可以结合if（Thread.currentThread（）.isInterrupted（）==true）代码来决定任务是否继续运行。

当你使用 invokeAny 执行任务时，如果有多个任务在并行运行，一旦有一个任务成功完成并返回结果，invokeAny 会尝试中断其他正在执行的任务。这是为了避免浪费资源在已经不需要的任务上。

7. invokeAny(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit):

• 类似于 invokeAny，但允许设定一个超时时间。在规定时间内返回第一个完成任务的结果，如果超时，则抛出 TimeoutException。

线程池管理方法

8. shutdown():

• 启动一个有序的关闭过程。在这个过程中，已提交的任务将被执行，但不会接受新的任务提交。

9. shutdownNow():

• 试图停止所有正在执行的任务，并返回尚未执行的任务列表。这个方法会立即尝试停止线程池中的所有活动线程。

10. isShutdown():

• 检查线程池是否已被关闭。如果已调用 shutdown() 或 shutdownNow() 方法，该方法将返回 true。

11. isTerminated():

• 检查线程池是否已终止。在线程池关闭后并且所有任务都完成时返回 true。

12. awaitTermination(long timeout, TimeUnit unit):

• 阻塞当前线程，直到线程池中的所有任务完成执行或超时，或者当前线程被中断。返回一个布尔值，表示线程池是否在超时时间内终止。

2.invokeAny 与 invokeAll

2.1 invokeAny 如何处理中断

1. 成功完成任务后的中断:

• 当一个任务成功完成并返回结果后，invokeAny 会立即中断其他正在执行的任务。这些任务会抛出 InterruptedException 或 CancellationException，表示它们被中断或取消了。

2. 线程的中断状态:

• 如果在调用 invokeAny 方法的线程（主线程）本身被中断，则 invokeAny 方法会抛出 InterruptedException。此时，方法不会再等待任务完成，而是直接抛出异常。

3. 未捕获的异常:

当 invokeAny() 方法中所有任务都抛出异常时，它会抛出一个 ExecutionException，这个异常中包含的是其中一个失败任务的异常，但具体是哪一个任务的异常，并没有明确的规定。这取决于第一个完成并抛出异常的任务。invokeAny() 方法会并行执行传递的多个 Callable 任务，并返回最先成功完成的任务的结果。如果所有任务都失败，invokeAny() 会捕获这些异常，并在它们完成后抛出 ExecutionException。这个 ExecutionException 中包含的异常通常是第一个抛出异常的任务的异常，但这也可能是任何一个已完成任务的异常。由于 invokeAny() 的具体实现可能涉及到多个线程并发执行，因此很难预先确定哪个任务的异常会首先被捕获并作为 ExecutionException 的原因抛出。
在快的任务优先执行完毕后，执行慢的任务出现异常时，默认情况下不会在控制台输出异常信息。如果显式使用try-catch语句块则可以自定义捕获异常。
在执行快的任务出现异常时，在默认情况下是不在控制台输出异常信息的，除非显式使用try-catch捕获，而等待执行慢的任务返回的结果值。

2.2 invokeAny方法示例

任务A ，MyCallableA类代码如下：

package mycallable;

import java.util.concurrent.Callable;

public class MyCallableA implements Callable<String> {

    public String call() throws Exception {
        System.out.println("MyCallableA begin " + System.currentTimeMillis());
        for (int i = 0; i < 123456; i++) {
            Math.random();
            Math.random();
            Math.random();
            System.out.println("MyCallableA " + (i + 1));
        }
        System.out.println("MyCallableA   end " + System.currentTimeMillis());
        return "returnA";
    }

}

任务B，MyCallable B类代码如下：

package mycallable;

import java.util.concurrent.Callable;

public class MyCallableB implements Callable<String> {

    public String call() throws Exception {
        System.out.println("MyCallableB begin " + System.currentTimeMillis());
        for (int i = 0; i < 223456; i++) {
            Math.random();
            Math.random();
            Math.random();
            System.out.println("MyCallableB " + (i + 1));
        }
        System.out.println("MyCallableB   end " + System.currentTimeMillis());
        return "returnB";
    }

}

执行任务类Run1 ，代码如下：

package test.run;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import mycallable.MyCallableA;
import mycallable.MyCallableB1;

public class Run1 {

    public static void main(String[] args) {
        try {
            List list = new ArrayList();
            list.add(new MyCallableA());
            list.add(new MyCallableB());

            ExecutorService executor = Executors.newCachedThreadPool();
            // invokeAny只取得最先完成任务的结果值
            String getValueA = executor.invokeAny(list);
            System.out.println("============" + getValueA);
            System.out.println("ZZZZZZZZZZZZZZZZ");
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }
    }
}

上述代码，Run1 类的 main 方法使用了 ExecutorService 的 invokeAny 方法来执行两个 Callable 任务：MyCallableA 和 MyCallableB。invokeAny 的特点是它会执行多个 Callable 任务，并返回其中最先完成的任务的结果。一旦有一个任务完成了，invokeAny 会中断或取消其他仍在运行的任务。

当 MyCallableA 任务执行完成之后，MyCallableB 任务中的 System.out.println("MyCallableB " + (i + 1)); 是否还会继续输出，取决于以下几点：

1. 任务中断机制:

• 如果 MyCallableB 中的任务在执行过程中检测到线程被中断（即 Thread.currentThread().isInterrupted() 返回 true），或者抛出异常，它应当能够处理这个中断，进而停止执行剩余的代码。

2. 未处理或者捕获中断:

• 如果 MyCallableB1 中的任务未检测或未处理中断，即使线程被中断，任务依然可能继续执行，从而继续输出 System.out.println("MyCallableB " + (i + 1));。

在 MyCallableB 任务代码中，并没有显式地检查 Thread.currentThread().isInterrupted() 以决定是否终止任务执行。因此，即使 invokeAny 中断了 MyCallableB1 任务所属的线程，任务仍会继续执行，输出剩余的 System.out.println("MyCallableB " + (i + 1));

解决方法：

为了确保 MyCallableB1能够正确响应中断，在循环中加入以下检查：

for (int i = 0; i < 223456; i++) {
    if (Thread.currentThread().isInterrupted()) {
        System.out.println("MyCallableB was interrupted");
        break; // 退出循环，终止任务
    }
    Math.random();
    Math.random();
    Math.random();
    System.out.println("MyCallableB " + (i + 1));
}

通过这种方式，当 invokeAny 中断 MyCallableB任务的线程时，任务将检测到中断信号并适时终止，不会继续输出。

2.2.1 疑问解答

疑问：MyCallableB 中public String call() throws Exception，可以看到抛出了异常，为什么“即使 invokeAny 中断了 MyCallableB 任务所属的线程，任务可能仍会继续执行，输出剩余的 System.out.println("MyCallableB " + (i + 1));”呢？

答疑：这涉及Java中的线程中断机制和异常处理。因为代码中没有检查线程的中断状态，也没有在中断时抛出异常的地方，线程的中断状态只是被设置为“中断”，但线程的执行不会因此自动终止。

线程中断与异常抛出

在Java中，线程的中断机制是通过设置线程的中断状态来实现的。线程的中断状态并不会自动停止线程的执行，也不会直接抛出异常。相反，线程中断是一种“请求”操作，告诉线程它应该检查并决定是否要终止执行。

InterruptedException 的抛出

InterruptedException 是一种受检异常，当线程在等待、休眠或被阻塞时（例如调用 Thread.sleep() 或 Object.wait() 等方法），如果线程的中断状态被设置，则会抛出 InterruptedException 并清除线程的中断状态。

在你的 MyCallableB 类中，call() 方法的确声明抛出异常 Exception，但这并不意味着中断线程时一定会抛出 InterruptedException，除非线程在执行过程中触发了会抛出 InterruptedException 的方法（例如 Thread.sleep()）。即使在这些情况下，异常会被抛出，但任务的执行依然取决于代码如何处理该异常。

2.3 方法invokeAny（）与执行慢的任务异常

在快的任务优先执行完毕后，执行慢的任务出现异常时，默认情况下不会在控制台输出异常信息。如果显式使用try-catch语句块则可以自定义捕获异常。
创建类MyCallableA.java代码如下：

package mycallable;

import java.util.concurrent.Callable;

public class MyCallableA implements Callable<String> {

    @Override
    public String call() throws Exception {
        System.out.println("MyCallableA " + Thread.currentThread().getName()
                + " begin " + System.currentTimeMillis());
        for (int i = 0; i < 123456; i++) {
            String newString = new String();
            Math.random();
            Math.random();
            Math.random();
            Math.random();
            Math.random();
            System.out.println("MyCallableA 在运行中=" + (i + 1));
        }
        System.out.println("MyCallableA " + Thread.currentThread().getName()
                + "   end " + System.currentTimeMillis());
        return "returnA";
    }

}

创建类MyCallableB.java代码如下：

package mycallable;

import java.util.concurrent.Callable;

public class MyCallableB implements Callable<String> {

    @Override
    public String call() throws Exception {
        System.out.println("MyCallableB " + Thread.currentThread().getName()
                + " begin " + System.currentTimeMillis());
        for (int i = 0; i < 193456; i++) {
            String newString = new String();
            Math.random();
            Math.random();
            Math.random();
            Math.random();
            Math.random();
            System.out.println("MyCallableB 在运行中=" + (i + 1));
        }
        if (1 == 1) {
            System.out.println("xxxxxxxx=中断了");
            throw new NullPointerException();
        }
        System.out.println("MyCallableB_END "
                + Thread.currentThread().getName() + "   end "
                + System.currentTimeMillis());
        return "returnB";
    }

}

创建类Run.java代码如下：

package test;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import mycallable.MyCallableA;
import mycallable.MyCallableB;

public class Run {

    public static void main(String[] args) {

        try {
            List list = new ArrayList();
            list.add(new MyCallableA());
            list.add(new MyCallableB());

            ExecutorService service = Executors.newCachedThreadPool();
            String getString = service.invokeAny(list);
            System.out.println("zzzz=" + getString);
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }

    }

}

执行结果如下：

程序运行的结果：成功取得returnA字符串，线程B中断了，但抛出的空指针异常却没有在控制台输出。
如果想要在Callable中捕获异常信息，则需要显式地添加try-catch语句块。下面实验一下这个测试。更改MyCallableB.java类代码如下：

package mycallable;

import java.util.concurrent.Callable;

public class MyCallableB implements Callable<String> {

    @Override
    public String call() throws Exception {
        try {
            System.out.println("MyCallableB "
                    + Thread.currentThread().getName() + " begin "
                    + System.currentTimeMillis());
            for (int i = 0; i < 193456; i++) {
                String newString = new String();
                Math.random();
                Math.random();
                Math.random();
                Math.random();
                Math.random();
                System.out.println("MyCallableB 在运行中=" + (i + 1));
            }
            if (1 == 1) {
                System.out.println("xxxxxxxx=中断了");
                throw new NullPointerException();
            }
            System.out.println("MyCallableB_END "
                    + Thread.currentThread().getName() + "   end "
                    + System.currentTimeMillis());
        } catch (Exception e) {
            e.printStackTrace();
            System.out.println(e.getMessage() + " 通过显式try-catch捕获异常了");
            throw e;
        }
        return "returnB";
    }

}

更改Run.java类代码如下：

package test;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import mycallable.MyCallableA;
import mycallable.MyCallableB;

public class Run {

    public static void main(String[] args) {

        try {
            List list = new ArrayList();
            list.add(new MyCallableA());
            list.add(new MyCallableB());

            ExecutorService service = Executors.newCachedThreadPool();
            String getString = service.invokeAny(list);
            System.out.println("zzzz=" + getString);
        } catch (InterruptedException e) {
            e.printStackTrace();
            System.out.println("mainA");
        } catch (ExecutionException e) {
            e.printStackTrace();
            System.out.println("mainB");
        }

    }

}

程序运行结果如图:

从运行结果来看，加入显式的try-catch语句块可以捕获异常信息，但抛出去的异常在main（）方法中却没有得到捕获，也就是字符串mainA和mainB没有被打印，也就说明子线程出现异常时是不影响main线程的主流程的.

2.4 方法invokeAll

invokeAll(Collection tasks) 方法用于执行给定的一组 Callable 任务，并返回一个 List<Future<T>>，其中每个 Future 对象对应一个任务的结果。当所有任务完成（成功或失败）时，invokeAll 方法返回。

以下是 invokeAll() 方法的行为和结果：

invokeAll() 的行为

1. 所有任务都执行完成: invokeAll() 会等待所有任务完成，无论任务是正常完成还是抛出异常。在所有任务完成之前，invokeAll() 会阻塞。

2. 返回的 List<Future<T>>:

• invokeAll() 返回一个 List<Future<T>>，其中的每个 Future 对象对应传递给它的每个 Callable 任务。List 中的 Future 对象的顺序与 Collection 中传递的 Callable 顺序相同。

3. Future 对象的状态:

• 对于成功完成的任务，调用 Future.get() 会返回任务的结果。

• 对于失败（抛出异常）的任务，调用 Future.get() 会抛出 ExecutionException，其中封装了任务的异常。

invokeAll（）方法对Callable抛出去的异常是可以在invokeAll() 所在的线程main（）方法中处理的。

问：方法invokeAll（Collection tasks）快的任务A异常了，慢的任务B正常。invokeAll() 所在的线程main（）try-catch了异常并打印错误日志，那么结果会怎么样？

类MyCallableA.java代码如下：

package mycallable;

import java.util.concurrent.Callable;

public class MyCallableA implements Callable<String> {

    public String call() throws Exception {
        System.out.println("MyCallableA begin " + System.currentTimeMillis());
        for (int i = 0; i < 123; i++) {
            Math.random();
            Math.random();
            Math.random();
            System.out.println("MyCallableA " + (i + 1));
        }
        System.out.println("MyCallableA   end " + System.currentTimeMillis());
        if (1 == 1) {
            System.out.println("A报错了");
            throw new Exception("出现异常");
        }
        return "returnA";
    }

}

类MyCallableB.java代码如下：

package mycallable;

import java.util.concurrent.Callable;

public class MyCallableB implements Callable<String> {

    public String call() throws Exception {
        System.out.println("MyCallableB begin " + System.currentTimeMillis());
        for (int i = 0; i < 123456; i++) {
            Math.random();
            Math.random();
            Math.random();
            System.out.println("MyCallableB " + (i + 1));
        }
        System.out.println("MyCallableB   end " + System.currentTimeMillis());
        return "returnB";
    }

}

类Run.java代码如下：

package test.run;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

import mycallable.MyCallableA;
import mycallable.MyCallableB;

public class Run {

    public static void main(String[] args) {

        try {
            List list = new ArrayList();
            list.add(new MyCallableA());
            list.add(new MyCallableB());

            ExecutorService executor = Executors.newCachedThreadPool();
            System.out.println("Y=" + System.currentTimeMillis());
            List<Future<String>> listFuture = executor.invokeAll(list);
            System.out.println("Z=" + System.currentTimeMillis());
            for (int i = 0; i < listFuture.size(); i++) {
                System.out.println("result=" + listFuture.get(i).get());
            }
            System.out.println("mainEND");
        } catch (InterruptedException e) {
            System.out.println("报错了InterruptedException");
            e.printStackTrace();
        } catch (ExecutionException e) {
            System.out.println("报错了ExecutionException");
            e.printStackTrace();
        }
    }
}

需要注意的是，代码：

List list = new ArrayList();
list.add(new MyCallableA());
list.add(new MyCallableB());

向list添加Callable的顺序与运行结果有联系，在上面代码中先添加的MyCallableA任务，后添加的MyCallableB任务，则main方法中的代码为：

for (int i = 0; i < listFuture.size(); i++) {
    System.out.println("result=" + listFuture.get(i).get());
}

在第1次循环时取得的是MyCallableA的Future对象，由于MyCallableA速度快并且出现了异常，则在第一次for时出现ExecutionException异常，不再继续执行第2次循环，进入main方法中的catch语句块。

程序运行结果如图7-13所示。

图运行结果

因为线程A出现了异常，在第一次for循环取得返回值时产生异常并退出for循环，也就导致了没有执行第2次for循环，所以在main（）方法中没有获得线程B的返回值，程序流程直接进入catch语句块，导致mainEND字符串并没有输出。

使用invokeAll（）方法如果全部任务都出现异常时，打印的结果与此示例效果一样。