02Java进阶--线程池详解

Posted 2021-06-24

线程池

线程池做的工作主要是控制运行的线程的数量，处理过程中将任务加入队列，然后在线程创建后，启动这些任务，如果线程超过了最大数量，超出的数量的线程排队等候，等其他线程执行完毕，再从队列中取出任务来执行。

主要特点为：线程的复用、控制最大并发数、管理线程

1、降低资源消耗：通过重复利用创建的线程，降低线程创建和消耗造成的资源消耗。

2、提高响应速度，当任务到达时，任务可以不需要等到线程创建就能立即执行。

3、提高线程的可管理性。线程是稀缺资源，如果无限的创建，不仅会消耗资源，还会降低系统的稳定性，使用线程池可以进行统一分配，调优和监控。

Java中的线程池是通过Executor框架实现的，该框架中用到了Executor、Executors、ExecutorService、ThreadPoolExecutor这几个类。

三种编码实现

线程池的实现方式有五种，java8新推出Executors.newWorkStealingPool(int)，使用目前机器上可用的处理器作为他的并行级别，除此之外，最常使用的是以下四种：

Executors.newFixedThreadPool(int)

创建一个定长线程池，可控制线程最大并发数，超出的线程会在队列中等待。

这种方式创建的线程池corePoolSize和maximumPoolSize值是相等的，它使用的是LinkedBlockingQueue。

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

示例代码如下：

