Java高并发编程详解：多线程与系统设计笔记 - 第8章线程池原理以及自定义线程池 - 《Java》

1.线程池原理
2.线程池实现
- 2.1线程池接口定义
- 2.2线程池详细实现
  - 1.LinkedRunnableQueue
  - 2.初始化线程池
3.线程池应用

1.线程池原理

所谓线程池，通俗的理解就是有一个池子，里面存放着已经创建好的线程，当有任务提交给线程池执行时，池子中的某个线程会主动执行该任务。如果池子中的线程数量不够应付数量众多的任务时，则需要自动扩充新的线程到池子中，但是该数量是有限的，就好比池塘的水界线一样。当任务比较少的时候，池子中的线程能够自动回收，释放资源。为了能够异步地提交任务和缓存未被处理的任务，需要有一个任务队列，如图所示。

通过上面的描述可知，一个完整的线程池应该具备如下要素。

任务队列：用于缓存提交的任务。
线程数量管理功能：一个线程池必须能够很好地管理和控制线程数量，可通过如下三个参数来实现，比如创建线程池时初始的线程数量in it；线程池自动扩充时最大的线程数量max；在线程池空闲时需要释放线程但是也要维护一定数量的活跃数量或者核心数量core。有了这三个参数，就能够很好地控制线程池中的线程数量，将其维护在一个合理的范围之内，三者之间的关系是in it<=core<=max
任务拒绝策略：如果线程数量已达到上限且任务队列已满，则需要有相应的拒绝策略来通知任务提交者。
线程工厂：主要用于个性化定制线程，比如将线程设置为守护线程以及设置线程名称等。
Queue Size：任务队列主要存放提交的Runnable，但是为了防止内存溢出，需要有limit数量对其进行控制。

Keepedalive时间：该时间主要决定线程各个重要参数自动维护的时间间隔

2.线程池实现

线程池实现类图

2.1线程池接口定义

1.ThreadPool

ThreadPool主要定义了一个线程池应该具备的基本操作和方法，下面是ThreadPool接口定义的方法:

public interface ThreadPool {
// 提交任务到线程池
void execute(Runnable runnable);
// 关闭线程池
void shutdown();
// 获取线程池的初始化大小
int getInitSize();
// 获取线程池最大的线程数
int getMaxSize();
// 获取线程池的核心线程数量
int getCoreSize();
// 获取线程池中用于缓存任务队列的大小
int getQueueSize();
// 获取线程池中活跃线程的数量
int getActiveCount();
// 查看线程池是否已经被shutdown
boolean isShutdown();
}

2.RunnableQueue

RunanbleQueue主要用于存放提交的Runnable，该Runnable是一个BlockedQueue，并且有limit的限制，示例代码所示。

// 任务队列，主要用于缓存提交到线程池中的任务
public interface RunnableQueue {
// 当前新的任务进来时首先会offer到队列中
void offer(Runnable runnable);
// 工作线程通过take方法获取Runnable
Runnable take();
// 获取任务队列中任务的数量
int size();
}

3.ThreadFactory

ThreadFactory提供了创建线程的接口，以便于个性化地定制Thread，比如Thread应该被加到哪个Group中，优先级、线程名字以及是否为守护线程等，示例代码所示。

// 创建线程的工厂
@FunctionalInterface
public interface ThreadFactory {
// 用于创建线程
Thread createThread(Runnable runnable);
}

4.DenyPolicy

DenyPolicy主要用于当Queue中的runnable达到了limit上限时，决定采用何种策略通知提交者。该接口中定义了三种默认的实现，具体如代码所示。

@FunctionalInterface
public interface DenyPolicy {
    // 拒绝方法
    void reject(Runnable runnable, ThreadPool threadPool);
    // 该拒绝策略会直接将任务丢弃
    class DiscardDenyPolicy implements  DenyPolicy {
        @Override
        public void reject(Runnable runnable, ThreadPool threadPool) {
            // do nothing
        }
    }
    // 该拒绝策略会向任务提交者抛出异常
    class AbortDenyPolicy implements DenyPolicy {
        @Override
        public void reject(Runnable runnable, ThreadPool threadPool) {
//            throw new Run
        }
    }
    // 该拒绝策略会使任务提交者所在的线程中执行任务
    class RunnerDenyPolicy implements DenyPolicy {
        @Override
        public void reject(Runnable runnable, ThreadPool threadPool) {
            if( !threadPool.isShutdown()) {
                runnable.run();
            }
        }
    }
}

5.RunnableDenyException

RunnableDenyException是RuntimeException的子类，主要用于通知任务提交者，任务队列已无法再接收新的任务。示例代码如所示。

