线程池原理及使用

线程池继承关系

1.为什么使用线程池？

1.反复创建线程开销大;

2.过多线程会占用太多内存(执行任务易出现“内存溢出”);

3.加快程序响应速度;

4.合理利用CPU和内存;

5.统一管理线程;

2.创建和停止线程池

2.1.线程池参数解释

1.keppAliveTime

如果线程池当中的线程数量大于“corePoolSize”，当这些多余的线程空闲时间超过keepAliveTime时它们就会被回收;

2.threadFactory

新的线程是由threadFactory创建的默认使用Executors.defaultThreadFactory()，创建出来的线程都是用户线程且属于同一个线程组优先级都默认为5，如果自己指定threadFactory那么就可以改变线程名、线程组、优先级、是否是守护线程等信息;

3.workQueue

常见三种队列类型：

1.直接交换队列(SynchronousQueue)

即队列不存储任务，直接将任务发往线程池执行;

2.无界队列(LinkedBlockingQueue)

即队列存储任务数量没有上限(有内存溢出风险);

3.有界队列(ArrayBlockingQueue)

即队列中指定存储多少个任务;

2.2.线程池添加线程规则

1.如果线程数小于corePoolSize就创建一个新线程去运行新任务;

2.如果线程数大于等于corePoolSize但小于最大线程数(maxPoolSize)则将任务放入队列;

3.如果队列中任务已满，并且线程数小于maxPoolSize则创建一个新线程来执行任务(队列塞满才尝试去扩容);

4.如果队列已满且线程数大于maxPoolSize则拒绝该任务添加到线程池;

2.3.使用工具类创建线程池的四种方式

如下四种创建线程池各参数情况：

1.newFixedThreadPool

原理：

corePoolSize = maxPoolSize = 传入线程个数;

使用队列：LinkedBlockingQueue(无界队列容量无上限);

固定长度线程池,newFixedThreadPool由于使用的工作队列是LinkedBlockingQueue是没有容量上限的，所以当请求越来越多并且来不及处理时会出现请求堆积造成大量内存被占用出现“内存溢出”问题;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
/**
 * 演示使用newFixedThreadPool
 *      该线程池使用的是LinkedBlockingQueue(无界队列,即队列中可以存放任意多任务有“内存溢出”风险)
 */
public class FixedThreadPool {
    public static void main(String[] args) {
        // 创建固定数量线程池
        ExecutorService pool = Executors.newFixedThreadPool(4);
        // 假设需执行1000个任务,每个任务休眠200ms再打印出线程名
        for (int i = 1; i <= 1000 ; i++) {
            pool.execute(new Task());
        }
    }

    // 任务类
    static class Task implements Runnable{
        @Override
        public void run() {
            try {
                TimeUnit.MILLISECONDS.sleep(200);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("当前线程名：" + Thread.currentThread().getName());
        }
    }
}

2.newSingleThreadExecutor

原理：

corePoolSize = maxPoolSize = 1

使用队列：LinkedBlockingQueue(无界队列容量无上限)

单线程线程池(该线程池只有一个线程)其原理和newFixedThreadPool线程池相同，当请求堆积且来不及处理时易出现“内存溢出”问题;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
/**
 * 演示使用newSingleThreadExecutor(单线程线程池)
 * 该线程池使用的是LinkedBlockingQueue(无界队列,即队列中可以存放任意多任务有“内存溢出”风险)
 */
public class SingleThreadPool {
    public static void main(String[] args) {
        // 创建单线程线程池
        ExecutorService pool = Executors.newSingleThreadExecutor();
        // 假设需执行1000个任务,每个任务休眠200ms再打印出线程名
        for (int i = 1; i <= 1000 ; i++) {
            pool.execute(new SingleThreadPool.Task());
        }
    }

    // 任务类
    static class Task implements Runnable{
        @Override
        public void run() {
            try {
                TimeUnit.MILLISECONDS.sleep(200);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("当前线程名：" + Thread.currentThread().getName());
        }
    }
}

3.CachedThreadPool

可缓存线程池使用直接交换队列，队列中不存储任务当任务来了判断当前线程数是否小于Integer.MAX_VALUE，若小于则直接交给线程池去执行，CachedThreadPool具有回收多余线程的功能(默认情况下空闲60s的线程会被自动回收)，CachedThreadPool的弊端在于maxPoolSize的值为Integer.MAX_VALUE，这可能会创建非常多的线程可能会导致“内存溢出”问题;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
/**
 * 演示使用CachedThreadPool(可缓存线程池)
 *      该线程池使用的是SynchronousQueue(直接交换队列,即队列中不存储任务当任务来了判断当前线程数是否小于Integer.MAX_VALUE，若小于则直接交给线程池去执行)
 */
public class CachedThreadPool {
    public static void main(String[] args) {
        ExecutorService pool = Executors.newCachedThreadPool();
        for (int i = 1; i <= 1000 ; i++) {
            pool.execute(new Task());
        }
    }

