使用Java Executor框架实现多线程

亦凉 2023-06-24 15:25 29阅读 0赞

本文将涵盖两个主题：

\*   通过实现Callable接口创建线程

\*   在Java中使用Executor框架

实现Callable接口

为了创建一段可以在线程中运行的代码，我们创建了一个类，然后实现了Callable接口。这段代码完成的任务需要放在call（）函数中。在下面的代码中，你可以看到Callable task是一个实现Callable接口的类，在函数中完成了将0到4之间的数字相加的任务。

package com.mzc.common.thread;
    
    import java.util.concurrent.Callable;
    
    /**
     * <p class="detail">
     * 功能: 实现Callable接口
     * </p>
     *
     * @author Moore
     * @ClassName Callable task.
     * @Version V1.0.
     * @date 2019.12.23 10:06:23
     */
    public class CallableTask implements Callable<Integer> {
        /**
         * Computes a result, or throws an exception if unable to do so.
         *
         * @return computed result
         * @throws Exception if unable to compute a result
         */
        @Override
        public Integer call() throws Exception {
            int sum = 0;
            for (int i = 0; i < 5; i++) {
                sum += i;
            }
            return sum;
        }
    }

在上面的代码中，你会注意到Callable的参数为Integer。这表明此Callable的返回类型将为Integer。还可以看到调用函数返回了Integer类型。

1、创建线程并运行

下面的代码显示了如何创建线程，然后运行它们。

package com.mzc.common.thread;
    
    import java.util.concurrent.Callable;
    import java.util.concurrent.ExecutionException;
    import java.util.concurrent.FutureTask;
    
    public class CallableInterfaceDemo {
        public static void main(String[] args) {
            FutureTask<Integer>[] futureList = new FutureTask[5];
            for (int i = 0; i <= 4; i++) {
                Callable<Integer> callable = new CallableTask();
                futureList[i] = new FutureTask<Integer>(callable);
                Thread t = new Thread(futureList[i]);
                t.start();
            }
            for (int i = 0; i <= 4; i++) {
                FutureTask<Integer> result = futureList[i];
                try {
                    System.out.println("Future Task" + i + ":" + result.get());
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } catch (ExecutionException e) {
                    e.printStackTrace();
                }
            }
        }
    }

为了创建线程，首先我们需要创建一个CallableTask实例，该实例实现CallableI接口，如图所示：

Callable<Integer> callable = new CallableTask();

然后我们需要创建一个FutureTask类的实例，并将Callable Task的实例作为参数传递，如下所示：

futureList[i] = new FutureTask<Integer>(callable);

然后创建一个线程，我们创建一个Thread类的实例，并将FutureTask类的实例作为参数传递，如下所示：

Thread t = new Thread(futureList[i]);

最后，使用start()方法启动线程。

2、从线程获取结果

如果是Callables，线程实际上可以返回一个值。为了获得该值，我们可以在FutureTask实例上调用get（）函数。在我们的代码中，线程的返回值是0到4之间的数字之和。如下面的代码片段所示：

FutureTask<Integer> result = futureList[i];
    try {
      System.out.println("Future Task" + i + ":" + result.get());
    } catch (InterruptedException e) {
      e.printStackTrace();
    } catch (ExecutionException e) {     
      e.printStackTrace();
    }

线程可能抛出异常，可以使用try catch块进行处理。

运行结果：

Future Task0:10
    Future Task1:10
    Future Task2:10
    Future Task3:10
    Future Task4:10

\*\*小课堂\*\*

上面示例中，我们使用了new Thread()方式创建一个线程，但在实际项目中我们很少这样用，因为每次new的时候都是新建了一个线程对象，并且不能复用，只能释放线程资源，这种方式性能较差，而且如果不加限制，创建太多的线程会消耗太多系统资源，同时线程太多，如果要定期执行、关闭线程等，都不方便统一管理。

所以我们就需要线程池来管理线程，使用线程池有以下优点：

\*   可以复用线程，减少对象创建，就减少系统资源消耗

\*   可以控制最大并发线程数，提高系统资源利用率

\*   使用线程池可以避免资源竞争，也就减少了阻塞情况

\*   创建线程池简单方便，同时便于统一管理

Executor框架

每次创建线程都是资源密集型的。一个很好的替代方法是提前设置好一些线程，也就是我上面说的线程池，然后将我们的任务分配给这些线程。这就是Executors类和ExecutorService非常有用的地方。

![format_png][]

上图显示了具有4个线程的线程池。每当我们希望运行任何任务时，都可以将其分配给这些线程。任务完成后，线程将被释放以执行其他任务。

