java线程池

Posted 2021-11-27 imyhb

前言：线程池技术是通过对线程资源的统一管理来达到对线程资源的重复利用，降低线程频繁创建和销毁的开销。java jdk在java.util.concurrent并发包中有一套现成的对线程池的实现方案，我们可以直接拿来使用，快速实现多线程并发编程场景。这里对concurrent包中的线程池框架的实现进行一些分析。

 

java线程池使用代码示例

public class Test {
    public static void main(String[] args) throws Exception {
        Task task1 = new Task(1);
        Task task2 = new Task(2);

//        ExecutorService normalExecutor = new ThreadPoolExecutor(2, 4, 200, TimeUnit.MILLISECONDS,
//                new ArrayBlockingQueue<Runnable>(5));
//        ExecutorService singleExecutor = Executors.newSingleThreadExecutor();
//        ExecutorService cachedExecutor = Executors.newCachedThreadPool();
        //创建线程池服务
　　　　 ExecutorService executor = Executors.newFixedThreadPool(2);
　　　　　//将任务交给线程池执行
        executor.execute(task1);
        executor.execute(task2);
        executor.shutdown();
    }
}

//可以提交给线程池执行的任务类，线程池执行任务时会执行其中的run方法
class Task implements Runnable {
    private int taskNum;

    public Task(int num) {
        this.taskNum = num;
    }

    public void run() {
        System.out.println("开始执行任务：" + taskNum);
        try {
            Thread.currentThread().sleep(10000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("结束执行任务：" + taskNum);
    }
}

 

执行结果如下：

从结果可以看出后提交给线程池的任务先执行了。所以执行execute方法时只是将任务提交给线程池去管理，任务的执行顺序是由线程池内部去协调的。

 

java线程池实现原理

java线程池的核心实现类是ThreadPoolExecutor，该类的继承关系如下：

最底层其实现的接口Executor的定义如下：

public interface Executor {
    void execute(Runnable command);
}

 

 可以看到，该接口只有一个方法execute，ThreadPoolExecutor实现该方法后，通过该方法的调用将任务提交给线程池。所以ThreadPoolExecutor.execute里的逻辑就是线程池执行任务的密码所在。

这里先关注下ThreadPoolExecutor类中如下几个比较重要的常量

//记录当前线程池中工作线程的数量
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
//将一个整形的32位分为两部分，高3位和低29位
private static final int COUNT_BITS = Integer.SIZE - 3;
//将整形的低29位用于存储工作线程数，所以可开启的最大线程数为2的29次方
private static final int CAPACITY   = (1 << COUNT_BITS) - 1;

//将整形的高3位用于存储线程池的当前状态值
private static final int RUNNING    = -1 << COUNT_BITS;
private static final int SHUTDOWN   =  0 << COUNT_BITS;
private static final int STOP       =  1 << COUNT_BITS;
private static final int TIDYING    =  2 << COUNT_BITS;
private static final int TERMINATED =  3 << COUNT_BITS;

 

ThreadPoolExecutor里很多地方使用了类似的位运算的方式进行状态值的存储和逻辑运算来提高运行效率

现在看下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.acc = System.getSecurityManager() == null ?
                null :
                AccessController.getContext();
        this.corePoolSize = corePoolSize;
        this.maximumPoolSize = maximumPoolSize;
        this.workQueue = workQueue;
        this.keepAliveTime = unit.toNanos(keepAliveTime);
        this.threadFactory = threadFactory;
        this.handler = handler;
    }

 

这里涉及的几个核心变量解释如下：

corePoolSize：核心线程数，线程池运行稳定后维持的线程数
maximumPoolSize：最大线程数，线程池最多可以使用的线程数
keepAliveTime：超时时间，线程在该超时时间间隔内从任务队列未获取到任务时，若线程池工作线程数超过corePoolSize，则关闭当前线程，且线程池线程数减1
unit：keepAliveTime使用的时间单位
workQueue：任务存放的队列
threadFactory：线程工厂，线程池使用该工厂类的方法产生线程
handler：当线程池中线程数已达最大值，且任务队列已满，无法处理新加入的任务时。由自定义的handler处理该任务

ThreadPoolExecutor.execute对任务的处理流程如下图：

 

 ThreadPoolExecutor.execute的执行示意图如下：

 

 通过ThreadPoolExecutor构造函数的参数，我们发现如果我们要通过ThreadPoolExecutor创建一个适合我们业务场景的线程池，需要对ThreadPoolExecutor的运行原理和几个核心参数有比较深入的理解。线程池的设计者在这方面也做了一定的考虑，在concurrent包中，提供了一个有用的工具类Executors，这个类提供了一些工厂方法可以帮助我们简单方便的创建出适用于各种场景的线程池，有些方法就是对ThreadPoolExecutor做了简单的封装。其中，业务上比较常用到的获取线程池的工厂方法有如下几个：

//创建固定大小的线程池,在并发任务比较多的场景中比较常用
public static ExecutorService newFixedThreadPool(int nThreads) {
    return new ThreadPoolExecutor(nThreads, nThreads,
                                  0L, TimeUnit.MILLISECONDS,
                                  new LinkedBlockingQueue<Runnable>());
}

//创建一个单线程化的线程池,线程池只使用一个线程执行所有的任务，可以保证任务的执行顺序
public static ExecutorService newSingleThreadExecutor() {
    return new FinalizableDelegatedExecutorService
        (new ThreadPoolExecutor(1, 1,
                                0L, TimeUnit.MILLISECONDS,
                                new LinkedBlockingQueue<Runnable>()));
}

//创建一个可缓存线程池，队列只能存放一个元素，任务会及时被线程处理，适用于对任务处理的及时性要求比较高的场景
public static ExecutorService newCachedThreadPool() {
    return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                  60L, TimeUnit.SECONDS,
                                  new SynchronousQueue<Runnable>());
}