二十二. 并发实战-ThreadPoolExecutor使用

- 日理万妓 2024-03-29 16:20 70阅读 0赞

### 前言 ###

本篇文章将对线程池`ThreadPoolExecutor`的使用进行说明。

### 正文 ###

#### 一. 执行无返回值任务 ####

通过`ThreadPoolExecutor`的`execute()`方法，能执行`Runnable`任务，示例如下。

public class ThreadPoolExecutorTest {
        
    
        @Test
        public void ThreadPoolExecutor执行简单无返回值任务() throws Exception {
        
            // 创建一个线程池
            ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(2, 4,
                    60, TimeUnit.SECONDS, new ArrayBlockingQueue<>(300));
    
            // 创建两个任务
            Runnable firstRunnable = new Runnable() {
        
                @Override
                public void run() {
        
                    System.out.println("第一个任务执行");
                }
            };
            Runnable secondRunnable = new Runnable() {
        
                @Override
                public void run() {
        
                    System.out.println("第二个任务执行");
                }
            };
    
            // 让线程池执行任务
            threadPoolExecutor.execute(firstRunnable);
            threadPoolExecutor.execute(secondRunnable);
    
            // 让主线程睡眠1秒，等待线程池中的任务被执行完毕
            Thread.sleep(1000);
        }
        
    }

运行测试程序，结果如下。

![在这里插入图片描述][2d2125141cd74b9c97338ba55aa418d0.jpeg_pic_center]

#### 二. 执行有返回值任务 ####

通过`ThreadPoolExecutor`的`submit()`方法，能够执行`Callable`任务，通过`submit()`方法返回的`RunnableFuture`能够拿到异步执行的结果。示例如下。

public class ThreadPoolExecutorTest {
        
    
        @Test
        public void ThreadPoolExecutor执行简单有返回值任务() throws Exception {
        
            // 创建一个线程池
            ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(2, 4,
                    60, TimeUnit.SECONDS, new ArrayBlockingQueue<>(300));
    
            // 创建两个任务，任务执行完有返回值
            Callable<String> firstCallable = new Callable<String>() {
        
                @Override
                public String call() throws Exception {
        
                    return "第一个任务返回值";
                }
            };
            Callable<String> secondCallable = new Callable<String>() {
        
                @Override
                public String call() throws Exception {
        
                    return "第二个任务返回值";
                }
            };
    
            // 让线程池执行任务
            Future<String> firstFuture = threadPoolExecutor.submit(firstCallable);
            Future<String> secondFuture = threadPoolExecutor.submit(secondCallable);
    
            // 获取执行结果，拿不到结果会阻塞在get()方法上
            System.out.println(firstFuture.get());
            System.out.println(secondFuture.get());
        }
    
    }

运行测试程序，结果如下。

![在这里插入图片描述][acc03e00fe574ce0892ed60fb28ef0c7.jpeg_pic_center]

#### 三. 执行有返回值任务时抛出错误 ####

如果`ThreadPoolExecutor`在执行`Callable`任务时，在`Callable`任务中抛出了异常并且没有捕获，那么这个异常是可以通过`Future`的`get()`方法感知到的。示例如下。

public class ThreadPoolExecutorTest {
        
    
        @Test
        public void ThreadPoolExecutor执行简单有返回值任务时抛出错误() {
        
            // 创建一个线程池
            ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(2, 4,
                    60, TimeUnit.SECONDS, new ArrayBlockingQueue<>(300));
    
            // 创建一个任务，任务有返回值，但是执行过程中抛出异常
            Callable<String> exceptionCallable = new Callable<String>() {
        
                @Override
                public String call() throws Exception {
        
                    throw new RuntimeException("发生了异常");
                }
            };
    
            // 让线程池执行任务
            Future<String> exceptionFuture = threadPoolExecutor.submit(exceptionCallable);
    
            try {
        
                System.out.println(exceptionFuture.get());
            } catch (Exception e) {
        
                System.out.println(e.getMessage());
            }
        }
    
    }

运行测试程序，结果如下。

![在这里插入图片描述][c6367c1f26264e48b6f69eb4259af346.jpeg_pic_center]

#### 四. ThreadPoolExecutor通过submit方式执行Runnable ####

`ThreadPoolExecutor`可以通过`submit()`方法来运行`Runnable`任务，并且还可以异步获取执行结果。示例如下。

public class ThreadPoolExecutorTest {
        
    
        @Test
        public void ThreadPoolExecutor通过submit的方式来提交并执行Runnable() throws Exception {
        
            // 创建一个线程池
            ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(2, 4,
                    60, TimeUnit.SECONDS, new ArrayBlockingQueue<>(300));
    
            // 创建结果对象
            MyResult myResult = new MyResult();
            // 创建Runnable对象
            Runnable runnable = new Runnable() {
        
                @Override
                public void run() {
        
                    myResult.setResult("任务执行了");
                }
            };
    
            // 通过ThreadPoolExecutor的submit()方法提交Runnable
            Future<MyResult> resultFuture = threadPoolExecutor.submit(runnable, myResult);
    
            // 获取执行结果
            MyResult finalResult = resultFuture.get();
            // myResult和finalResult的地址实际相同
            Assert.assertEquals(myResult, finalResult);
            // 打印执行结果
            System.out.println(resultFuture.get().getResult());
        }
    
        private static class MyResult {
        
            String result;
    
            public MyResult() {
        }
    
            public MyResult(String result) {
        
                this.result = result;
            }
    
            public String getResult() {
        
                return result;
            }
    
            public void setResult(String result) {
        
                this.result = result;
            }
        }
    
    }

运行测试程序，结果如下。