1、如何使用Executor框架

我们还是先复用上面实现了Callable接口的CallableTask实例，然后我们改造一下创建线程方式，不再使用new Thread() 创建一个线程，而是先创建一个ExecutorService，Executors类具有ExecutorService的多个实现，这儿我们使用Executors类创建大小为4的固定线程池(newFixedThreadPool)。

ExecutorService executors = Executors.newFixedThreadPool(4);

接下来，我们需要将我们的任务提交给ExecutorService，像这样：

Future<Integer> future = executors.submit(w);

提交任务后，我们将获得一个Future对象的实例。Future对象将存储Task的结果。完整代码：

package com.mzc.common.thread;
    
    import java.util.concurrent.*;
    
    /**
     * <p class="detail">
     * 功能: 使用Executor创建多线程
     * </p>
     *
     * @author Moore
     * @ClassName Executor demo.
     * @Version V1.0.
     * @date 2019.12.23 10:54:31
     */
    public class ExecutorDemo {
    
        public static void main(String[] args) {
            /**
             * 创建线程池的6种方式
             */
    
            // 1、创建一个定长线程池，可控制线程最大并发数，超出的线程会在队列中等待。
            ExecutorService executors = Executors.newFixedThreadPool(4);
    
            // 2、创建一个可缓存线程池，如果线程池长度超过处理需要，可灵活回收空闲线程，若无可回收，则新建线程。
    //        ExecutorService executors = Executors.newCachedThreadPool();
    
            // 3、创建一个定长线程池，支持定时及周期性任务执行。
    //        ExecutorService executors = Executors.newScheduledThreadPool(4);
    
            // 4、创建一个单线程化的线程池，它只会用唯一的工作线程来执行任务，保证所有任务按照指定顺序(FIFO, LIFO, 优先级)执行。
    //        ExecutorService executors = Executors.newSingleThreadExecutor();
    
            // 5、newWorkStealingPool工作窃取线程池，它是新的线程池类ForkJoinPool的扩展，
            // 但是都是在统一的一个Executors类中实现，由于能够合理的使用CPU进行对任务操作（并行操作），
            // 所以适合使用在很耗时的任务中
    //        ExecutorService executors = Executors.newWorkStealingPool(4);
    
            // 6、newSingleThreadScheduledExecutor创建线程池同时放入多个线程时，每个线程都会按照自己的调度来执行，
            // 但是当其中一个线程被阻塞时，其它的线程都会受到影响被阻塞，
            // 不过依然都会按照自身调度来执行，但是会存在阻塞延迟。
    //        ExecutorService executors = Executors.newSingleThreadScheduledExecutor();
            Future<Integer>[] futures = new Future[5];
            Callable<Integer> w = new CallableTask();
            try {
                for (int i = 0; i < 5; i++) {
                    Future<Integer> future = executors.submit(w);
                    futures[i] = future;
                }
    
                for (int i = 0; i < futures.length; i++) {
                    try {
                        System.out.println("Result from Future " + i + ":" + futures[i].get());
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    } catch (ExecutionException e) {
                        e.printStackTrace();
                    }
                }
            } finally {
                executors.shutdown();
            }
        }
    }

2、获取结果

为了获得每个任务的结果，我们可以调用Future实例的get（）方法，方法和上面的一样：

for (int i = 0; i < futures.length; i++) {
                    try {
                        System.out.println("Result from Future " + i + ":" + futures[i].get());
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    } catch (ExecutionException e) {
                        e.printStackTrace();
                    }
                }

运行结果

Result from Future 0:10
    Result from Future 1:10
    Result from Future 2:10
    Result from Future 3:10
    Result from Future 4:10

固定线程池(newFixedThreadPool)的运行情况说明：

\*   在上面的示例中，我们创建了一个大小为4的固定线程池。

\*   如果我们总共向ExecutorService提交3个任务，那么所有3个任务都将分配给线程池，并且它们将开始执行。

\*   如果我们向ExecutorService提交4个任务，那么所有这4个任务将再次分配给线程池，并且它们将开始执行。

\*   如果我们向该线程池提交5个任务，只有4个任务将分配给线程池。这是因为线程池的大小为4。仅当释放池中的线程之一时，才会分配第五个任务。

关闭ExecutorService

当我们不再需要线程时，需要关闭ExecutorService，这样做能确保JVM不会消耗其他资源。我们可以使用下面这个命令关闭ExecutorService：

executors.shutdown();

