Java线程池的实现原理

Posted 2023-02-21 yxz1025

Java中的线程池的使用场景有很多种，大多数并发框架及并发场景都使用了线程池来管理线程，使用线程池带来的好处主要有以下几点。

第一：提升系统资源利用。重复的创建线程对于系统开销很大，通过重复利用已经创建的线程提高系统的性能

第二：提高系统响应时间。当一个任务被提交不需要等待去创建线程，从而提升响应时间

第三：方便管理线程。通过合理的配置线程池，来优化及使用线程

Java线程池的实现原理是怎样的呢？当向一个线程池提交一个任务后，线程池是如何运转的呢？首先看如下流程图：

线程池调度流程图（摘自Java并发编程的艺术）

 

 如上图所示，当一个任务被提交给线程池是做了如下处理

1. 首先判断核心线程池是否已经满了，如果未满则创建线程执行任务，当创建的工作线程已经达到核心线程数，则进入阻塞队列
2. 阻塞队列有很多种，常用的比如ArrayBlockingQueue、LinkedBlockingQueue等等，阻塞队列是一种基于AQS实现的线程安全的队列，当任务进入阻塞队列时，首先判断队列是否已经满了，如果未满则加入到队列中，否则判断最大线程池。
3. 如果已经达到最大线程池数量，则采用拒绝策略阻止任务提交，否则创建线程进行任务执行

下面看看线程池ThreadPoolExecutor的execute方法的执行流程

execute方法执行流程图（摘自Java并发编程的艺术）

 

如图所示当主线程执行execute的方法时流程如下4步：

• 首先用核心线程来执行任务，同时通过死循环不断从阻塞队列获取提交的等待任务
• 当核心线程数不够用时，将多余任务加入到阻塞队列
• 当阻塞队列满时，使用最大线程数创建新线程
• 如果最大线程数已经达到上限，则执行第四步操作进入拒绝策略

下面看看execute方法的源码：

    public void execute(Runnable command) 
        if (command == null)
            throw new NullPointerException();
        int c = ctl.get();
        //若工作线程总数是否小于核心线程数，将任务包装为Work任务，并启动线程执行
        if (workerCountOf(c) < corePoolSize) 
            if (addWorker(command, true))
                return;
            c = ctl.get();
        
        //若已经超过了核心线程数，则将当前任务加入到阻塞队列，否则执行拒绝策略
        if (isRunning(c) && workQueue.offer(command)) 
            int recheck = ctl.get();
            if (! isRunning(recheck) && remove(command))
                reject(command);
            else if (workerCountOf(recheck) == 0)
                addWorker(null, false);
        
        else if (!addWorker(command, false))
            reject(command);
    

当向线程池提交一个任务时，线程池将任务封装为一个worker,worker继承了AQS并实现了Runnable, Worker在执行完自身任务后，通过run方法循环从队列获取任务进行执行

    final void runWorker(Worker w) 
        Thread wt = Thread.currentThread();
        Runnable task = w.firstTask;
        w.firstTask = null;
        w.unlock(); // allow interrupts
        boolean completedAbruptly = true;
        try 
            //这里先获取当前任务task，如果存在并且不为Null,则执行当前任务，执行结束设置为Null，其他情况直接getTask从阻塞队列循环获取任务并启动执行
            while (task != null || (task = getTask()) != null) 
                w.lock();
                if ((runStateAtLeast(ctl.get(), STOP) ||
                     (Thread.interrupted() &&
                      runStateAtLeast(ctl.get(), STOP))) &&
                    !wt.isInterrupted())
                    wt.interrupt();
                try 
                    beforeExecute(wt, task);
                    Throwable thrown = null;
                    try 
                        task.run();
                     catch (RuntimeException x) 
                        thrown = x; throw x;
                     catch (Error x) 
                        thrown = x; throw x;
                     catch (Throwable x) 
                        thrown = x; throw new Error(x);
                     finally 
                        afterExecute(task, thrown);
                    
                 finally 
                    task = null;
                    w.completedTasks++;
                    w.unlock();
                
            
            completedAbruptly = false;
         finally 
            processWorkerExit(w, completedAbruptly);
        
    

线程池的创建,我们通过使用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;
    

在创建线程池时的几个参数：

1. corePoolSize表示核心线程池，当向线程池提交任务时，只要核心线程数未达到设定的值，都会创建任务，即便有空闲线程可以用，为了性能考虑，可以在创建线程池后调用prestartCoreThread方法一次性创建所有核心线程数
2. maximumPoolSize表示线程池最大的上线数
3. keepAliveTime表示线程存活时间，配合unit使用
4. unit表示存活的单位，秒、分钟等
5. workQueue表示阻塞队列，当工作线程数达到核心线程数量时，新的任务就会进入阻塞队列，关于阻塞队列后续专门找时间分享
6. threadFactory表示用来通过executor创建新的线程对象
7. handler表示拒绝策略，当线程数量达到最大线程数时，新的任务就会被拒绝

在向线程池提交任务，除了execute方法外，可以使用submit方法

    public Future<?> submit(Runnable task) 
        if (task == null) throw new NullPointerException();
        RunnableFuture<Void> ftask = newTaskFor(task, null);
        execute(ftask);
        return ftask;
    

此方法会返回一个future用来获取执行结果，具体可以看看AbstractExecutorService抽象类提供的方法。当需要关闭线程池时，有几种方法可以选，一种是shutdown

    public void shutdown() 
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try 
            checkShutdownAccess();
            advanceRunState(SHUTDOWN);
            interruptIdleWorkers();
            onShutdown(); // hook for ScheduledThreadPoolExecutor
         finally 
            mainLock.unlock();
        
        tryTerminate();
    

调用此关闭线程池方法，会遍历所有的工作线程将其进行中断，已经提交的任务会继续执行，新的提交的任务将会被中断 。最后说一下合理的使用线程池很有必要，线程池也不是设置越大越好，这会导致额外的线程切换带来的系统性能开销，至于为何这么说，大家可以读读操作系统原理这本书。(如需了解更多请关注鄙人公众号?)