![在这里插入图片描述][d3c8741778c04f6096b739bd25ec5473.jpeg_pic_center]

实际上`ThreadPoolExecutor`的`submit()`方法无论是提交`Runnable`任务还是`Callable`任务，都是将任务封装成了`RunnableFuture`接口的子类`FutureTask`，然后调用`ThreadPoolExecutor`的`execute()`方法来执行`FutureTask`。

#### 五. 关闭线程池 ####

关闭线程池可以通过`ThreadPoolExecutor`的`shutdown()`方法，但是`shutdown()`方法不会去中断正在执行任务的线程，所以如果线程池里有`Worker`正在执行一个永远不会结束的任务，那么`shutdown()`方法是无法关闭线程池的。示例如下。

public class ThreadPoolExecutorTest {
        
    
        @Test
        public void 通过shutdown关闭线程池() {
        
            // 创建一个线程池
            ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(2, 4,
                    60, TimeUnit.SECONDS, new ArrayBlockingQueue<>(300));
    
            // 创建Runnable对象
            Runnable runnable = new Runnable() {
        
                @Override
                public void run() {
        
                    while (!Thread.currentThread().isInterrupted()) {
        
                        LockSupport.parkNanos(1000 * 1000 * 1000);
                    }
                    System.out.println(Thread.currentThread().getName() + " 被中断");
                }
            };
    
            // 让线程池执行任务
            threadPoolExecutor.execute(runnable);
            threadPoolExecutor.execute(runnable);
    
            // 调用shutdown方法关闭线程池
            threadPoolExecutor.shutdown();
    
            // 等待3秒观察现象
            LockSupport.parkNanos(1000 * 1000 * 1000 * 3L);
        }
    
    }

运行测试程序，会发现在主线程中等待3秒后，也没有得到预期的打印结果。如果上述测试程序中使用`shutdownNow`，则是可以得到预期打印结果的，示例如下。

public class ThreadPoolExecutorTest {
        
    
        @Test
        public void 通过shutdownNow关闭线程池() {
        
            // 创建一个线程池
            ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(2, 4,
                    60, TimeUnit.SECONDS, new ArrayBlockingQueue<>(300));
    
            // 创建Runnable对象
            Runnable runnable = new Runnable() {
        
                @Override
                public void run() {
        
                    while (!Thread.currentThread().isInterrupted()) {
        
                        LockSupport.parkNanos(1000 * 1000 * 1000);
                    }
                    System.out.println(Thread.currentThread().getName() + " 被中断");
                }
            };
    
            // 让线程池执行任务
            threadPoolExecutor.execute(runnable);
            threadPoolExecutor.execute(runnable);
    
            // 调用shutdown方法关闭线程池
            threadPoolExecutor.shutdownNow();
    
            // 等待3秒观察现象
            LockSupport.parkNanos(1000 * 1000 * 1000 * 3L);
        }
    
    }

运行测试程序，打印如下。

![在这里插入图片描述][94cdddc3dfa24703a5886ecfd998291a.jpeg_pic_center]

因为测试程序中的任务是响应中断的，而`ThreadPoolExecutor`的`shutdownNow()`方法会中断所有`Worker`，所以执行`shutdownNow()`方法后，正在运行的任务会响应中断并结束运行，最终线程池关闭。

假如线程池中运行着一个永远不会结束的任务，且这个任务不响应中断，那么无论是`shutdown()`方法还是`shutdownNow()`方法，都是无法关闭线程池的。

### 总结 ###

`ThreadPoolExecutor`的使用总结如下。

1.  通过`ThreadPoolExecutor`的`execute()`方法能够执行`Runnable`任务；
2.  通过`ThreadPoolExecutor`的`submit()`方法能够执行`Runnable`任务和`Callable`任务，并且能够获取异步的执行结果；
3.  `ThreadPoolExecutor`的`submit()`方法会返回一个`Future`对象（实际就是`FutureTask`），如果任务执行过程中发生了异常且未捕获，那么可以通过`Future`的`get()`方法感知到异常；
4.  `ThreadPoolExecutor`的`submit()`方法无论是提交`Runnable`任务还是`Callable`任务，都是将任务封装成了`RunnableFuture`接口的子类`FutureTask`，然后调用`ThreadPoolExecutor`的`execute()`方法来执行`FutureTask`；
5.  关闭线程池时，如果运行的任务可以在有限时间内运行完毕，那么可以使用`shutdown()`方法来关闭线程池，这能够保证在关闭线程池时，正在运行的任务会顺利运行完毕；
6.  关闭线程池时，如果运行的任务永远不会结束但是响应中断，那么可以使用`shutdownNow()`方法来关闭线程池，这种方式不保证任务顺利运行完毕；
7.  如果任务永远不会结束且不响应中断，那么无论是`shutdown()`方法还是`shutdownNow()`方法，都无法关闭线程池。

[2d2125141cd74b9c97338ba55aa418d0.jpeg_pic_center]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/28/c471f6c981d142c98b849ea2333fb047.png
[acc03e00fe574ce0892ed60fb28ef0c7.jpeg_pic_center]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/28/b27dce1965274a059ac5ef423aec06f7.png
[c6367c1f26264e48b6f69eb4259af346.jpeg_pic_center]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/28/94fc40d8d5d442de8ea7aa047a3cd7e3.png
[d3c8741778c04f6096b739bd25ec5473.jpeg_pic_center]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/28/d52d97ffddff4eb38168d6b341ea9bbb.png
[94cdddc3dfa24703a5886ecfd998291a.jpeg_pic_center]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/28/5ced8b62250042bfacd85ee0f0e25395.png