public class RunnableDenyException extends RuntimeException {
    public RunnableDenyException(String message) {
       super(message);
    }
}

6.InternalTask

InternalTask是Runnable的一个实现，主要用于线程池内部，该类会使用到RunnableQueue，然后不断地从queue中取出某个runnable，并运行runnable的run方法，示例代码如代码所示。

public class InternalTask  implements Runnable{
    private final RunnableQueue runnableQueue;
    private volatile  boolean running = true;
    public InternalTask(RunnableQueue runnableQueue) {
        this.runnableQueue = runnableQueue;
    }
    @Override
    public void run() {
        // 如果当前任务为running并且没有被中断，则其将不断地从queue中获取runnable，然后执行run方法
        while( running && !Thread.currentThread().isInterrupted() ) {
            try {
                Runnable task = runnableQueue.take();
                task.run();
            } catch (Exception e) {
                running = false;
                break;
            }
        }
    }
    public void stop() {
        this.running = false;
    }
}

2.2线程池详细实现

本节将对线程池进行详细的实现，其中会涉及很多同步的技巧和资源竞争，我们将结合本书第一部分的大多数知识灵活使用，也算是一个综合练习。

1.LinkedRunnableQueue

LinkedRunnableQueue代码实现方式：


import java.util.LinkedList;
public class LinkedRunnableQueue implements RunnableQueue{
    // 任务队列的最大容量，在构造时传入
    private final int limit;
    // 若任务队列中的任务已经满了，则需要执行拒绝策略
    private final DenyPolicy denyPolicy;
    // 存放任务的队列
    private final LinkedList<Runnable> runnableLinkedList = new LinkedList<>();
    private final ThreadPool threadPool;
    public LinkedRunnableQueue(int limit, DenyPolicy denyPolicy, ThreadPool threadPool) {
        this.limit = limit;
        this.denyPolicy = denyPolicy;
        this.threadPool = threadPool;
    }
    /*
        在LinkedRunnableQueue中有几个重要的属性，第一个是limit，也就是Runnable队
        列的上限；当提交的Runnable数量达到limit上限时，则会调用DenyPolicy的reject方法；
        runnableList是一个双向循环列表，用于存放Runnable任务，示例代码如下：
     */
    /*
    offer方法是一个同步方法，如果队列数量达到了上限，则会执行拒绝策略，否则会将runnable存放至队列中，同时唤醒take任务的线程：
     */
    @Override
    public void offer(Runnable runnable) {
        synchronized (runnableLinkedList) {
            if ( runnableLinkedList.size() >= limit ) {
                // 无法容纳新的任务时执行拒绝策略
                denyPolicy.reject(runnable, threadPool);
            } else {
                // 将任务加入到队尾，并且唤醒阻塞中的线程
                runnableLinkedList.addLast(runnable);
                runnableLinkedList.notifyAll();
            }
        }
    }
    /*
    take方法也是同步方法，线程不断从队列中获取Runnable任务，当队列为空的时候工
    作线程会陷入阻塞，有可能在阻塞的过程中被中断，为了传递中断信号需要在catch语句块
    中将异常抛出以通知上游(Internal Task)，示例代码如下：
     */
    @Override
    public Runnable take() throws InterruptedException{
        synchronized (runnableLinkedList) {
            while(runnableLinkedList.isEmpty()) {
                try {
                    // 如果任务队列中没有可执行任务，则当前线程将会挂起，
                    // 进入runnableList关联的monitor waitset中等待唤醒(新的任务加入）
                    runnableLinkedList.wait();
                } catch (InterruptedException e) {
                    // 被中断时需要将异常抛出
                    throw e;
                }
            }
            return runnableLinkedList.removeFirst();
        }
    }
    // size方法用于返回runnable List的任务个数。
    @Override
    public int size() {
        synchronized (runnableLinkedList) {
            // 返回当前任务队列中的任务数
            return runnableLinkedList.size();
        }
    }
}

2.初始化线程池

根据前面的讲解，线程池需要有数量控制属性、创建线程工厂、任务队列策略等功能，线程池初始化代码如所示


