JavaSE-线程池（5）- ThreadPoolExecutor常用方法

invokeAll

ExecutorService 接口中定义的方法，给定一组任务，在所有任务执行完成时返回一个 Futures 列表，其中包含它们的状态和结果。

/**
 * Executes the given tasks, returning a list of Futures holding
 * their status and results when all complete.
 * {@link Future#isDone} is {@code true} for each
 * element of the returned list.
 * Note that a <em>completed</em> task could have
 * terminated either normally or by throwing an exception.
 * The results of this method are undefined if the given
 * collection is modified while this operation is in progress.
 *
 * @param tasks the collection of tasks
 * @param <T> the type of the values returned from the tasks
 * @return a list of Futures representing the tasks, in the same
 *         sequential order as produced by the iterator for the
 *         given task list, each of which has completed
 * @throws InterruptedException if interrupted while waiting, in
 *         which case unfinished tasks are cancelled
 * @throws NullPointerException if tasks or any of its elements are {@code null}
 * @throws RejectedExecutionException if any task cannot be
 *         scheduled for execution
 */
<T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
    throws InterruptedException;

invokeAll 使用方式

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class ThreadPoolExecutorInvokeAllTest {
    static class MyTask implements Callable<Boolean> {

        private int id;

        public MyTask(int id) {
            this.id = id;
        }

        @Override
        public Boolean call() throws Exception {
            try {
                System.out.println(
                    "time:" + System.currentTimeMillis() + " " + Thread.currentThread() + " execute task " + id +
                        " start");
                Thread.sleep(2000);
                System.out.println(
                    "time:" + System.currentTimeMillis() + " " + Thread.currentThread() + " execute task " + id +
                        " finish");
            } catch (InterruptedException e) {
                e.printStackTrace();
                return false;
            }
            return true;
        }
    }

    public static void main(String[] args) {
        long start = System.currentTimeMillis();
        ThreadPoolExecutor executor =
            new ThreadPoolExecutor(10, 40, 5000,
                TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>(2));
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());

        List<Callable<Boolean>> tasks = new ArrayList<>();
        for (int i = 1; i <= 5; i++) {
            tasks.add(new MyTask(i));
        }
        List<Future<Boolean>> futures = null;
        try {
            futures = executor.invokeAll(tasks);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("time:" + System.currentTimeMillis() + " 主线程会等待invokeAll执行完成才继续执行");
        for (int i = 0; i < futures.size(); i++) {
            try {
                System.out.println(futures.get(i).get());
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (ExecutionException e) {
                e.printStackTrace();
            }
        }
        executor.shutdown();
        System.out.println("耗时：" + (System.currentTimeMillis() - start));
    }
}

如打印结果所示，主线程会等待 invokeAll 方法中的任务执行完后才继续执行

time:1677229181926 Thread[pool-1-thread-1,5,main] execute task 1 start
time:1677229181927 Thread[pool-1-thread-2,5,main] execute task 2 start
time:1677229181928 Thread[pool-1-thread-3,5,main] execute task 3 start
time:1677229181929 Thread[pool-1-thread-4,5,main] execute task 4 start
time:1677229181930 Thread[pool-1-thread-5,5,main] execute task 5 start
time:1677229183942 Thread[pool-1-thread-2,5,main] execute task 2 finish
time:1677229183942 Thread[pool-1-thread-1,5,main] execute task 1 finish
time:1677229183942 Thread[pool-1-thread-3,5,main] execute task 3 finish
time:1677229183942 Thread[pool-1-thread-5,5,main] execute task 5 finish
time:1677229183942 Thread[pool-1-thread-4,5,main] execute task 4 finish
time:1677229183965 主线程会等待invokeAll执行完成才继续执行
true
true
true
true
true
耗时：2141

一般在使用 invokeAll 方法时建议加上等待时间，防止任务执行时间过长线程一直阻塞，方法定义如下：

<T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks,
                                  long timeout, TimeUnit unit)
        throws InterruptedException;

如果在 1 秒钟时间内任务没有结束将会被取消

futures = executor.invokeAll(tasks,1,TimeUnit.SECONDS);

执行结果如下：