    // 任务类
    static class Task implements Runnable{
        @Override
        public void run() {
            try {
                TimeUnit.MILLISECONDS.sleep(200);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("当前线程名：" + Thread.currentThread().getName());
        }
    }
}

4.ScheduledThreadPool

支持定期及周期性任务执行的线程池(内部使用了延迟队列);

import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
/**
 * 演示使用ScheduledThreadPool(周期性执行线程池)
 */
public class ScheduledThreadPool {
    public static void main(String[] args) {
        ScheduledExecutorService pool = Executors.newScheduledThreadPool(10);
        // 用法一(延迟三秒后执行)
        pool.schedule(new Task(),3,TimeUnit.SECONDS);

        System.out.println("====================    分割线    ====================");

        // 用法二(周期性执行(首次执行延迟一秒后执行，后面每间隔3秒周期性执行))
        pool.scheduleAtFixedRate(new Task(),1,3,TimeUnit.SECONDS);
    }

    // 任务类
    static class Task implements Runnable{
        @Override
        public void run() {
            try {
                TimeUnit.MILLISECONDS.sleep(200);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("当前线程名：" + Thread.currentThread().getName());
        }
    }
}

2.4.如何设置线程池线程数量

1.CPU密集型：例如：加密，计算等高耗CPU的操作，最佳线程数为CPU核心数的1-2倍左右;

2.耗时IO型：例如：读写数据库，文件，网络读写操作，最佳线程数一般大于CPU核心数的很多倍，以JVM线程监控显示繁忙情况为依据确保线程空闲能够衔接上;

3.公式：CPU核心数 * (1 + 平均等待时间/平均工作时间);

2.5.停止线程池相关方法

1.shutdown

shutdown命令用于停止线程池，线程池调用shutdown命令并不是立即去停止线程池的，使用shutdown命令线程池会等到正在执行的任务和队列中的任务都执行完毕后才会停止，当线程池调用shutdown命令后对于新提交过来的任务线程池会抛出“RejectedExecutionException”异常去拒绝接收新任务;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
/**
 * 演示停止线程池方法 - shutdown命令
 */
public class shutDown {
    public static void main(String[] args) throws InterruptedException {

        ExecutorService pool = Executors.newFixedThreadPool(10);
        // 提交1000个任务
        for (int i = 1; i <= 1000 ; i++) {
            pool.execute(new Task());
        }

        // 休眠三秒让1000个任务全部已经提交给线程池
        TimeUnit.SECONDS.sleep(3);
        pool.shutdown();
        boolean shutdown = pool.isShutdown();
        System.out.println("isShutdown()用于判断当前线程池是否执行了shutdown或shutdownNow命令：" + shutdown);
        boolean terminated = pool.isTerminated();
        System.out.println("isTerminated()用于判断当前线程池是否真正停止(正在执行的任务 + 队列中任务)都执行完毕且已调用shutdown或shutdowNow命令： = " + terminated);
    }

    // 任务
    static class Task implements Runnable{
        @Override
        public void run() {
            try {
                TimeUnit.MILLISECONDS.sleep(500);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println(Thread.currentThread().getName());
        }
    }
}

2.shutdownNow

shutdownNow命令用于停止线程池，当线程池调用shutdownNow命令时会立即停止线程池执行shutdownNow命令会做两件事;

1.线程池会对正在执行的线程发出中断信号去中断当前正在执行的线程;

2.对于在队列中等待执行的任务shutdownNow命令返回值会返回在队列中的所有任务，一般执行shutdownNow命令时我们需要对返回值做处理(比如将这些队列中的任务存储到数据库中或使用其它线程池去执行队列中的任务);

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

public class ShutdownNow {
    public static void main(String[] args) throws InterruptedException {
        ExecutorService pool = Executors.newFixedThreadPool(10);

        // 提交100个任务
        for (int i = 1; i <= 100 ; i++) {
            pool.execute(new Task());
        }
        // 休眠2秒让100个任务全部添加到线程池中去
        TimeUnit.SECONDS.sleep(2);

        // 队列中剩余任务个数
        List<Runnable> taskQueue = pool.shutdownNow();
        System.out.println("队列中剩余任务个数： " + taskQueue.size());
    }