import java.util.ArrayDeque;
import java.util.Queue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
public class BasicThreadPool extends Thread implements ThreadPool {
    // 初始化线程数量
    private final int initSize;
    // 线程池最大线程数量
    private final int maxSize;
    // 线程池核心线程数量
    private final int coreSize;
    // 当前活跃的线程数量
    private int activeCount;
    // 创建线程所需的工厂
    private final ThreadFactory threadFactory;
    // 任务队列
    private  final RunnableQueue runnableQueue;
    // 线程池是否已经被shutdown
    private volatile boolean isShutdown = false;
    // 工作线程队列
    private final Queue<ThreadTask> threadQueue = new ArrayDeque<>();
    private final static DenyPolicy DEFAULT_DENY_POLICY = new DenyPolicy.DiscardDenyPolicy();
    private final static ThreadFactory DEFAULT_THREAD_FACTORY=new DefaultThreadFactory();
    private final long keepAliveTime;
    private final TimeUnit timeUnit;
    // 构造时需要传递的参数：初始化的线程数量，最大的线程数量，核心线程数量，任务队列的最大数量
    public BasicThreadPool(int initSize, int maxSize, int coreSize, int queueSize) {
        this(initSize, maxSize, coreSize, DEFAULT_THREAD_FACTORY, queueSize, DEFAULT_DENY_POLICY,10,TimeUnit.SECONDS);
    }
    public BasicThreadPool(int initSize, int maxSize, int coreSize, ThreadFactory threadFactory,
                           int queueSize, DenyPolicy denyPolicy, long keepAliveTime,
                           TimeUnit timeUnit) {
        this.initSize = initSize;
        this.maxSize = maxSize;
        this.coreSize = coreSize;
        this.threadFactory = threadFactory;
        this.runnableQueue = new LinkedRunnableQueue(queueSize, denyPolicy, this);
        this.keepAliveTime = keepAliveTime;
        this.timeUnit = timeUnit;
        this.init();
    }
    // 初始化时，先创建initSize个线程
    private void init() {
        start();
        for(int i = 0; i < initSize; i++) {
            newThread();
        }
    }
    /*
        一个线程池除了控制参数之外，最主要的是应该有活动线程，其中Queue<Thread-Task>主要用来存放活动线程，
        BasicThreadPool同时也是Thread的子类，它在初始化的时候启动， 在keepalive时间间隔到了之后再自动维护
        活动线程数量(采用继承Thread的方式其实不是一种好的方法，因为Basic ThreadPool会暴露Thread的方法，
        建议将继承关系更改为组合关系，读者可以自行修改)。
     */
    /*
    提交任务非常简单，只是将Runnable插入runnableQueue中即可。
     */
    @Override
    public void execute(Runnable runnable) {
        if( this.isShutdown ) {
            throw new IllegalStateException("The thread pool is destory");
        }
        // 提交任务只是简单地往任务队列中插入Runnable
        this.runnableQueue.offer(runnable);
    }
    // 4.线程池自动维护
    /*
    线程池中线程数量的维护主要由run负责， 这也是为什么BasicThreadPool继承自Thread了，
     不过笔者不推荐使用直接继承的方式， 线程池自动维护代码如下：
     */
    private void newThread() {
        // 创建任务线程，并且启动
        InternalTask internalTask = new InternalTask(runnableQueue);
        Thread thread = this.threadFactory.createThread(internalTask);
        ThreadTask threadTask = new ThreadTask(thread, internalTask);
        threadQueue.offer(threadTask);
        this.activeCount++;
        this.start();
    }
    private void removeThread(){
        //从线程池中移除某个线程
        ThreadTask threadTask = threadQueue.remove();
        threadTask.internalTask.stop();
        this.activeCount--;
    }
    @Override
    public void run() {
        // run方法继承自Thread主要用于维护线程数量，比如扩容，回收等工作
        while(!isShutdown && !isInterrupted() ) {
            try {
                timeUnit.sleep(keepAliveTime);
            } catch (InterruptedException e) {
                isShutdown = true;
                break;
            }
            synchronized (this) {
                if ( isShutdown) break;
                // 当前的队列中有任务尚未处理，并且activeCount<coreSize 则继续扩容
                if ( runnableQueue.size() > 0 && activeCount < coreSize ) {
                    for(int i = initSize; i < coreSize; i++) {
                        newThread();
                    }
                    // continue的目的在于不想让线程的扩容直接达到maxSize
                    continue;
                }
                // 当前的队列中有任务尚未处理，并且activeCount < maxSize则继续扩容
                if ( runnableQueue.size() > 0 && activeCount < maxSize ) {
                    for( int i = coreSize; i < maxSize; i++) {
                        newThread();
                    }
                }
                // 如果任务队列中没有任务，则需要回收，回收至coreSize即可
                if ( runnableQueue.size() == 0 && activeCount > coreSize) {
                    for( int i = coreSize; i < activeCount; i++) {
                        removeThread();
                    }
                }
            }
        }
    }
    /*
    5. 线程池销毁
    线程池的销毁同样需要同步机制的保护，
    主要是为了防止与线程池本身的维护线程引起数据冲突，线程池销毁代码如下：
     */
    /*
    销毁线程池主要为了是停止BasicThreadPool线程， 停止线程池中的活动线程并且将
    isShutdown开关变量更改为true
     */
    /*
    1.下面重点来解说线程自动维护的方法，自动维护线程的代码块是同步代码块，主要是为了阻止在线程维护过程中线程池销毁引起的数据不一致问题。
    2.任务队列中若存在积压任务， 并且当前活动线程少于核心线程数， 则新建coreSize initSize数量的线程，并且将其加入到活动线程队列中
    为了防止马上进行coreSize数量的扩充，建议使用continue终止本次循环。
    3.任务队列中有积压任务，并且当前活动线程少于最大线程数，则新建maxSize-coreSize数量的线程，并且将其加入到活动队列中。
    4.当前线程池不够繁忙时，则需要回收部分线程，回收到coreSize数量即可，回收时调用removeThread()方法，
    在该方法中需要考虑的一点是，如果被回收的线程恰巧从Runnable任务取出了某个任务，
    则会继续保持该线程的运行， 直到完成了任务的运行为止，详见Internal Task的run方法。
     */
    @Override
    public void shutdown() {
        synchronized (this) {
            if ( isShutdown ) return;
            isShutdown = true;
            threadQueue.forEach(threadTask -> {
                threadTask.internalTask.stop();
                threadTask.thread.interrupt();
            });
            this.interrupt();
        }
    }
    @Override
    public int getInitSize() {
        if( isShutdown)
            throw new IllegalStateException("The thread pool is destory");
        return this.initSize;
    }
    @Override
    public int getMaxSize() {
        if ( isShutdown )
            throw new IllegalStateException("The thread pool is destory");
        return this.maxSize;
    }
    @Override
    public int getCoreSize() {
        if ( isShutdown )
            throw new IllegalStateException("The thread pool is destory");
        return this.coreSize;
    }
    @Override
    public int getQueueSize() {
        if ( isShutdown )
            throw new IllegalStateException("The thread pool is destory");
        return runnableQueue.size();
    }
    @Override
    public int getActiveCount() {
        synchronized (this) {
            return this.activeCount;
        }
    }
    @Override
    public boolean isShutdown() {
        return this.isShutdown;
    }
    private static class DefaultThreadFactory implements  ThreadFactory {
        private static final AtomicInteger GROUP_COUNTER = new AtomicInteger(1);
        private static final ThreadGroup group = new ThreadGroup("MyThreadPool-"
        + GROUP_COUNTER.getAndDecrement());
        private static final AtomicInteger COUNTER = new AtomicInteger(0);
        @Override
        public Thread createThread(Runnable runnable) {
            return new Thread(group, runnable, "thread-pool-" + COUNTER.getAndDecrement());
        }
    }
    private static class ThreadTask {
        Thread thread;
        InternalTask internalTask;
        public ThreadTask(Thread thread, InternalTask internalTask) {
            this.thread = thread;
            this.internalTask = internalTask;
        }
    }
}