time:1677231068880 Thread[pool-1-thread-1,5,main] execute task 1 start
time:1677231068881 Thread[pool-1-thread-3,5,main] execute task 3 start
time:1677231068880 Thread[pool-1-thread-2,5,main] execute task 2 start
time:1677231068881 Thread[pool-1-thread-4,5,main] execute task 4 start
time:1677231068881 Thread[pool-1-thread-5,5,main] execute task 5 start
java.lang.InterruptedException: sleep interrupted
	at java.lang.Thread.sleep(Native Method)
	at test.executors.ThreadPoolExecutorInvokeAllTest$MyTask.call(ThreadPoolExecutorInvokeAllTest.java:27)
	at test.executors.ThreadPoolExecutorInvokeAllTest$MyTask.call(ThreadPoolExecutorInvokeAllTest.java:13)
	at java.util.concurrent.FutureTask.run$$$capture(FutureTask.java:266)
	at java.util.concurrent.FutureTask.run(FutureTask.java)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
	at java.lang.Thread.run(Thread.java:750)
。。。省略部分报错
java.util.concurrent.CancellationException
	at java.util.concurrent.FutureTask.report(FutureTask.java:121)
	at java.util.concurrent.FutureTask.get(FutureTask.java:192)
	at test.executors.ThreadPoolExecutorInvokeAllTest.main(ThreadPoolExecutorInvokeAllTest.java:59)
java.util.concurrent.CancellationException
	at java.util.concurrent.FutureTask.report(FutureTask.java:121)
	at java.util.concurrent.FutureTask.get(FutureTask.java:192)
	at test.executors.ThreadPoolExecutorInvokeAllTest.main(ThreadPoolExecutorInvokeAllTest.java:59)
。。。省略部分报错
time:1677231069895 主线程会等待invokeAll执行完成才继续执行
耗时：1113

invokeAny

ExecutorService 接口中定义的方法，给定一组任务，只要有一个任务执行完成就返回这个任务的结果

/**
* Executes the given tasks, returning the result
* of one that has completed successfully (i.e., without throwing
* an exception), if any do. Upon normal or exceptional return,
* tasks that have not completed are cancelled.
* The results of this method are undefined if the given
* collection is modified while this operation is in progress.
* 执行给定的任务，返回成功完成的任务的结果（即没有抛出异常），如果有的话。
* 在正常或异常返回时，未完成的任务将被取消。如果在执行此操作时修改了给定的集合，则此方法的结果是未定义的。
*/
<T> T invokeAny(Collection<? extends Callable<T>> tasks)
   throws InterruptedException, ExecutionException;

invokeAny 使用方式

如下，有5个任务，每个任务等待时间为 0-3 秒，线程池调用 invokeAny 方法获取最终执行的任务名称

import java.util.ArrayList;
import java.util.List;
import java.util.Random;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

public class ThreadPoolExecutorInvokeAnyTest {
    static class MyTask implements Callable<String> {

        private int id;

        public MyTask(int id) {
            this.id = id;
        }

        @Override
        public String call() throws Exception {
            Integer randomTime = new Random().nextInt(4) * 1000;
            try {
                System.out.println(
                    "time:" + System.currentTimeMillis() + " " + Thread.currentThread() + " execute task " + id +
                        " start,random time :"+randomTime);
                Thread.sleep(randomTime);
                System.out.println(
                    "time:" + System.currentTimeMillis() + " " + Thread.currentThread() + " execute task " + id +
                        " finish");
            } catch (InterruptedException e) {
                e.printStackTrace();
                return "";
            }
            return "task" + id;
        }
    }

    public static void main(String[] args) {
        long start = System.currentTimeMillis();
        ThreadPoolExecutor executor =
            new ThreadPoolExecutor(10, 40, 5000,
                TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>(2));
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());

        List<Callable<String>> tasks = new ArrayList<>();
        for (int i = 1; i <= 5; i++) {
            tasks.add(new MyTask(i));
        }
        String futureResult = null;
        try {
            futureResult = executor.invokeAny(tasks, 3, TimeUnit.SECONDS);
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        } catch (TimeoutException e) {
            e.printStackTrace();
        }
        System.out.println("time:" + System.currentTimeMillis() + " 主线程会等待invokeAny执行完成才继续执行");
        System.out.println("执行的任务为：" + futureResult);
        executor.shutdown();
        System.out.println("耗时：" + (System.currentTimeMillis() - start));
    }
}

执行结果如下，task5等待0秒，所以 invokeAny方法返回结果为task5，其他任务等待时间都超过0秒，所以都被取消执行