public class Main {
    public static void main(String[] args) {
        ExecutorService fixedThreadPool = Executors.newFixedThreadPool(3);
        for (int i = 0; i < 10; i++){
            final int index = i;
            fixedThreadPool.execute(new Runnable() {
                @Override
                public void run() {
                    System.out.println(index);
                    try {
                        Thread.sleep(2000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            });
        }
    }
}

因为线程池大小为3，每个任务输出index后sleep 2秒，所以每两秒打印3个数字。

指定长线程池的大小最好根据系统资源进行设置。

Runtime.getRuntime().availableProcessors()

Executors.newSingleThreadExecutor()

创建一个单线程的线程池，只会用唯一的工作线程来执行任务，保证所有任务按照指定顺序执行。

这种方式将corePoolSize和maximumPoolSize都设置为1，使用的是LinkedBlockingQueue。

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

示例代码如下：

public class Main {
    public static void main(String[] args) {
        ExecutorService singleThreadExecutor = Executors.newSingleThreadExecutor();
        for (int i = 0; i < 10; i++){
            final int index = i;
            singleThreadExecutor.execute(new Runnable() {
                @Override
                public void run() {
                    System.out.println(index);
                    try {
                        Thread.sleep(2000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            });
        }
    }
}

结果依次输出，相当于顺序执行各个任务。

Executors.newCachedThreadPool()

创建一个可缓存线程池，如果线程池长度超过处理需要，可灵活回收空闲线程，若无可回收的，则新建线程。

这种方式将corePoolSize设置为0，将maximumPoolSize设置为Integer.MAX_VALUE，使用的SynchronousQueue。来了任务就创建线程运行，当线程空闲超过60s，就销毁线程。

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

示例代码如下：

public class Main {
    public static void main(String[] args) {
        ExecutorService newCachedThreadPool = Executors.newCachedThreadPool();
        for (int i = 0; i < 10; i++) {
            final int index = i;
            try {
                Thread.sleep(2000);
            }catch (InterruptedException e){
                e.printStackTrace();
            }
            newCachedThreadPool.execute(new Runnable() {
                @Override
                public void run() {
                    System.out.println(index);
                }
            });
        }
    }
}

线程池为无限大，当执行第二个任务时第一个任务已经完成，会复用执行第一个任务的线程，而不用每次新建线程。

Executors.newScheduledThreadPool

创建一个定长线程池，支持定时及周期性任务执行。

public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
    return new ScheduledThreadPoolExecutor(corePoolSize);
}

示例代码如下：

public class Main {
    public static void main(String[] args) {
        ScheduledExecutorService scheduledThreadPool = Executors.newScheduledThreadPool(5);
        // 表示延迟3秒执行
        scheduledThreadPool.schedule(new Runnable() {
            @Override
            public void run() {
                System.out.println("delay 3 seconds");
            }
        }, 3, TimeUnit.SECONDS);

        // 延迟1秒后每隔3秒执行一次
        scheduledThreadPool.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                System.out.println("delay 1 seconds, and executor every 3 seconds");
            }
        }, 1, 3, TimeUnit.SECONDS);
    }
}

schedule()方法用于延迟的定时任务，scheduleAtFixedRate()是周期性的定时任务。

七个重要参数

线程池包含如下七个重要的参数：

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

corePoolSize：线程池中常驻核心线程数。当线程池的线程数达到corePoolSize后，就会把到达的任务放到缓存队列当中。

maximumPoolSize：线程池能够容纳同时执行的最大线程数，必须大于等于1，

keepAliveTime：多余的空闲线程的存活时间，当前线程池数量超过corePoolSize时，档口空闲时间达到

keepAliveTime值时，多余空闲线程会被销毁，直到只剩下corePoolSize个线程为止。

unit：keepAliveTime的单位。

workQueue：任务队列，被提交但尚未被执行的任务。

threadFactory：表示生成线程池中工作线程的线程工厂，用于创建线程一般用默认的即可。

handler：拒绝策略，表示当队列满了，并且工作线程大于等于线程池的最大线程数（maximumPoolSize）时，如何来拒绝请求执行的runable的策略。

四种拒绝策略

当等待队列满时，且达到最大线程数，再有新任务到来，就需要启动拒绝策略。JDK提供了四种拒绝策略：

AbortPolicy(默认)：直接抛出RejectedExecutionException异常，阻止系统正常运行。

CallerRunsPolicy：不会抛弃任务，也不会抛出异常，而是将某些任务回退到调用者，来降低流量。

DiscardOldestPolicy：抛弃队列中等待最久的任务，然后把当前任务加入队列中尝试再次提交当前任务。

DiscardPolicy：直接丢弃任务，不予任何处理也不抛异常。

以上四种拒绝策略，均实现了RejectedExecutionHandler接口。

自定义线程池

在实际的开发过程中，一般都是设置七个参数来自定义线程池操作。

示例代码如下：

public class Main {
    public static void main(String[] args) {
        ExecutorService executorService = new ThreadPoolExecutor(3, 5, 1l,
                TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(3),
                Executors.defaultThreadFactory(),
                new ThreadPoolExecutor.DiscardPolicy());
        try {
            for (int i = 0; i < 10; i++) {
                executorService.execute(new Runnable() {
                    @Override
                    public void run() {
                        // 运行会保持8个任务，多余的会丢失，依靠拒绝策略来断定
                        System.out.println(Thread.currentThread().getName() + "\\t办理业务");
                    }
                });
            }

        }catch (Exception e){
            e.printStackTrace();
        }finally {
            executorService.shutdown();
        }
    }
}

合理配置参数

CPU密集型

CPU密集的意思是该任务需要大量的运算，而没有阻塞，CPU一直全速运行，这种类型的任务只有在真正多核CPU上才可能得到加速（通过多线程），而在单核CPU上，无论你开几个模拟的多线程该任务都不可能得到加速，因为CPU总的运算能力就那些。

CPU密集型任务配置尽可能少的线程数量，参考公式：

线程池的线程数 = CPU核数+1

IO密集型

IO密集型任务线程并不是一直在执行任务，则应配置经可能多的线程，如

IO密集型，即该任务需要大量的IO，即大量的阻塞。在单线程上运行IO密集型的任务会导致浪费大量的 CPU运算能力浪费在等待。
所以在IO密集型任务中使用多线程可以大大的加速程序运行，即使在单核CPU上，这种加速主要就是利用了被浪费掉的阻塞时间。
IO密集型时，大部分线程都阻塞，故需要多配置线程数，

参考公式1：

线程池的线程数 = CPU核数 * 2

参考公式2：

线程池的线程数 = CPU核数 /（1-阻塞系数）
阻塞系数在0.8~0.9之间
例如八核CPU的线程数 = 8 / (1-0.9) = 80