线程池详解

Author: Snailclimb

code/java/ThreadPoolExecutorDemo/.idea/uiDesigner.xml

@@ -0,0 +1,49 @@
+package callable;
+
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.List;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+
+import static common.ThreadPoolConstants.CORE_POOL_SIZE;
+import static common.ThreadPoolConstants.KEEP_ALIVE_TIME;
+import static common.ThreadPoolConstants.MAX_POOL_SIZE;
+import static common.ThreadPoolConstants.QUEUE_CAPACITY;
+
+public class CallableDemo {
+    public static void main(String[] args) {
+        //使用阿里巴巴推荐的创建线程池的方式
+        //通过ThreadPoolExecutor构造函数自定义参数创建
+        ThreadPoolExecutor executor = new ThreadPoolExecutor(
+                CORE_POOL_SIZE,
+                MAX_POOL_SIZE,
+                KEEP_ALIVE_TIME,
+                TimeUnit.SECONDS,
+                new ArrayBlockingQueue<>(QUEUE_CAPACITY),
+                new ThreadPoolExecutor.CallerRunsPolicy());
+
+        List<Future<String>> futureList = new ArrayList<>();
+        Callable<String> callable = new MyCallable();
+        for (int i = 0; i < 10; i++) {
+            //提交任务到线程池
+            Future<String> future = executor.submit(callable);
+            //将返回值 future 添加到 list，我们可以通过 future 获得 执行 Callable 得到的返回值
+            futureList.add(future);
+        }
+        for (Future<String> fut : futureList) {
+            try {
+                System.out.println(new Date() + "::" + fut.get());
+            } catch (InterruptedException | ExecutionException e) {
+                e.printStackTrace();
+            }
+        }
+        //关闭线程池
+        executor.shutdown();
+    }
+}
+

code/java/ThreadPoolExecutorDemo/out/production/ThreadPoolExecutorDemo/ThreadPoolExecutorDemo.class

@@ -0,0 +1,13 @@
+package callable;
+
+import java.util.concurrent.Callable;
+
+public class MyCallable implements Callable<String> {
+
+    @Override
+    public String call() throws Exception {
+        Thread.sleep(1000);
+        //返回执行当前 Callable 的线程名字
+        return Thread.currentThread().getName();
+    }
+}

code/java/ThreadPoolExecutorDemo/out/production/ThreadPoolExecutorDemo/WorkerThread.class

@@ -0,0 +1,11 @@
+package common;
+
+public class ThreadPoolConstants {
+    public static final int CORE_POOL_SIZE = 5;
+    public static final int MAX_POOL_SIZE = 10;
+    public static final int QUEUE_CAPACITY = 100;
+    public static final Long KEEP_ALIVE_TIME = 1L;
+    private ThreadPoolConstants(){
+
+    }
+}

code/java/ThreadPoolExecutorDemo/out/production/ThreadPoolExecutorDemo/callable/CallableDemo.class

@@ -1,14 +1,16 @@
+package threadPoolExecutor;
+
 import java.util.Date;
 
 /**
  * 这是一个简单的Runnable类，需要大约5秒钟来执行其任务。
  * @author shuang.kou
  */
