30、ThreadPool线程池

发表于 分类于
本文字数： 4.4k 阅读时长 ≈ 4 分钟
30、ThreadPool线程池

10.1 线程池概述

线程池（ThreadPool）：一种线程使用模式。线程过多会带来调度开销，进而影响缓存局部性和整体性能。而线程池维护着多个线程，等待着监督管理者分配可并发执行的任务，这避免了在处理短时间任务时创建与销毁线程的代价。线程池不仅能保证内核的充分使用，还能防止过分调度。

线程池的优势：线程池的工作是控制运行的线程数量。处理过程中将任务放入队列，然后创建线程来进行执行。如果线程数量超多了最大数量，超出数量的线程排队等候。等其他线程执行完毕后，再从队列中取出任务来进行执行。

它的主要特点：

  • 降低资源消耗，通过重复利用已创建的线程降低线程创建和销毁造成的消耗。
  • 提高响应速度，当任务到达时，任务可以不需要等待线程创建就可以立即执行。
  • 提高线程的可管理性：线程是稀缺资源，如果无限制地创建，不仅会消耗系统资源，还会降低系统的稳定性，使用线程池可以进行统一的分配、调优和监控。
  • Java中的线程池是通过Executor框架实现的，该框架中用到了Executor、Executors、ExecutorService、ThreadPoolExecutor这几个类。

10.2 线程池的架构

10.3 线程池使用方式

1、 Executors.newFixedThreadPool(int)；

创建一个线程池，线程池中有固定的n个线程

特点：

  • 线程池中线程的数量是一定的，可以很好地控制线程的并发量
  • 线程可以重复利用。在显式关闭之前，都将一直存在
  • 当线程被使用完时，新来的任务需要等待

2、 Executors.newSingleThreadExecutor()；

一池一线程，一个任务一个任务地执行

3、 Executors.newCachedThreadPool()；

线程池根据需求创建线程，可扩容

代码演示：

一池多线程：

package pool;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class ThreadPoolDemo1 {

    public static void main(String[] args) {
        //一池五线程
        ExecutorService threadPool1 = Executors.newFixedThreadPool(5);
        //十个请求
        try {
            for(int i = 0; i < 10; i++) {
                threadPool1.execute(()->{
                    System.out.println(Thread.currentThread().getName() + "正在办理业务");
                });
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            threadPool1.shutdown();
        }
    }
}

输出：

可以发现，最多出现5个线程，说明线程池是重复利用已经创建的线程。

一池一线程：

package pool;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class ThreadPoolDemo2 {

    public static void main(String[] args) {
        //一池一线程
        ExecutorService threadpool2 = Executors.newSingleThreadExecutor();

        try{
            for(int i = 0; i < 10; i++) {
                threadpool2.execute(()->{
                    System.out.println(Thread.currentThread().getName() + "正在办理业务");
                });
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            threadpool2.shutdown();
        }
    }
}

输出：

说明确实是只有一个线程，然后被反复利用。

一池可扩容线程：

package pool;

import java.lang.reflect.Executable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class ThreadPoolDemo3 {

    public static void main(String[] args) {

        ExecutorService threadPool3 = Executors.newCachedThreadPool();

        try {
            for(int i = 0; i < 10; i++) {
                threadPool3.execute(()->{
                    System.out.println(Thread.currentThread().getName() + "正在办理业务");
                });
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            threadPool3.shutdown();
        }
    }
}

输出：

10.4 创建线程池的7个参数

public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue,
                              ThreadFactory threadFactory,
                              RejectedExecutionHandler handler)

1、 corePoolSize；

常驻（核心）线程数量

2、 maximumPoolSize；

最大线程数量

3、 keepAliveTime；

当线程数大于核心时，此为在关闭前多余的空闲线程等待新任务的最长时间，也称为缓存时间

4、 unit；

设置最大存活时间时的单位

5、 workQueue；

执行前用于保持任务的阻塞队列。此队列仅保持由 execute方法提交的 Runnable任务。

6、 threadFactory；

线程工厂，用于创建线程

7、 handler；

拒绝策略

定义固定长度线程池：

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

定义一池一线程：

public static ExecutorService newSingleThreadExecutor() {
        return new FinalizableDelegatedExecutorService
            (new ThreadPoolExecutor(1, 1,
                                    0L, TimeUnit.MILLISECONDS,
                                    new LinkedBlockingQueue<Runnable>()));
    }

定义可扩容线程池：

public static ExecutorService newCachedThreadPool() {
        return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                      60L, TimeUnit.SECONDS,
                                      new SynchronousQueue<Runnable>());
    }

10.5 线程池底层工作流程

请看链接：线程池底层工作原理，7张图讲清楚线程池底层的工作原理。

4种拒绝策略：

  • AbortPolicy：直接抛出RejectedExecutionException异常阻止系统正常运行
  • CallerRunsPolicy：既不抛弃任务，也不抛出异常，而是将某些任务回退给调用者，从而降低任务的流量。
  • DiscardOldestPolicy：抛弃等待队列中最先等待的任务，然后把当前任务加入队列中
  • DiscardPolicy：既不处理也不抛出异常。如果允许任务丢弃，这是最好的方法。

10.6 自定线程池

由Executors创建的线程池的弊端：

1、 FixedThreadPool和SingleThreadPool，允许请求队列的长读为Integer.MAX_VALUE，有可能造成大量请求的堆积，最终造成OOM；

2、 CachedThreadPool和ScheduledThreadPool，允许创建的线程的最大数目为Integer.MAX_VALUE，可能会创建大量的线程，最终造成OOM；

package pool;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class ThreadPoolDemo4 {

    public static void main(String[] args) {

        ThreadPoolExecutor threadPool4 = new ThreadPoolExecutor(
                2,
                5,
                2L,
                TimeUnit.SECONDS,
                new ArrayBlockingQueue<>(3),
                Executors.defaultThreadFactory(),
                new ThreadPoolExecutor.AbortPolicy()
        );

        try {
            for(int i = 0; i < 10; i++) {
                threadPool4.execute(()->{
                    System.out.println(Thread.currentThread().getName() + "正在处理任务");
                });
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            threadPool4.shutdown();
        }
    }
}