目录

自定义线程池

BlockingQueue 的编写思路

阻塞队列方法编写

poll增强（待更新）

自定义线程池的实现（待更新）

任务提交 和 Worker的实现（待更新）

take死等和 poll超时（待更新）

当队伍线程已满（待更新）

offer增强（待更新）

拒绝策略（待更新）

本篇学习笔记利用自定义线程池去更深入的理解JDK的线程池API。

线程是一种系统资源，每创建一个新的线程，就要占用一定的内存，当在高并发的情况下，来了很多任务，如果要为每个任务创建一个线程，将耗费很大的系统资源，对内存的占用很大。

线程并不是越多越好，频繁的上下文切换也降低系统性能，尤其在高并发的情况下。

出于以上的原因，有了线程池的出现。

线程池可以让线程重复利用，既可以内存的占用，也可以减少线程的数量，避免频繁发生上下文的切换。

自定义线程池

自定义线程池需要两个组件，ThreadPool（线程池），BlockingQueue（阻塞队列），这两个组件。

ThreadPool里会有一些线程，让这些线程能够重复的利用。

Blocking Queue，它体现的是在一种生产者和消费者模式下，平衡它们之间差异的组件。

ThreadPool里的线程 相当于消费者，它去不断的获取任务和执行任务。除此以外还有生产者线程，它会源源不断的产生新的任务，在这个模式下，它们的速率是不一致的。

比如生产者，如果迟迟没有提交新任务，线程池里的线程就需要在阻塞队列里等待去执行任务。

又或者，生产者产生了太多的任务，线程们忙不过来，多出来的任务也需要放进阻塞队列里面等待被执行。

所以阻塞队列再此扮演了很重要的角色，它是生产者和消费者之间的桥梁。

先从阻塞队列入手，来自定义线程池。

BlockingQueue 的编写思路

通过对BlockingQueue的结构理出如下思路，看看编写BlockingQueue需要些什么东西 ：

BlockingQueue 是一个任务队列，用于存放任务对象。这个队列的数据结构应该选择一个链表，因为队列的特性是先进先出。

链表结构可以用 Deque 类，它是一个双向链表，通过ArrayDeque来实现链表，因为它的效率高于LinkedList。

还需要锁，线程池有多个线程，它们都需要从任务队列头部获取任务，多个线程不可能同时获取正任务，只有一个线程能获取成功，其他线程则需要等待拿去下个任务。所以需要利用锁来保护头部元素，同理队列尾部也应该受到锁的保护，因为可能有多个线程都往队列里添加新任务。

锁的实现选择 ReentrantLock，这样比较灵活。

还需要两个变量，一个消费者的条件变量，和一个生产者的条件变量，消费者在没有任务的时候需要等待，而生产者也不能无限制的往队列里添加元素，阻塞队列都有容量限制，在阻塞队列容量已满的时候，它也需要去等待。因此需要消费者和生产者的条件变量。

还需要为阻塞队列定义容量，因为阻塞队列要有容量限制。

阻塞队列还需要一些个方法，一个方法来获取元素，一个方法来添加元素，一个方法来获取阻塞队列的大小，因为需要知道队列里的任务还有多少个。

阻塞队列需要的属性和方法一览：

• 阻塞队列
• 锁
• 生产者和消费者的条件变量
• 阻塞获取的方法
• 阻塞添加的方法
• 得到队列任务多少的方法

结构一览：

 类一览（仅做结构展示，方法还未实现）：

/**
 * @author Claw
 * @date 2022/3/8 22:49.
 */
public class BlockingQueue<T> {
    // 任务队列
    private Deque<T> queue = new LinkedList();
    // 锁
    private ReentrantLock lock = new ReentrantLock();
    // 生产者条件变量
    private Condition fullWaitSet = lock.newCondition();
    // 消费者条件变量
    private Condition emptyWaitSet = lock.newCondition();
    // 容量
    private int capacity;

    /**
     * 阻塞获取
     *
     * @return
     */
    public T take() {

    }

    /**
     * 阻塞添加
     *
     * @return
     */
    public void pull() {

    }

    /**
     * 获取任务数量
     *
     * @return
     */
    public int size() {

    }
}