3.线程池应用

本节将写一个简单的程序分别测试线程池的任务提交、线程池线程数量的动态扩展，以及线程池的销毁功能，代码如所示。


import java.util.concurrent.TimeUnit;
public class ThreadPoolTest {
    public static void main(String[] args) throws InterruptedException {
        // 定义线程池，初始化线程数量2， 核心线程数为4， 最大线程数为6， 任务队列最多允许1000个任务
        final ThreadPool threadPool = new BasicThreadPool(2,6,4,1000);
        // 定义20个任务并且提交给线程池
        for(int i = 0; i < 20; i++ ) {
            threadPool.execute(
                    ()-> {
                        try {
                            TimeUnit.SECONDS.sleep(10);
                            System.out.println(Thread.currentThread().getName() + " is running and done.");
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
            );
        }
        // 用于获取一些线程池信息
//        for(;;) {
//            System.out.println("getActiveCount:" + threadPool.getActiveCount());
//            System.out.println("getQueueSize:" + threadPool.getQueueSize());
//            System.out.println("getCoreSize:" + threadPool.getCoreSize());
//            System.out.println("getMaxSize:" + threadPool.getMaxSize());
//            TimeUnit.SECONDS.sleep(5);
//        }
        Thread.currentThread().join();
    }
}

程序运行12秒之后，线程池将被销毁，线程池中的线程都将被销毁，同样为了验证所有线程是否被成功销毁，也可以借助JVM工具查看堆栈信息，这也是在最后使用current Thread join进行阻塞以便于查看的原因。

第8章 线程池原理以及自定义线程池