Java线程池

Posted 2022-10-30 ty_laurel

  多线程的软件设计方法确实可以最大限度地发挥现代多核处理器的计算能力，提高生产系统的吞吐量和性能。但是，若不加控制和管理的随意使用线程，对系统的性能反而会产生不利的影响。

什么是线程池

  为了避免系统频繁地创建和销毁线程，我们可以让创建的线程进行复用。比如数据库中的数据库连接池，为了避免每次数据库查询都重新建立和销毁数据库连接，我们可以使用数据库连接池维护一些数据库连接，让他们长期保持在一个激活状态。当系统需要使用数据库时，并不是创建一个新的连接，而是从连接池中获得一个可用的连接即可。反之，当需要关闭连接时，并不真的把连接关闭，而是将这个连接“还”给连接池。通过连接池这种方法可以节约不少创建和销毁对象的时间。

  线程池也是类似的概念。线程池中，总有几个活跃的线程。当你需要使用线程时，可以从池子中随便拿一个空闲线程，当完成工作时，并不急关闭线程，而是将这个线程退回到池子，方便其他人使用。也就是说，创建线程变成了从线程池获得空闲线程，关闭线程变成了向池子归还线程。

  调用Executors的相关静态方法（下边介绍）创建线程池后，就可以调用ThreadPoolExecutor类的execute(task)(task是Runnable类型的任务)方法执行给定的Runnable任务task了。接下来先看下Executors的几个常用的创建线程池的静态方法。

newFixedThreadPool()方法

该静态方法返回一个固定线程数量的线程池。该线程池中的线程数量始终不变。有新任务提交时，线程池中有空闲线程，则立即执行；若没有，则新任务被暂存在一个任务队列中，直到有空闲线程。若在关闭前的执行期间由于失败而导致任何线程终止，那么一个新线程将代替它执行后续的任务（若需要）。

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

可以看到该方法调用ThreadPoolExecutor方法的参数情况，corePool和maximumPoolSize是相同的，也就是该线程池中线程数达到这个数目后就不会再继续创建新的线程，而是添加到LinkedBlockingQueue()队列中，该队列是按FIFO先进先出排序元素，在此处的容量为Integer.MAX_VALUE（此容量的队列有时也称为无界队列）。

newSignalThreadExecutor()方法

返回只有一个worker线程的线程池。以无界队列LinkedBlockingQueue方式运行该线程。如果因为在关闭前的执行期间出现失败而终止了此单个线程，若需要，一个新县城将代替它执行后续的任务。多余任务保存在任务队列，先入先出顺序执行。

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

newCachedThreadPool()方法

返回一个可根据实际情况调整线程数量的线程池。线程池的线程数量不确定，但若有空闲线程可以复用，则会优先使用可复用的线程。若所有线程均在工作，又有新任务提交，则会创建新的线程处理任务。所有线程在当前任务执行完毕后，将返回线程池进行复用。

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

由定义可看出，该线程池最大线程数量为Integer.MAX_VALUE，若线程空闲60s未被使用则被终止，调用SynchronousQueue作为任务队列，该队列是一个没有数据缓冲的BlockingQueue，每个插入操作必须等待另一个线程的对应移除操作，否则将被挂起。

newScheduledThreadPool()方法

返回ScheduledExecutorService对象，但该方法创建一个指定线程数量的线程池，可安排在给定延迟后运行命令或顶起地执行。

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


public ScheduledThreadPoolExecutor(int corePoolSize) 
        super(corePoolSize, Integer.MAX_VALUE, 0, TimeUnit.NANOSECONDS,
              new DelayedWorkQueue());

ScheduledThreadPoolExecutor继承了ThreadPoolExecutor类，使用super调用其父类的构造方法，实现线程池的创建。

线程池内部实现ThreadPoolExecutor

  Executors类则扮演者线程池工厂的角色，通过Executors的静态方法可以取得一个拥有特定功能的线程池。上一步介绍的几个核心线程池，虽看起来创建的线程有着完全不同的功能特点，但其内部实现均使用了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.corePoolSize = corePoolSize;
        this.maximumPoolSize = maximumPoolSize;
        this.workQueue = workQueue;
        this.keepAliveTime = unit.toNanos(keepAliveTime);
        this.threadFactory = threadFactory;
        this.handler = handler;

可以通过给定初始化参数创建一个新的ThreadPoolExecutor，对于几个参数的含义如下介绍:

• corePoolSize：线程池的基本大小，字面意思就是线程池核心大小，当提交一个任务到线程池时，线程池就会创建一个新的线程来执行任务，直到线程池中的线程数达到了corePoolSize大小，，就会优先使用空闲线程。
• maximumPoolSize：线程池中允许创建的最大线程数。如果线程数达到了这个数目，即使有更多的任务在排队等待被执行，线程池也不会创建新的线程了。
• keepAliveTime：如果线程数比corePoolSize多，并且有线程是空闲的，这些线程不会立即销毁，而是会等待keepAliveTime这么长时间以后才会销毁。
• **unit：**keepAliveTime的单位。可以是天、小时、分钟、毫秒、微妙和纳秒。
• workQueue：任务的排队队列。当线程池中线程比较少，任务比较多的时候，任务就一定会排队。任务排队的时候就放在此处指定的数据结构中。
• threadFactory：线程工厂。默认defaultThreadFactory。
• handler：如果线程池已经满了，任务队列也满了，对于新提交的任务就会执行handler指定的策略。默认是执行AbortPolicy，直接终止。

workQueue是一个BlockingQueue接口的对象，仅用于存放Runnable对象。根据队列功能分类，在ThreadPoolExecutor的构造函数中可使用以下几种BlockingQueue：

