深入理解java线程池

清疚 2023-07-18 11:29 273阅读 0赞

### 线程池的优势 ###

1.  降低系统资源消耗，通过重用已存在的线程，降低线程创建和销毁造成的消耗；
2.  提高系统响应速度，当有任务到达时，通过复用已存在的线程，无需等待新线程的创建便能立即执行；
3.  方便线程并发数的管控。因为线程若是无限制的创建，可能会导致内存占用过多而产生OOM，并且会造成cpu过度切换（cpu切换线程是有时间成本的（需要保持当前执行线程的现场，并恢复要执行线程的现场））。
4.  提供更强大的功能，延时定时线程池。

### java中提供的线程池 ###

Executors类提供了4种不同的线程池：newCachedThreadPool, newFixedThreadPool, newScheduledThreadPool, newSingleThreadExecutor  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2Z5ajEzOTI1NDc1OTU3_size_16_color_FFFFFF_t_70]

> **newCachedThreadPool**：用来创建一个可以无限扩大的线程池，适用于负载较轻的场景，执行短期异步任务。（可以使得任务快速得到执行，因为任务时间执行短，可以很快结束，也不会造成cpu过度切换）

> **newFixedThreadPool**：创建一个固定大小的线程池，因为采用无界的阻塞队列，所以实际线程数量永远不会变化，适用于负载较重的场景，对当前线程数量进行限制。（保证线程数可控，不会造成线程过多，导致系统负载更为严重）

> **newSingleThreadExecutor**：创建一个单线程的线程池，适用于需要保证顺序执行各个任务。

> **newScheduledThreadPool**：适用于执行延时或者周期性任务。

代码示例：

public class ExecutorsTest { 
    
        public static void main(String[] args) { 
            // 创建固定数据的线程的线程池,线程池数量为5
    // ExecutorService executorService = Executors.newFixedThreadPool(5);
            // 创建一个无线扩大的线程池
    // ExecutorService executorService = Executors.newCachedThreadPool();
            // 创建一个单线程的线程池，适用于需要保证顺序执行各个任务
            ExecutorService executorService = Executors.newSingleThreadExecutor();
            try { 
                // 加入10个任务
                for (int i = 0; i < 10; i++) { 
                    final int tempInt = i;
                    executorService.execute(new Runnable() { 
                        @Override
                        public void run() { 
                            System.out.println(Thread.currentThread().getName() + "->" + tempInt);
                        }
                    });
                }
            } finally { 
                executorService.shutdown();
            }
        }
    }

java提供的这个4个线程池。底层实际上都是调用的ThreadPoolExecutor类的构造方法：

public static ExecutorService newFixedThreadPool(int nThreads) { 
            return new ThreadPoolExecutor(nThreads, nThreads,
                                          0L, TimeUnit.MILLISECONDS,
                                          new LinkedBlockingQueue<Runnable>());
        }

**但是更阿里巴巴的开发手册，工作中一般不用Executors提供线程池**:  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2Z5ajEzOTI1NDc1OTU3_size_16_color_FFFFFF_t_70 1]

#### execute()和submit()方法 ####

1.  execute()，执行一个任务，没有返回值。
2.  submit()，提交一个线程任务，有返回值。
    
    > 1.  submit(Callable task)能获取到它的返回值，通过future.get()获取（阻塞直到任务执行完）。一般使用FutureTask+Callable配合使用（IntentService中有体现）。
    > 2.  submit(Runnable task, T result)能通过传入的载体result间接获得线程的返回值。  
    >     submit(Runnable task)则是没有返回值的，就算获取它的返回值也是null。
    > 3.  Future.get方法会使取结果的线程进入阻塞状态，知道线程执行完成之后，唤醒取结果的线程，然后返回结果。  
    >     。

### 线程池的七大参数 ###

有的只说5个参数或者6个，但是实际上`ThreadPoolExecutor`是有7个参数的，如下源码 ：

public ThreadPoolExecutor(int corePoolSize,
                                  int maximumPoolSize,
                                  long keepAliveTime,
                                  TimeUnit unit,
                                  BlockingQueue<Runnable> workQueue) { 
            this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
                 Executors.defaultThreadFactory(), defaultHandler);
    
    
    	public ThreadPoolExecutor(int corePoolSize,
                                  int maximumPoolSize,
                                  long keepAliveTime,
                                  TimeUnit unit,
                                  BlockingQueue<Runnable> workQueue,
                                  ThreadFactory threadFactory,
                                  RejectedExecutionHandler handler)

构造器ThreadPoolExecutor，里面调用了带有7个参数的构造器this

下面来看看每一个参数的具体意思：

`corePoolSize`：线程池中常驻的核心线程数

> 当向线程池提交一个任务时，若线程池已创建的线程数小于corePoolSize，即便此时存在空闲线程，也会通过创建一个新线程来执行该任务，直到已创建的线程数大于或等于corePoolSize时

`maximumPoolSize`:线程池能够容纳同时执行的最大线程数，此值必须大于等于1