阻塞队列方法编写

take方法要从队列中获取元素，需要判断队列是否为空，如果为空则需要等待， 如果不为空则拿取队列头部元素（队列先入先出特性），再唤醒在等待放入任务的生产者线程。

pull方法要为队列中添加元素，需要判断队列是否已满，如果已满则需要等待，如果队列没有满，则在尾部添加元素，同时唤醒等待拿去任务的消费者线程。

size方法则获取队列里的元素多少即可。

    /**
     * 阻塞获取
     *
     * @return
     */
    public T take() {
        lock.unlock();
        try {
            // 获取之前,判断队列是否为空,如果为空则需要等待
            while (queue.isEmpty()) {
                try {
                    fullWaitSet.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            // 如果不为空 则拿取队列头部元素 因为队列特性 拿取头部元素 也是移除头部元素 因此方法是removeFirst()
            T t = queue.removeFirst();
            // 唤醒等待放入队列的生产者线程
            fullWaitSet.signal();
            return t;
        } finally {
            lock.unlock();
        }
    }

    /**
     * 阻塞添加
     *
     * @return
     */
    public void pull(T element) {
        lock.lock();
        try {
            // 当阻塞队列容量已满,则进入等待
            while (queue.size() == capacity) {
                try {
                    fullWaitSet.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            // 当阻塞队列没有满时,则可以往里添加元素(添加至尾部)
            queue.addLast(element);
            // 唤醒等待的消费者线程
            emptyWaitSet.signal();
        } finally {
            lock.unlock();
        }
    }

    /**
     * 获取任务数量
     *
     * @return
     */
    public int size() {
        lock.lock();
        try {
            return queue.size();
        } finally {
            lock.unlock();
        }
    }

poll增强（待更新）

自定义线程池的实现（待更新）

任务提交 和 Worker的实现（待更新）

take死等和 poll超时（待更新）

当队伍线程已满（待更新）

offer增强（待更新）

拒绝策略（待更新）

一、说明

ThreadPoolExecutor

• Java提供的线程池Executor框架相关的工具类中，最核心的是ThreadPoolExecutor
• 它有多个构造方法来实现自定义创建线程池，以内部线程池的形式对外提供管理任务执行，线程调度，线程池管理等

二、理解

ThreadPoolExecutor
java.util.cocurrent 包下ThreadPoolExecutor类继承AbstractExecutorService

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

• corePoolSize 核心线程数，默认为0，有任务时将创建线程去执行，当线程数达到corePoolSize时，停止线程创建，将任务放到队列中等待；调用prestartAllCoreThreads()或prestartCoreThread()方法，可以预创建线程
• maximumPoolSize 线程池线程数，当线程数达到核corePoolSize时，如果任务队列已满，线程池会创建新的线程，直到线程数量达到最maxPoolSize
• keepAliveTime 线程池中超过corePoolSize的空闲线程最大存活时间，超时则销毁，直到线程数量等于corePoolSize
• TimeUnit keepAliveTime时间单位
• workQueue 阻塞任务队列，用来存储等待执行的任务
• threadFactory 线程工厂，可以更改线程的名称，线程组，优先级，守护进程状态等
• RejectedExecutionHandler 拒绝执行策略，当提交任务数超过maximumPoolSize+workQueue时，再提交任务的会交给拒绝策略去处理

workQueue

• 当线程池任务线程数量 < corePoolSize，执行器会创建一个新的线程来执行新添加的任务
• 当线程池任务线程数量 > corePoolSize，且workQueue未满时，执行器将新添加的任务放到workQueue中
• 当线程池任务线程数量 > corePoolSize，且workQueue已满时，行器会创建一个新的线程来执行新添加的任务，直到超过maximumPoolSize执行拒绝策略

队列策略

• Direct handoffs 直接握手队列，使用SynchronousQueue队列，提交的任务会马上执行，不会被保存，如果执行任务线程数小于maximumPoolSize，则尝试创建新的进程，如果达到maximumPoolSize，则执行拒绝策略

• Bounded queues 有界队列，一般使用ArrayBlockingQueue制定队列的最大长度，当创建线程数达到corePoolSize时，若有新任务加入，则直接进入任务队列等待，若等待队列已满，即超过ArrayBlockingQueue初始化的容量，则继续创建线程，直到线程数达到maximumPoolSize，则执行拒绝策略

• Unbounded queues 无界队列，一般使用无预定长度的LinkedBlockingQueue，当线程数达到corePoolSize后，若有新任务加入，则直接进入任务队列等待，任务队列可以无限添加新的任务，maximumPoolSize参数是无效的

RejectedExecutionHandler

• 当线程池已经被关闭，或者任务数超过maximumPoolSize+workQueue时执行拒绝策略
• ThreadPoolExecutor.AbortPolicy 默认拒绝策略，丢弃任务并抛出RejectedExecutionException异常
• ThreadPoolExecutor.DiscardPolicy 直接丢弃任务，但不抛出异常
• ThreadPoolExecutor.DiscardOldestPolicy 丢弃任务队列最先加入的任务，再执行execute方法把新任务加入队列执行
• ThreadPoolExecutor.CallerRunsPolicy 会调用当前线程池的所在的线程去执行被拒绝的任务

线程池任务的执行流程

三、实现

1.SynchronousQueue

创建 ThreadPoolExecutorTest类，默认使用ThreadPoolExecutor.AbortPolicy拒绝策略，队列是SynchronousQueue，超出核心线程的任务会创建新的线程来执行，设置核心线程数最大值为4，线程池线程数最大值为8，最大等待时间为5秒

public class ThreadPoolExecutorTest {
    public static void main(String[] args) throws InterruptedException {
        // 1.创建自定义线程池
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(4, 8, 5, 
        TimeUnit.SECONDS, 
        new SynchronousQueue<>(),
        Executors.defaultThreadFactory(),
        new ThreadPoolExecutor.AbortPolicy());
        // 2.创建线程任务
        Runnable runnable = new Runnable() {
            @Override
            public void run() {
                try {
                    Thread.sleep(2000);
                    System.out.println(Thread.currentThread().getName() + " run");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }

            }
        };
        // 3.执行任务
        System.out.println("设置核心线程数最大值为4，线程池线程数最大值为8");
        System.out.println("&&&&开始执行线程&&&&");

        System.out.println("----执行4个任务----");
        threadPoolExecutor.execute(runnable);
        threadPoolExecutor.execute(runnable);
        threadPoolExecutor.execute(runnable);
        threadPoolExecutor.execute(runnable);
        System.out.println("当前核心线程数" + threadPoolExecutor.getCorePoolSize());
        System.out.println("当前线程池线程数" + threadPoolExecutor.getPoolSize());
        System.out.println("当前队列任务数" + threadPoolExecutor.getQueue().size());

        System.out.println("----再执行4个任务----");
        threadPoolExecutor.execute(runnable);
        threadPoolExecutor.execute(runnable);
        threadPoolExecutor.execute(runnable);
        threadPoolExecutor.execute(runnable);
        System.out.println("当前核心线程数" + threadPoolExecutor.getCorePoolSize());
        System.out.println("当前线程池线程数" + threadPoolExecutor.getPoolSize());
        System.out.println("当前队列任务数" + threadPoolExecutor.getQueue().size());

        Thread.sleep(10000);
        System.out.println("----休眠10秒后----");
        System.out.println("当前核心线程数" + threadPoolExecutor.getCorePoolSize());
        System.out.println("当前线程池线程数" + threadPoolExecutor.getPoolSize());
        System.out.println("当前队列任务数" + threadPoolExecutor.getQueue().size());
        // 4.关闭线程池
        threadPoolExecutor.shutdown();
    }
}

一共有4个核心，超过核心线程将创建非核心线程，核心线程默认情况下不会被回收，不受时间限制，而超时的非核心线程将被回收

但如果再执行4个任务，线程数超过maximumPoolSize，再提交任务将被丢弃并抛出RejectedExecutionException异常

2.ArrayBlockingQueue

当线程数达到corePoolSize后，若有新任务加入，则直接进入任务队列等待，超出队列的任务会创建新的线程来执行

        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(4, 8, 5, 
        TimeUnit.SECONDS, 
        new ArrayBlockingQueue<>(4),
        Executors.defaultThreadFactory(),
        new ThreadPoolExecutor.AbortPolicy());

一共有4个核心，当线程数超过corePoolSize+workQueue时，将创建非核心线程，核心线程默认情况下不会被回收，不受时间限制，而超时的非核心线程将被回收

但如果再执行4个任务，线程数超过maximumPoolSize+workQueue，再提交任务将被丢弃并抛出RejectedExecutionException异常

3.LinkedBlockingQueue

当线程数达到corePoolSize后，若有新任务加入，则直接进入任务队列等待，任务队列可以无限添加新的任务，maximumPoolSize参数是无效的

        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(4, 8, 5, 
        TimeUnit.SECONDS, 
        new LinkedBlockingDeque<>(), 
        Executors.defaultThreadFactory(),
        new ThreadPoolExecutor.AbortPolicy());    

一共有4个核心，超过核心线程将创建非核心线程，核心线程默认情况下不会被回收，不受时间限制，而超时的非核心线程将被回收

但当设置LinkedBlockingDeque容量限制为4时，当线程数超过corePoolSize+workQueue时，将创建非核心线程，核心线程默认情况下不会被回收，不受时间限制，而超时的非核心线程将被回收

        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(4, 8, 5, 
        TimeUnit.SECONDS, 
        new LinkedBlockingDeque<>(4), 
        Executors.defaultThreadFactory(),
        new ThreadPoolExecutor.AbortPolicy());

但如果再执行4个任务，线程数超过maximumPoolSize+workQueue，再提交任务将被丢弃并抛出RejectedExecutionException异常

面试题：一个线程池，core 7；max 20；queue：50；100 个并发进来如何分配：

# 7个会立即执行，50个会进入队列，再开13个进行执行，剩余30个使用拒绝策略

线程池：3 大方法，7 大参数，4 种拒绝策略

池化技术

程序的运行，本质：占用系统的资源！优化资源的使用！

线程池、连接池、内存池、对象池。。。创建、销毁 => 十分浪费资源

池化技术：事先准备好一些资源，有人要用，就来拿，用完之后归还

线程池的好处

1、降低资源的损耗

通过重复利用已经创建好的线程降低线程的创建和销毁带来的损耗

2、提高响应的速度

当线程池中的线程数没有超过线程池的最大上限时，有的线程处于等待分配任务的状态，当任务来时无需创建新的线程就能执行

3、方便管理

线程池会根据当前系统特点，对池内的线程进行优化处理，减少创建和销毁线程带来的系统开销。无限的创建和销毁线程不仅消耗系统资源，还降低系统的稳定性，使用线程池可以统一分配

线程复用，可以控制最大并发数，管理线程

线程池：3 大方法

package com.xoste.pool;

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

/**
 * @author 小斯特
 * @date 2021/11/23 16:27
 * Executors 工具类、3大方法
 */
public class ExecutorsDemo {
    public static void main(String[] args) {
        // 单个线程
        ExecutorService threadPool = Executors.newSingleThreadExecutor();
        // 创建一个固定的线程池大小
        ExecutorService threadPool2 = Executors.newFixedThreadPool(5);
        // 可伸缩的，遇强则强，遇弱则弱
        ExecutorService threadPool3 = Executors.newCachedThreadPool();
        try {
           for (int i = 0; i < 100; i++) {
               threadPool3.execute(() -> {
                   System.out.println(Thread.currentThread().getName() + " ok");
               });
           }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            threadPool3.shutdown();
        }
    }
}

线程池：7大参数

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-CeaP33zo-1646724702419)(JUC.assets\image-20211123162036611-1637673805309-1637673810285.png)]

// 单线程的线程池，后台从队列里面获取任务，挨个执行
public static ExecutorService newSingleThreadExecutor() {
    return new FinalizableDelegatedExecutorService
        (new ThreadPoolExecutor(1, 1,
                                0L, TimeUnit.MILLISECONDS,
                                new LinkedBlockingQueue<Runnable>()));
}
// 固定大小，core=max，都不可回收
public static ExecutorService newFixedThreadPool(int nThreads) {
        return new ThreadPoolExecutor(nThreads, nThreads,
                                      0L, TimeUnit.MILLISECONDS,
                                      new LinkedBlockingQueue<Runnable>());
    }
// core=0，所有都可回收
public static ExecutorService newCachedThreadPool() {
        return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                      60L, TimeUnit.SECONDS,
                                      new SynchronousQueue<Runnable>());
    }
// 定时任务的线程池
 public static ScheduledExecutorService newScheduledThreadPool(
            int corePoolSize, ThreadFactory threadFactory) {
        return new ScheduledThreadPoolExecutor(corePoolSize, threadFactory);
    }

// 本质：ThreadPoolExecutor()
public ThreadPoolExecutor(int corePoolSize, // 核心线程池大小
                              int maximumPoolSize, // 最大核心线程池大小
                              long keepAliveTime, // 超时了没有人就会释放
                              TimeUnit unit, // 超时单位
                              BlockingQueue<Runnable> workQueue, // 阻塞队列
                              ThreadFactory threadFactory, // 线程工厂，创建线程，一般不用动
                              RejectedExecutionHandler handler // 拒绝策略) {
        if (corePoolSize < 0 ||
            maximumPoolSize <= 0 ||
            maximumPoolSize < corePoolSize ||
            keepAliveTime < 0)
            throw new IllegalArgumentException();
        if (workQueue == null || threadFactory == null || handler == null)
            throw new NullPointerException();
        this.corePoolSize = corePoolSize;
        this.maximumPoolSize = maximumPoolSize;
        this.workQueue = workQueue;
        this.keepAliveTime = unit.toNanos(keepAliveTime);
        this.threadFactory = threadFactory;
        this.handler = handler;
    }`

4 种拒绝策略

默认拒绝策略

手动创建一个线程池

package com.xoste.pool;

import java.util.concurrent.*;

/**
 * @author 小斯特
 * @date 2021/11/23 16:27
 * Executors 工具类、3大方法
 *
 *  // 银行满人了，还有人进来，不处理这个人，抛异常
 *  new ThreadPoolExecutor.AbortPolicy()
 *  //既不抛弃任务也不抛出异常，而是将某些任务回退到调用者，让调用者去执行它（哪来的，去哪里）
 *  new ThreadPoolExecutor.CallerRunsPolicy()
 *  // 队列满了，不会抛出异常！
 *  new ThreadPoolExecutor.DiscardPolicy()
 *  // 队列满了，直接丢弃最老的任务，重新尝试执行当前任务
 *  new ThreadPoolExecutor.DiscardOldestPolicy()
 *
 */
public class ExecutorsDemo {
    public static void main(String[] args) {
        // 自定义线程池！工作 ThreadPoolExecutor
        ThreadPoolExecutor threadPool = new ThreadPoolExecutor(
                2,// 核心线程池大小
                5,// 最大核心线程池大小
                3,// 超时了没有人就会释放（存活时间）
                TimeUnit.SECONDS,// 超时单位
                new LinkedBlockingQueue<>(3),// 阻塞队列容器
                Executors.defaultThreadFactory(),// 线程工厂，创建线程，一般不用动
                // 队列满了，直接丢弃最老的任务，重新尝试执行当前任务
                new ThreadPoolExecutor.DiscardOldestPolicy() // 拒绝策略);
        try {
           for (int i = 0; i < 9; i++) {
               threadPool.execute(() -> {
                   // 使用了线程池之后，使用线程池来创建线程
                   System.out.println(Thread.currentThread().getName() + " ok");
               });
           }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
             // 线程池用完，程序结束，关闭线程池
            threadPool.shutdown();
        }
    }
}

小结–拓展

池的最大的大小如何去设置！

了解：CPU密集型，IO密集型（调优）

// 最大线程到底该如何定义
// 1.CPU 密集型，几核，就是几，可以保持 CPU 的效率最高！
// 2.IO 密集型 > 判断程序中，十分耗 IO 的线程，
// 程序 15个大型任务 IO十分占用资源

// 获取  CPU 的核数
System.out.println(Runtime.getRuntime().availableProcessors());