多线程线程池

Posted 2022-12-30 吞吞吐吐大魔王

文章目录

• 1. 线程池介绍
• 2. 标准库中的线程池使用
• • 2.1 ThreadPoolExecutor
  • • 2.1.1 构造方法和参数说明
    • 2.1.2 线程工厂
    • 2.1.3 拒绝策略
    • 2.1.4 关闭线程池
    • 2.1.5 代码示例
  • 2.2 Executors
  • • 2.2.1 四种功能线程池
    • 2.2.2 代码示例
    • 2.2.3 Executors 4 个功能线程的弊端
• 3. 实现线程池

1. 线程池介绍

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

线程池的优点：

• 降低资源消耗

  通过重复利用自己已创建的线程降低线程创建和销毁造成的消耗。

• 提高响应速度

  当任务到达时，任务可以不需要等到线程创建就能立即执行。

• 提高线程的可管理性

  线程时稀缺资源，如果无限制的创建，不仅会消耗系统资源，还会降低系统的稳定性，使用线程池就可以进行统一的分配、调度和监控。

2. 标准库中的线程池使用

2.1 ThreadPoolExecutor

2.1.1 构造方法和参数说明

Java 标准库中内置了线程池的实现类 ThreadPoolExecutor，不过这个类的使用比较复杂，构造方法有如下四个：

这些构造方法中都有很多参数，汇总如下：

参数	是否必须	说明	补充
corePoolSize	是	核心线程数	默认情况下，核心线程会一直存活，但是当将 allowCoreThreadTimeout 设置为 true 时，核心线程也会超时回收。
maximumPoolSize	是	线程所能容纳的最大线程数	当活跃线程数达到该数值后，后续的新任务将会阻塞。
keepAliveTime	是	线程闲置超时时长	如果超过该时长，非核心线程就会被回收。如果将 allowCoreThreadTimeout 设置为 true 时，核心线程也会超时回收。
unit	是	指定 keepAliveTime 参数的时间单位	常用的有：TimeUnit.MILLISECONDS（毫秒）、TimeUnit.SECONDS（秒）、TimeUnit.MINUTE（分）。
workQueue	是	任务队列	通过线程池的 execute() 方法提交的 Runnable 对象将存储在该参数中。其采用阻塞队列实现。
threadFactory	否	线程工厂	用于指定线程池创建新线程的方式。
handler	否	拒绝策略	当线程池和队列都满了，则表明需要执行饱和策略。

2.1.2 线程工厂

线程工厂 threadFactory 是用来执行线程池创建新线程的方式的，由同一个 threadFactory 创建的线程，属于同一个 ThreadGroup，创建的线程优先级都为 Thread.NORM_PRIORITY，以及是非守护进程状态。threadFactory创建的线程是采用 new Thread() 方式，threadFactory创建的线程名都具有统一的风格：pool-m-thread-n（m为线程池的编号，n为线程池内的线程编号）。

线程工厂指定创建线程的方式，需要实现 ThreadFactory 接口，并重写 newThread(Runnable r) 方法。线程工厂这个参数可以不用指定，Executors 框架已经为我们实现了一个默认的线程工厂：

private static class DefaultThreadFactory implements ThreadFactory 
    private static final AtomicInteger poolNumber = new AtomicInteger(1);
    private final ThreadGroup group;
    private final AtomicInteger threadNumber = new AtomicInteger(1);
    private final String namePrefix;
 
    DefaultThreadFactory() 
        SecurityManager s = System.getSecurityManager();
        group = (s != null) ? s.getThreadGroup() :
                              Thread.currentThread().getThreadGroup();
        namePrefix = "pool-" +
                      poolNumber.getAndIncrement() +
                     "-thread-";
    
 
    public Thread newThread(Runnable r) 
        Thread t = new Thread(group, r,
                              namePrefix + threadNumber.getAndIncrement(),
                              0);
        if (t.isDaemon())
            t.setDaemon(false);
        if (t.getPriority() != Thread.NORM_PRIORITY)
            t.setPriority(Thread.NORM_PRIORITY);
        return t;
    

2.1.3 拒绝策略

拒绝策略是当线程和队列都满了的时候，即线程池饱和的时候，要执行的策略。拒绝策略需要实现 RejectedExecutionHandler 接口，并重写 rejectedExecution(Runnable r, ThreadPoolExecutor executor) 方法。拒绝策略这个参数可以不用指定，Executors 框架已经实现了如下4种拒绝策略：

拒绝策略	说明
ThreadPoolExecutor.AbortPolicy	（默认策略）处理程序遭到拒绝，则直接抛出运行时异常 RejectedExecutionException。
ThreadPoolExecutor.CallerRunsPolicy	调用者所在线程来运行该任务，此策略提供简单的反馈控制机制，能够减缓新任务的提交速度。
ThreadPoolExecutor.DiscardPolicy	无法执行的任务将被删除，但是不会抛出异常。
ThreadPoolExecutor.DiscardOldestPolicy	如果执行程序尚未关闭，则位于工作队列头部的任务将被删除，然后重新尝试执行任务（如果再次失败，则重复此过程）。