ExecutorService的关闭操作通常位于\`\`finally''块中。这是为了确保即使在发生任何异常的情况下，关闭操作始终在代码的末尾被执行。如果关闭操作不正确，那么如果发生任何异常，则ExecutorService仍将运行并消耗其他JVM资源。

总结

线程池的一些常用方法

\*   submit()：提交任务，能够返回执行结果execute+Future

\*   shutdown()：关闭线程池，等待任务都执行完

\*   getPoolSize()：线程池当前线程数量

\*   getActiveCount()：当前线程池中正在执行任务的线程数量

\*   shutdownNow()：关闭线程池，不等待任务执行完

\*   getTaskCount()：线程池已执行和未执行的任务总数

\*   getCompletedTaskCount()：已完成的任务数量

创建线程池的6种方式

1.  newFixedThreadPool()：创建一个定长线程池，可控制线程最大并发数，超出的线程会在队列中等待。

2.  EnewCachedThreadPool()：创建一个可缓存线程池，如果线程池长度超过处理需要，可灵活回收空闲线程，若无可回收，则新建线程。

3.  newScheduledThreadPool()：创建一个定长线程池，支持定时及周期性任务执行。

4.  newSingleThreadExecutor()：创建一个单线程化的线程池，它只会用唯一的工作线程来执行任务，保证所有任务按照指定顺序(\*FIFO\*,\* LIFO\*,优先级)执行。

5.  newWorkStealingPool()：工作窃取线程池，它是新的线程池类\*ForkJoinPool\*的扩展，但是都是在统一的一个\*Executors\*类中实现，由于能够合理的使用\*CPU\*进行对任务操作（并行操作），所以适合使用在很耗时的任务中。

6.  newSingleThreadScheduledExecutor()：\*newSingleThreadScheduledExecutor\*创建线程池同时放入多个线程时，每个线程都会按照自己的调度来执行，但是当其中一个线程被阻塞时，其它的线程都会受到影响被阻塞，不过依然都会按照自身调度来执行，但是会存在阻塞延迟。

代码示例

// 1、创建一个定长线程池，可控制线程最大并发数，超出的线程会在队列中等待。
            ExecutorService executors = Executors.newFixedThreadPool(4);
    
            // 2、创建一个可缓存线程池，如果线程池长度超过处理需要，可灵活回收空闲线程，若无可回收，则新建线程。
            ExecutorService executors = Executors.newCachedThreadPool();
    
            // 3、创建一个定长线程池，支持定时及周期性任务执行。
            ExecutorService executors = Executors.newScheduledThreadPool(4);
    
            // 4、创建一个单线程化的线程池，它只会用唯一的工作线程来执行任务，保证所有任务按照指定顺序(FIFO, LIFO, 优先级)执行。
            ExecutorService executors = Executors.newSingleThreadExecutor();
    
            // 5、newWorkStealingPool工作窃取线程池，它是新的线程池类ForkJoinPool的扩展，
            // 但是都是在统一的一个Executors类中实现，由于能够合理的使用CPU进行对任务操作（并行操作），
            // 所以适合使用在很耗时的任务中
            ExecutorService executors = Executors.newWorkStealingPool(4);
    
            // 6、newSingleThreadScheduledExecutor创建线程池同时放入多个线程时，每个线程都会按照自己的调度来执行，
            // 但是当其中一个线程被阻塞时，其它的线程都会受到影响被阻塞，
            // 不过依然都会按照自身调度来执行，但是会存在阻塞延迟。
            ExecutorService executors = Executors.newSingleThreadScheduledExecutor();

如果觉得本文觉得还行，可以关注我的个人公众号：码之初。给与我更多写下去的动力，谢谢！码之初将为您奉献：  
\*   Java基础到架构知识全方位涵盖

\*   海量面试资料，从文档到视频，从基础到底层，面试无忧

\*   程序猿生活记录，程序猿的生活也值得关注

\*   不定期的福利发放，惊喜就是在不经意间来临。

![format_png 1][]

[format_png]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly91cGxvYWQtaW1hZ2VzLmppYW5zaHUuaW8vdXBsb2FkX2ltYWdlcy8yMjM4NDE4LTcwYmJhZTM4ODcyOWE5YzMucG5n?x-oss-process=image/format,png
[format_png 1]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly91cGxvYWQtaW1hZ2VzLmppYW5zaHUuaW8vdXBsb2FkX2ltYWdlcy8yMjM4NDE4LTcyYzY0YjNiNjc0YjMyMTIucG5n?x-oss-process=image/format,png