-public class WorkerThread implements Runnable {
+public class MyRunnable implements Runnable {
 
     private String command;
 
-    public WorkerThread(String s) {
+    public MyRunnable(String s) {
         this.command = s;
     }
 

code/java/ThreadPoolExecutorDemo/out/production/ThreadPoolExecutorDemo/callable/MyCallable.class

@@ -1,14 +1,17 @@
+package threadPoolExecutor;
+
 import java.util.concurrent.ArrayBlockingQueue;
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
 
-public class ThreadPoolExecutorDemo {
-    private static final int CORE_POOL_SIZE = 5;
-    private static final int MAX_POOL_SIZE = 10;
-    private static final int QUEUE_CAPACITY = 100;
-    private static final Long KEEP_ALIVE_TIME = 1L;
+import static common.ThreadPoolConstants.CORE_POOL_SIZE;
+import static common.ThreadPoolConstants.KEEP_ALIVE_TIME;
+import static common.ThreadPoolConstants.MAX_POOL_SIZE;
+import static common.ThreadPoolConstants.QUEUE_CAPACITY;
 
 
+public class ThreadPoolExecutorDemo {
+
     public static void main(String[] args) {
 
         //使用阿里巴巴推荐的创建线程池的方式
@@ -23,7 +26,7 @@ public static void main(String[] args) {
 
         for (int i = 0; i < 10; i++) {
             //创建WorkerThread对象（WorkerThread类实现了Runnable 接口）
-            Runnable worker = new WorkerThread("" + i);
+            Runnable worker = new MyRunnable("" + i);
             //执行Runnable
             executor.execute(worker);
         }

code/java/ThreadPoolExecutorDemo/out/production/ThreadPoolExecutorDemo/common/ThreadPoolConstants.class

@@ -355,25 +355,79 @@ ThreadLocal.ThreadLocalMap inheritableThreadLocals = null;
 
 ### 4.1. 为什么要用线程池？
 
-线程池提供了一种限制和管理资源（包括执行一个任务）。 每个线程池还维护一些基本统计信息，例如已完成任务的数量。 
+> **池化技术相比大家已经屡见不鲜了，线程池、数据库连接池、Http 连接池等等都是对这个思想的应用。池化技术的思想主要是为了减少每次获取资源的消耗，提高对资源的利用率。**
 
-这里借用《Java并发编程的艺术》提到的来说一下使用线程池的好处：
+**线程池**提供了一种限制和管理资源（包括执行一个任务）。 每个**线程池**还维护一些基本统计信息，例如已完成任务的数量。
 
-- **降低资源消耗。** 通过重复利用已创建的线程降低线程创建和销毁造成的消耗。
-- **提高响应速度。** 当任务到达时，任务可以不需要的等到线程创建就能立即执行。
-- **提高线程的可管理性。** 线程是稀缺资源，如果无限制的创建，不仅会消耗系统资源，还会降低系统的稳定性，使用线程池可以进行统一的分配，调优和监控。
+这里借用《Java 并发编程的艺术》提到的来说一下**使用线程池的好处**：
+
+- **降低资源消耗**。通过重复利用已创建的线程降低线程创建和销毁造成的消耗。
+- **提高响应速度**。当任务到达时，任务可以不需要的等到线程创建就能立即执行。
+- **提高线程的可管理性**。线程是稀缺资源，如果无限制的创建，不仅会消耗系统资源，还会降低系统的稳定性，使用线程池可以进行统一的分配，调优和监控。
 
 ### 4.2. 实现Runnable接口和Callable接口的区别
 
-如果想让线程池执行任务的话需要实现的Runnable接口或Callable接口。 Runnable接口或Callable接口实现类都可以被ThreadPoolExecutor或ScheduledThreadPoolExecutor执行。两者的区别在于 Runnable 接口不会返回结果但是 Callable 接口可以返回结果。
+`Runnable`自Java 1.0以来一直存在，但`Callable`仅在Java 1.5中引入,目的就是为了来处理`Runnable`不支持的用例。**`Runnable` 接口**不会返回结果或抛出检查异常，但是**`Callable` 接口**可以。所以，如果任务不需要返回结果或抛出异常推荐使用 **`Runnable` 接口**，这样代码看起来会更加简洁。
+
+工具类 `Executors` 可以实现 `Runnable` 对象和 `Callable` 对象之间的相互转换。（`Executors.callable（Runnable task`）或 `Executors.callable（Runnable task，Object resule）`）。
+
+`Runnable.java`
+
+```java
+@FunctionalInterface
+public interface Runnable {
+   /**
+    * 被线程执行，没有返回值也无法抛出异常
+    */
+    public abstract void run();
+}
+```
 
-  **备注：** 工具类`Executors`可以实现`Runnable`对象和`Callable`对象之间的相互转换。（`Executors.callable（Runnable task）`或`Executors.callable（Runnable task，Object resule）`）。
+`Callable.java`
+
+```java
+@FunctionalInterface
+public interface Callable<V> {
+    /**
+     * 计算结果，或在无法这样做时抛出异常。
+     * @return 计算得出的结果
+     * @throws 如果无法计算结果，则抛出异常
+     */
+    V call() throws Exception;
+}
+```
 
 ### 4.3. 执行execute()方法和submit()方法的区别是什么呢？
 
-  1)**`execute()` 方法用于提交不需要返回值的任务，所以无法判断任务是否被线程池执行成功与否；**
+1. **`execute()`方法用于提交不需要返回值的任务，所以无法判断任务是否被线程池执行成功与否；**
+2. **`submit()`方法用于提交需要返回值的任务。线程池会返回一个 `Future` 类型的对象，通过这个 `Future` 对象可以判断任务是否执行成功**，并且可以通过 `Future` 的 `get()`方法来获取返回值，`get()`方法会阻塞当前线程直到任务完成，而使用 `get（long timeout，TimeUnit unit）`方法则会阻塞当前线程一段时间后立即返回，这时候有可能任务没有执行完。
+
+我们以**`AbstractExecutorService`**接口中的一个 `submit` 方法为例子来看看源代码：
 
-  2)**`submit()` 方法用于提交需要返回值的任务。线程池会返回一个Future类型的对象，通过这个Future对象可以判断任务是否执行成功**，并且可以通过future的get()方法来获取返回值，get()方法会阻塞当前线程直到任务完成，而使用 `get（long timeout，TimeUnit unit）`方法则会阻塞当前线程一段时间后立即返回，这时候有可能任务没有执行完。
+```java
+    public Future<?> submit(Runnable task) {
+        if (task == null) throw new NullPointerException();
+        RunnableFuture<Void> ftask = newTaskFor(task, null);
+        execute(ftask);
+        return ftask;
+    }
+```
+
+上面方法调用的 `newTaskFor` 方法返回了一个 `FutureTask` 对象。
+
+```java
+    protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) {
+        return new FutureTask<T>(runnable, value);
+    }
+```
+
+我们再来看看`execute()`方法：
+
+```java
+    public void execute(Runnable command) {
+      ...
+    }
+```
 
 ### 4.4. 如何创建线程池