2.1.4 关闭线程池

关闭线程池方式	说明
shutdown()	将线程池状态设置为 SHUTDOWN 状态，然后中断没有正在执行任务的线程。
shutdownNow()	将线程池的状态设置成 STOP 状态，然后中断所有任务（包括正在执行的）的线程，并返回等待执行任务的列表。

2.1.5 代码示例

public class Demo17 
    // 创建线程池
    private static ThreadPoolExecutor threadPool = new ThreadPoolExecutor(
            4, // 核心线程数
            10, // 最大线程数
            300,  // 线程闲置超时时长
            TimeUnit.SECONDS, // 时间单位
            new LinkedBlockingQueue<>()); // 任务队列

    public static void main(String[] args) 
        // 向线程池提交任务
        threadPool.execute(new Runnable() 
            @Override
            public void run() 
                // 线程执行的具体任务
            
        );
        
        // 关闭线程
        threadPool.shutdownNow();
    

2.2 Executors

2.2.1 四种功能线程池

Java 标准库中也提供了一个封装版本的线程池 Executors，这个类对 ThreadPoolExecutor 的复杂的参数进行了不同的封装，最终提供了四个静态方法，通过以下四种方法就能够创建出具体策略的线程池：

方法	说明	corePoolSize	maximumPoolSize	keepAliveTime	workQueue
newFixedThreadPool	创建一个指定大小的线程池。可控制线程的最大并发数，超出的线程会在 LinkedBlockingQueue 阻塞队列中等待。	需指定	同左	0	LinkedBlockingQueue
newCachedThreadPool	创建一个可缓存的线程数动态变化的无界线程池，如果线程池长度超过处理需要，可灵活回收空线程，若无可回收，则新建线程。当当线程池中的线程空闲时间超过60s，则会自动回收该线程，当任务超过线程池的线程数则创建新的线程。	0	Integer.MAX_VALUE	60s	SynchronousQueue
newSingleThreadExecutor	创建只包含单个线程的线程池。	1	1	0	LinkedBlockingQueue
newScheduledThreadPool	创建一个定长的线程池，可以指定线程池核心线程数，支持定时及周期性任务的执行。	需指定	Integer.MAX_VALUE	0	DelayedWorkQueue

以上四种创建线程的方法的线程工厂的默认类为 DefaultThreadFactory，默认的拒绝策略为 ThreadPoolExecutor.AbortPolicy。

2.2.2 代码示例

public class Demo18 

    public static void main(String[] args) 
        // 创建一个线程池
        ExecutorService threadPool = Executors.newFixedThreadPool(10);

        // 通过 submit 将任务添加到线程池中
        threadPool.submit(new Runnable() 
            @Override
            public void run() 
                System.out.println("具体任务");
            
        );
        
        // 关闭线程
        threadPool.shutdown();
    

2.2.3 Executors 4 个功能线程的弊端

• newFixedThreadPool 和 newSingleThreadExecutor：主要问题是堆积的请求处理队列均采用 LinkedBlockingQueue，可能会耗费非常大的内存，甚至 OOM。
• newCachedThreadPool 和 newScheduledThreadPool：主要问题是线程数最大是 Integer.MAX_VALUE，可能会创建数量非常多的线程，甚至 OOM。

OOM（Out Of Memory）表示内存用完了。当 JVM 因为没有足够的内存来为对象分配空间并且垃圾回收器也已经没有空间可回收时，就会抛出这个 java.lang.OutOfMemoryError。

3. 实现线程池

实现线程池，需要完成以下几个工作：

1. 通过阻塞队列来组织要完成的任务
2. 实现一个将任务注册到线程池的任务队列中的方法
3. 创建一个工作线程类，能够获取到任务队列，并能够反复的从任务队列中获取任务来执行
4. 通过数据结构将工作线程组织起来，并通过线程池的构造方法来确定工作线程的个数

class MyThreadPool 
    // 描述任务，直接使用 Runnable
    // 组织任务，使用一个阻塞队列来保存若干任务
    private BlockingQueue<Runnable> runnables = new LinkedBlockingQueue<>();

    // 描述一个工作线程
    static class Worker extends Thread 
        private BlockingQueue<Runnable> runnables = null;

        public Worker(BlockingQueue<Runnable> runnables)
            this.runnables = runnables;
        

        @Override
        public void run()
            // 一个线程要做的工作就是反复的从队列中获取任务，执行任务
            while (true)
                try 
                    // 如果任务队列为空，就会产生阻塞，直到有元素加入
                    Runnable take = runnables.take();
                    take.run();
                 catch (InterruptedException e) 
                    e.printStackTrace();
                
            
        
    

    // 组织若干个工作线程
    private List<Worker> workerList = new ArrayList<>();

    // 通过构造方法，指定线程池的个数
    public MyThreadPool(int n)
        for(int i=0; i<n; i++)
            Worker worker = new Worker(runnables);
            // 开启每个线程
            worker.start();
            workerList.add(worker);
        
    

    // 将任务注册到线程池中
    public void submit(Runnable runnable)
        try 
            runnables.put(runnable);
         catch (InterruptedException e) 
            e.printStackTrace();