    // 任务
    static class Task implements Runnable{
        @Override
        public void run() {
            try {
                TimeUnit.MILLISECONDS.sleep(500);
            } catch (InterruptedException e) {
                System.out.println("线程池已执行shutdownNow命令，当前线程：" + Thread.currentThread().getName() + " 被中断了...");
            }
            System.out.println(Thread.currentThread().getName());
        }
    }
}

3.isShutdown

判断线程池是否已停止，当调用shutdown()或shutdownNow() 方法后则此方法返回true;

具体使用见本章shutdown命令演示代码;

4.isTerminated

判断线程池是否完全停止，当调用shutdown()或shutdownNow() 方法后且线程池中正在执行的任务及队列中任务都已经执行完毕则此方法返回true;

具体使用见本章shutdown命令演示代码;

5.awaitTermination

等待指定时间后检测当前线程池是否已关闭，一般情况下会和shutdown方法组合使用;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
/**
 * 演示：awaitTermination命令(等待指定时间后检测当前线程池是否已关闭)
 */
public class AwaitTermination {

    public static void main(String[] args) throws InterruptedException {
        ExecutorService pool = Executors.newFixedThreadPool(10);

        for (int i = 1; i <= 1000 ; i++) {
            pool.execute(new Task());
        }

        // 等待指定时间后检测当前线程池是否已关闭
        boolean flag = pool.awaitTermination(3, TimeUnit.SECONDS);
        System.err.println(flag);
    }

    // 任务
    static class Task implements Runnable{
        @Override
        public void run() {
            try {
                TimeUnit.MILLISECONDS.sleep(500);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println(Thread.currentThread().getName());
        }
    }
}

2.6.线程池拒绝策略

1.线程池何时拒绝新任务

1.1.线程池调用shutdown或shutdownNow关闭线程池方法后还有新任务提交过来，此时线程池会抛出“RejectedExecutionException”异常去拒绝新任务;

1.2.线程池最大线程数和工作队列到达上限时会拒绝新任务;

2.线程池的四种拒绝策略

1.AbortPolicy

当有新任务想要添加到线程池时直接抛出异常;

2.DiscardPolicy

当有新任务想要添加到线程池时不处理也不添加任务到工作队列(什么也不做);

3.DiscardOldestPolicy

当有新任务想要添加到线程池时抛弃存放在工作队列中最久的任务并添加新任务到队列;

4.CallerRunsPolicy

当有新任务想要添加到线程池时将新任务交给当前线程去执行(即主线程去执行)。这样做的好处在于不丢弃已有工作队列中的任务，同时给线程池处理已有工作队列中任务的时间，若这段时间线程池处理完一些任务后续再往线程池中添加新任务时就能成功添加);

2.7.线程池的“钩子”方法

通过线程池的钩子方法我们可以在每个线程执行的前后做一些事;

import java.util.concurrent.*;
/**
 * 演示每个线程池任务执行前后执行自定义的钩子函数
 *       可用于线程的日志搜集;
 */
public class ThreadPoolHookMethod  extends ThreadPoolExecutor {

    public ThreadPoolHookMethod(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
    }

    public ThreadPoolHookMethod(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory);
    }

    public ThreadPoolHookMethod(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, RejectedExecutionHandler handler) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, handler);
    }

    public ThreadPoolHookMethod(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler);
    }


    /**
     *  钩子方法一：在线程池执行之前做某些事
     * @param t 将要执行任务的线程
     * @param r 将要执行的任务
     */
    @Override
    protected void beforeExecute(Thread t, Runnable r) {
        super.beforeExecute(t, r);
        System.out.println("===============  我是线程池的钩子方法beforeExecute(在线程池执行任务前我执行了) 当前线程池名称：" + t.getName() + " ===============");
    }


    /**
     * 钩子方法二：在线程池执行之后做某些事
     * @param r 执行的任务
     * @param t 异常信息
     */
    @Override
    protected void afterExecute(Runnable r, Throwable t) {
        super.afterExecute(r, t);
        System.out.println("===============  我是线程池的钩子方法afterExecute(在线程池执行任务后我执行了)  ===============");
    }


    public static void main(String[] args) {
        ThreadPoolHookMethod pool = new ThreadPoolHookMethod(5,5,0L,TimeUnit.MILLISECONDS,new LinkedBlockingQueue<>());
        pool.execute(new Task());
    }

    // 任务
    static class Task implements Runnable{
        @Override
        public void run() {
            System.out.println("我是线程：" + Thread.currentThread().getName() +" 正在执行run方法中任务...");
        }
    }
}