• 直接提交的队列：该功能由SynchronousQueue对象提供，其是一个特殊的BlockingQueue，没有容量，每一个插入操作都要等待一个相应的删除操作
• 有界的任务队列：可以使用ArrayBlockingQueue实现，其构造函数必须带一个容量参数，表示该队列最大容量。
• 无界的任务队列：通过LinkedBlockingQueue类实现，与有界队列相比，除非资源耗尽，否则无界的任务队列不存在任务入队失败的情况。
• 优先任务队列：带有执行优先级的队列，通过PriorityBlockingQueue实现，可以控制人物的执行先后顺序，是一个特殊的无界队列。

当把一个Runnable交给线程池去执行的时候，其处理流程大概如下：

• 1.先判断线程池中的核心线程们是否空闲，如果空闲，就把这个新的任务指派给某一个空闲线程去执行。如果没有空闲，并且当前线程池中的核心线程数还小于corePoolSize，那就再创建一个核心线程。
• 2.如果线程池的线程数已经达到核心线程数，并且这个线程都繁忙，就把这个新来的任务放到等待队列中去。
• 3.如果等待队列满了，则会查看下当前线程数是否达到了maximumPoolSize，如果还未达到，就继续创建线程。如果已经达到，就交给RejectedExecutionHandler来决定怎么处理这个任务。

ThreadPoolExecutor线程池的核心调度代码是execute函数，该段代码充分体现了线程池的工作逻辑：

public void execute(Runnable command) 
    if(command == null)
        throw new NullPointerException();
    int c = ctl.get();
    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))    //17行
        reject(command);

代码第五行的workerCountOf()函数取得了当前线程池的线程总数。当线程总数小于corePoolSize核心线程数时，会将任务通过addWorker()方法直接调度执行。否则，则在第十行代码处（workQueue.offer()）进入等待队列。如果进入等待队列失败（比如有界队列到达了上限或者使用了SynchronousQueue），则会执行第17行，将任务直接提交给线程池。如果当前线程数已经达到了maximumPoolSize，则提交失败，就执行18行的拒绝策略。

ThreadPoolExecutor的任务调度逻辑：

超负载了怎么办：拒绝策略

  拒绝策略可以说是系统超负荷运行时的补救措施，通常由于压力太大而引起的，也就是线程池中的线程用完了，无法继续为新任务服务，同时，等待队列中也已经排满了，再也塞不下新任务了。这时就需要一套机制合理的处理这个问题。

JDK内置提供了四种拒绝策略：

• AbortPolicy策略：会直接抛出异常，阻止系统正常工作，默认策略；
• CallerRunsPolicy策略：只要线程池未关闭，该策略直接在调用者线程中，运行当前被丢弃的任务。显然这样做不会真的丢弃任务，但是，任务提交现车你给的额性能极有可能会急剧下降。
• DiscardOledestPolicy策略：该策略将丢弃最老的一个请求，也就是即将被执行的一个任务，并尝试再次提交当前任务。
• DiscardPolicy策略：该策略默默地丢弃无法处理的任务，不予任何处理。如果允许任务丢失，我觉得这可能是最好的一种方案了吧。
  若以上策略仍无法满足实际应用需要，完全可以自己扩展RejectExecutionHandler接口。

ThreadFactory工厂

线程池通过ThreadFactory接口来创建线程，其只有一个方法：
Thread newThread(Runnable r);
ThreadPoolExecutor类中若没有指定线程工程ThreadFactory，就会调用默认的工厂defaultThreadFactory，该静态方法中封装了DefaultThreadFactory类，其实现了ThreadFactory接口及newThread方法，其实现如下：

static class DefaultThreadFactory implements ThreadFactory 
        private static final AtomicInteger poolNumber = new AtomicInteger(1);
        private final ThreadGroup group;
        private final AtomicInteger threadNumber = new AtomicInteger(1);
        private final String namePrefix;

        DefaultThreadFactory() 
            SecurityManager s = System.getSecurityManager();
            group = (s != null) ? s.getThreadGroup() :
                                  Thread.currentThread().getThreadGroup();
            namePrefix = "pool-" +
                          poolNumber.getAndIncrement() +
                         "-thread-";
        

        public Thread newThread(Runnable r) 
            Thread t = new Thread(group, r,
                                  namePrefix + threadNumber.getAndIncrement(),
                                  0);
            if (t.isDaemon())
                t.setDaemon(false);
            if (t.getPriority() != Thread.NORM_PRIORITY)
                t.setPriority(Thread.NORM_PRIORITY);
            return t;
        

很明显，newThread方法就是该工厂类的工厂方法，也就是其核心实现。

扩展线程池

若想要对线程池做一些扩展，比如，监控每个任务执行的开始和结束时间，或者其他一些自定义的增强功能，如何去做？
其实，ThreadPoolExecutor是一个可扩展的线程池。它提供了beforeExecute()、afterExecute()和terminated()三个接口对线程池进行控制。
如下使用：

boolean ran = false;
beforeExecute(thread, task);             //运行前
try 
    task.run();                          //运行任务
    ran = true;
    afterExecute(task, null);             //运行结束
    ++completedTasks;
 catch (RuntimeExecption ex) 
    if(!ran)
        afterExecute(task, ex);           //运行结束
    throw ex;

默认的ThreadPoolExecutor实现中，提供了空的beforeExecute()和afterExecute()实现。实际应用中，可以对其进行扩展来实现对线程池运行状态的跟踪，输出一些有用的调试信息，以帮助系统故障诊断，这对于多线程程序错误排查是很有帮助的。

在提交任务时若直接使用submit方法，会导致没有异常堆栈提示信息（如除法运算，除数为0）。可以改用execute()方法如下:
pools.execute(new Task());
或者改造submit():

Future re = pools.submit(new Task());
re.get();