线程池 · Rwhither’s Blog

概述 #

背景：经常创建和销毁、使用量特别大的资源，比如并发情况下的线程，对性能影响很大。

思路：提前创建好多个线程，放入线程池中，使用时直接获取，使用完放回池中。可以避免频繁创建销毁，实现重复利用。类似生活中的公共交通工具。

好处 #

降低资源的消耗；
提高响应的速度；
方便管理；

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

普通使用 #

JDK5.0 起提供了线程池相关API： ExecutorService 和 Executors

ExecutorService：真正的线程池接口。常见子类ThreadPoolExecutor

void execute（Runnable command）：执行任务/命令，没有返回值，一般用来执行Runnable
<T>Future<T> submit Callable<T> task）：执行任务，有返回值，一般用来执行Callable
void shutdown（）：关闭连接池

Executors：工具类、线程池的工厂类，用于创建并返回不同类型的线程池

public class Demo6_CreateCallable implements Callable<Boolean> {

    private String url;// 网络图片地址
    private String name;// 报错扥文件名

    // 有参构造
    public Demo6_CreateCallable(String url, String name) {
        this.url = url;
        this.name = name;
    }

    // 下载图片线程的执行体
    public Boolean call() throws Exception {
        WebDownloader webDownloader = new WebDownloader();
        webDownloader.downloader(url, name);
        System.out.println(" 下载了文件名为:" + name);
        return true;
    }

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        Demo6_CreateCallable c = new Demo6_CreateCallable("https://img-home.csdnimg.cn/images/20201124032511.png", "1.png");
        Demo6_CreateCallable c1 = new Demo6_CreateCallable("https://img-home.csdnimg.cn/images/20201124032511.png", "2.png");
        Demo6_CreateCallable c2 = new Demo6_CreateCallable("https://img-home.csdnimg.cn/images/20201124032511.png", "3.png");
        // 创建执行服务
        ExecutorService ser = Executors.newFixedThreadPool(3);
        // 提交执行
        Future<Boolean> r = ser.submit(c);
        Future<Boolean> r1 = ser.submit(c1);
        Future<Boolean> r2 = ser.submit(c2);
        // 获取结果
        boolean res = r.get();
        boolean res1 = r1.get();
        boolean res2 = r2.get();
        // 关闭服务
        ser.shutdownNow();
    }
}
//class WebDownloader 在前面下载图片已经定义了，这里就不用再次写，直接使用就好

//测试线程池
public class Demo35_ThreadPool {
    public static void main(String[] args) {
        // 1. 创建服务
        // newFixedThreadPool(线程池大小)
        ExecutorService service = Executors.newFixedThreadPool(10);
        //执行
        service.execute(new MyThread());
        service.execute(new MyThread());
        service.execute(new MyThread());
        service.execute(new MyThread());
        service.execute(new MyThread());
        service.execute(new MyThread());
        //关闭连接
        service.shutdown();
    }
}

class MyThread implements Runnable {
    @Override
    public void run() {
        System.out.println(Thread.currentThread().getName());
    }
}

三大方法 #

ExecutorService threadPool = Executors.newSingleThreadExecutor();//单个线程
ExecutorService threadPool2 = Executors.newFixedThreadPool(5); //创建一个固定的线程池的大小
ExecutorService threadPool3 = Executors.newCachedThreadPool(); //可伸缩的

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

//工具类 Executors 三大方法；
public class Main {
    public static void main(String[] args) {

        ExecutorService threadPool = Executors.newSingleThreadExecutor();//单个线程
        ExecutorService threadPool2 = Executors.newFixedThreadPool(5); //创建一个固定的线程池的大小
        ExecutorService threadPool3 = Executors.newCachedThreadPool(); //可伸缩的

        //线程池用完必须要关闭线程池
        //通过选择不同的线程池可以得到不同的运行结果
        try {

            for (int i = 1; i <=100 ; i++) {
                //通过线程池创建线程
                threadPool3.execute(()->{
                    System.out.println(Thread.currentThread().getName()+ " ok");
                });
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            threadPool.shutdown();
        }
    }
}

Executors和ThreadPoolExecutor的区别 #

查看阿里巴巴开发手册可以发现：编程规约->并发处理

并不推荐我们使用 Executors 创建线程池，查看源码发现：

 public static ExecutorService newSingleThreadExecutor() {
        return new FinalizableDelegatedExecutorService
            (new ThreadPoolExecutor(1, 1,
                                    0L, TimeUnit.MILLISECONDS,
                                    new LinkedBlockingQueue<Runnable>()));
    }
    
    
public static ExecutorService newFixedThreadPool(int nThreads) {
        return new ThreadPoolExecutor(nThreads, nThreads,
                                      0L, TimeUnit.MILLISECONDS,
                                      new LinkedBlockingQueue<Runnable>());
    }
    
    
public static ExecutorService newCachedThreadPool() {
        return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                      60L, TimeUnit.SECONDS,
                                      new SynchronousQueue<Runnable>());
    }

使用 Executors 创建线程池，本质上就是使用 new ThreadPoolExecutor(....)来创建线程池。只不过后者更加自定义化。

七大参数 #

查看创建new 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;
}

四种拒绝策略 #

new ThreadPoolExecutor.AbortPolicy()： //该拒绝策略为：银行满了，还有人进来，不处理这个人的，并抛出异常。超出最大承载，就会抛出异常：队列容量大小+maxPoolSize
new ThreadPoolExecutor.CallerRunsPolicy()： //该拒绝策略为：哪来的去哪里 main线程进行处理
new ThreadPoolExecutor.DiscardPolicy(): //该拒绝策略为：队列满了,丢掉异常，不会抛出异常。
new ThreadPoolExecutor.DiscardOldestPolicy()： //该拒绝策略为：队列满了，尝试去和最早的进程竞争，不会抛出异常

自定义线程池 #

如何设置线程池的大小？

CPU密集型：电脑的核数是几核就选择几；选择maximunPoolSize的大小
I/O密集型：在程序中有15个大型任务，IO十分占用资源；I/O密集型就是判断我们程序中十分耗I/O的线程数量，大约是最大I/O数的一倍到两倍之间。

import java.util.concurrent.*;

public class Main {
    public static void main(String[] args) {
        // 获取cpu 的核数
        int max = Runtime.getRuntime().availableProcessors();
        ExecutorService service =new ThreadPoolExecutor(
                2,
                max,
                3,
                TimeUnit.SECONDS,
                new LinkedBlockingDeque<>(3),
                Executors.defaultThreadFactory(),
                new ThreadPoolExecutor.AbortPolicy()
        );
        //这里使用AbortPolicy所以，如果有7个（队列容量+maxPoolSize）线程进来是可以执行的，但是如果有8个的话会报RejectedExecutionException 
        try {
            for (int i = 1; i <= 8; i++) {
                service.execute(() -> {
                    System.out.println(Thread.currentThread().getName() + "ok");
                });
            }
        }catch (Exception e) {
            e.printStackTrace();
        }
        finally {
            service.shutdown();
        }
    }
}