> 线程池所允许的最大线程个数。当队列满了，且已创建的线程数小于maximumPoolSize，则线程池会创建新的线程来执行任务。另外，对于无界队列，可忽略该参数

`keepAliveTime`:多余的空闲线程的存活时间。

> 当线程池中线程数大于核心线程数时，线程的空闲时间如果超过线程存活时间，那么这个线程就会被销毁，直到线程池中的线程数小于等于核心线程数

`TimeUnit` ：keepAliveTime时间单位

`workQueue`：任务队列，被提交单尚未被执行的任务。

`threadFactory`：表示生产线程池中工作线程的线程工厂，用于创建线程一般用默认即可。

> 用于创建新线程。threadFactory创建的线程也是采用new Thread()方式，threadFactory创建的线程名都具有统一的风格：pool-m-thread-n（m为线程池的编号，n为线程池内的线程编号）

`handler`：拒绝策略，表示当队列满了，并且工作线程大于等于最大的线程数，就启用拒绝策略

### 线程池的执行流程 ###

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2Z5ajEzOTI1NDc1OTU3_size_16_color_FFFFFF_t_70 2]

1.  判断核心线程池是否已满，没满则创建一个新的工作线程来执行任务。已满则放入任务队列。
2.  判断任务队列是否已满，没满则将新提交的任务添加在工作队列，已满则继续3
3.  判断整个线程池是否已满，没满则创建一个新的工作线程来执行任务，已满则执行饱和策略。
    
    > 1、判断线程池中当前线程数是否大于核心线程数，如果小于，在创建一个新的线程来执行任务，如果大于则  
    > 2、判断任务队列是否已满，没满则将新提交的任务添加在工作队列，已满则。  
    > 3、判断线程池中当前线程数是否大于最大线程数，如果小于，则创建一个新的线程来执行任务，如果大于，则执行饱和策略。）

**线程池为什么要使用阻塞队列而不使用非阻塞队列？**

1.  阻塞队列可以保证任务队列中没有任务时阻塞获取任务的线程，使得线程进入wait状态，释放cpu资源。
2.  当队列中有任务时才唤醒对应线程从队列中取出消息进行执行。  
    使得在线程不至于一直占用cpu资源。
    
    （线程执行完任务后通过循环再次从任务队列中取出任务进行执行，代码片段如下  
    while (task != null || (task = getTask()) != null) \{\}）。

### 线程池的4种拒绝策略及其线程池自己实现 ###

1.  AbortPolicy(默认)：直接抛出RejectedExecutionException异常阻止系统正常运行
2.  CallerRunPolicy："调用者运行"一种调节机制，该策略既不会抛弃任务，也不会抛出异常，而是将某些任务回退给调用者，
3.  DiscardOldestPolicy：抛弃队列中等待最久的任务，然后把当前任务加入到队列中尝试再次提交。
4.  DiscardPolicy：直接丢弃任务，不予任何处理，也不抛出异常，如果允许任务丢失，这是一种最好的方案

代码示例：  
**拒绝策略是：AbortPolicy**

public class MyThreadPool { 
        /** * 该线程池的核心线程数2，最大线程数为5，任务队列容量为3，拒绝策略是：AbortPolicy * @return */
        public static ExecutorService getThreadPool() { 
            return new ThreadPoolExecutor(
                    2,
                    5,
                    1L,
                    TimeUnit.SECONDS,
                    new LinkedBlockingQueue<>(3),
                    Executors.defaultThreadFactory(),
                    new ThreadPoolExecutor.AbortPolicy());
        }
    
        public static void main(String[] args) { 
            ExecutorService threadPool = MyThreadPool.getThreadPool();
            try { 
                // 加入10个任务
                for (int i = 0; i < 10; i++) { 
                    final int tempInt = i;
                    threadPool.execute(new Runnable() { 
                        @Override
                        public void run() { 
                            System.out.println(Thread.currentThread().getName() + "->" + tempInt);
                        }
                    });
                }
            } finally { 
                threadPool.shutdown();
            }
        }
    }

如上代码:线程池的核心线程数2，最大线程数为5，任务队列容量为3，拒绝策略是：AbortPolicy。  
当提交10 个任务时，运行结果：

pool-1-thread-4->6
    Exception in thread "main" java.util.concurrent.RejectedExecutionException: Task com.fangyajun.javasduty.juc.executors.MyThreadPool$1@135fbaa4 rejected from java.util.concurrent.ThreadPoolExecutor@45ee12a7[Running, pool size = 5, active threads = 3, queued tasks = 0, completed tasks = 5]
    pool-1-thread-2->2
    	at java.util.concurrent.ThreadPoolExecutor$AbortPolicy.rejectedExecution(ThreadPoolExecutor.java:2047)
    pool-1-thread-2->4
    pool-1-thread-3->5
    	at java.util.concurrent.ThreadPoolExecutor.reject(ThreadPoolExecutor.java:823)
    pool-1-thread-1->0
    pool-1-thread-4->3
    	at java.util.concurrent.ThreadPoolExecutor.execute(ThreadPoolExecutor.java:1369)
    pool-1-thread-5->7
    	at com.fangyajun.javasduty.juc.executors.MyThreadPool.main(MyThreadPool.java:28)

出抛出异常，终止程序执行。

**拒绝策略是：CallerRunPolicy**

/** * 该线程池的核心线程数2，最大线程数为5，任务队列容量为3，拒绝策略是：CallerRunsPolicy * @return */
        public static ExecutorService getThreadPool() { 
    
            return new ThreadPoolExecutor(
                    2,
                    5,
                    1L,
                    TimeUnit.SECONDS,
                    new LinkedBlockingQueue<>(3),
                    Executors.defaultThreadFactory(),
                    new ThreadPoolExecutor.CallerRunsPolicy());
        }
    
        public static void main(String[] args) { 
            //
            ExecutorService threadPool = MyThreadPool.getThreadPool();
            try { 
                // 加入10个任务
                for (int i = 0; i < 10; i++) { 
                    final int tempInt = i;
                    threadPool.execute(new Runnable() { 
                        @Override
                        public void run() { 
                            System.out.println(Thread.currentThread().getName() + "->" + tempInt);
                        }
                    });
                }
            } finally { 
                threadPool.shutdown();
            }
        }
    }

运行结果：

pool-1-thread-1->0
    pool-1-thread-5->7
    pool-1-thread-4->6
    pool-1-thread-4->4
    pool-1-thread-3->5
    pool-1-thread-2->1
    main->8
    pool-1-thread-5->3
    pool-1-thread-1->2
    pool-1-thread-4->9

从结果可以看出CallerRunPolicy策略既不会抛弃任务，也不会抛出异常，而是将某些任务回退给main线程执行。

**拒绝策略：DiscardOldestPolicy**

public class MyThreadPool { 
        /** * 该线程池的核心线程数2，最大线程数为5，任务队列容量为3，拒绝策略是：CallerRunsPolicy * @return */
        public static ExecutorService getThreadPool() { 
    
            return new ThreadPoolExecutor(
                    2,
                    5,
                    1L,
                    TimeUnit.SECONDS,
                    new LinkedBlockingQueue<>(3),
                    Executors.defaultThreadFactory(),
                    new ThreadPoolExecutor.DiscardOldestPolicy());
        }
    
        public static void main(String[] args) { 
            //
            ExecutorService threadPool = MyThreadPool.getThreadPool();
            try { 
                // 加入10个任务
                for (int i = 0; i < 10; i++) { 
                    final int tempInt = i;
                    threadPool.execute(new Runnable() { 
                        @Override
                        public void run() { 
                            System.out.println(Thread.currentThread().getName() + "->" + tempInt);
                        }
                    });
                }
            } finally { 
                threadPool.shutdown();
            }
        }
    }

运行结果：

pool-1-thread-1->0
    pool-1-thread-2->1
    pool-1-thread-3->5
    pool-1-thread-3->9
    pool-1-thread-2->8
    pool-1-thread-5->7
    pool-1-thread-1->4
    pool-1-thread-4->6

从结果可以看出，一共10个任务，结果就执行了8个，2个任务被丢弃了。

DiscardPolicy拒绝策略是当无法处理就直接丢弃任务，在此不代码举例了。

**使用场景：**  
四种拒绝策略是相互独立无关的，选择何种策略去执行，还得结合具体的业务场景

### 如何配置线程池 ###

**CPU密集型任务**  
尽量使用较小的线程池，一般为CPU核心数+1。 因为CPU密集型任务使得CPU使用率很高，若开过多的线程数，会造成CPU过度切换。

**IO密集型任务**  
可以使用稍大的线程池，一般为2\*CPU核心数。 IO密集型任务CPU使用率并不高，因此可以让CPU在等待IO的时候有其他线程去处理别的任务，充分利用CPU时间。

**混合型任务**  
可以将任务分成IO密集型和CPU密集型任务，然后分别用不同的线程池去处理。 只要分完之后两个任务的执行时间相差不大，那么就会比串行执行来的高效。  
因为如果划分之后两个任务执行时间有数据级的差距，那么拆分没有意义。  
因为先执行完的任务就要等后执行完的任务，最终的时间仍然取决于后执行完的任务，而且还要加上任务拆分与合并的开销，得不偿失。、

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2Z5ajEzOTI1NDc1OTU3_size_16_color_FFFFFF_t_70]: /images/20230528/f65181f5320f462b956cae478482afc0.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2Z5ajEzOTI1NDc1OTU3_size_16_color_FFFFFF_t_70 1]: /images/20230528/0c63274619c44157b94d896d7eeb167d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2Z5ajEzOTI1NDc1OTU3_size_16_color_FFFFFF_t_70 2]: /images/20230528/3e7ca750089a43ad814d686c65233edc.png