首页 > 代码库 > Java线程池

Java线程池

2024-08-15 01:23:38   219人阅读

一. 问题引出

网络请求通常有两种场景：第一种，请求不是很频繁，而且每次连接后会保持相当一段时间来读数据或者写数据，最后断开，如文件下载，网络流媒体等。第二种是请求频繁，但是连接上以后读/写很少量的数据就断开连接。
考虑到服务的并发问题，如果每个请求来到以后服务都为它启动一个线程，那么这对服务的资源可能会造成很大的浪费，特别是第二种情况。如果并发的线程数量很多，并且每个线程都是执行一个时间很短的任务就结束了，这样频繁创建线程就会大大降低系统的效率，因为频繁创建线程和销毁线程需要时间。
那么有没有一种办法使得线程可以复用，就是执行完一个任务，并不被销毁，而是可以继续执行其他的任务？
在Java中可以通过线程池来达到这样的效果。

二. Java中的ThreadPoolExecutor类

java.uitl.concurrent.ThreadPoolExecutor类是线程池中最核心的一个类，因此如果要透彻地了解Java中的线程池，必须先了解这个类。下面我们来看一下ThreadPoolExecutor类的具体实现源码。

在ThreadPoolExecutor类中提供了四个构造方法：

public class ThreadPoolExecutor extends AbstractExecutorService {    public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,            BlockingQueue<Runnable> workQueue);    public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,            BlockingQueue<Runnable> workQueue,ThreadFactory threadFactory);    public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,            BlockingQueue<Runnable> workQueue,RejectedExecutionHandler handler);    public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,        BlockingQueue<Runnable> workQueue,ThreadFactory threadFactory,RejectedExecutionHandler handler);    }
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13

从上面的代码可以得知，ThreadPoolExecutor继承了AbstractExecutorService类，并提供了四个构造器，事实上，通过观察每个构造器的源码具体实现，发现前面三个构造器都是调用的第四个构造器进行的初始化工作。

下面解释下一下构造器中各个参数的含义：

  • corePoolSize：核心池的大小，这个参数跟后面讲述的线程池的实现原理有非常大的关系。在创建了线程池后，默认情况下，线程池中并没有任何线程，而是等待有任务到来才创建线程去执行任务，除非调用了prestartAllCoreThreads()或者prestartCoreThread()方法，从这2个方法的名字就可以看出，是预创建线程的意思，即在没有任务到来之前就创建corePoolSize个线程或者一个线程。默认情况下，在创建了线程池后，线程池中的线程数为0，当有任务来之后，就会创建一个线程去执行任务，当线程池中的线程数目达到corePoolSize后，就会把到达的任务放到缓存队列当中；
  • maximumPoolSize：线程池最大线程数，这个参数也是一个非常重要的参数，它表示在线程池中最多能创建多少个线程；
  • keepAliveTime：表示线程没有任务执行时最多保持多久时间会终止。默认情况下，只有当线程池中的线程数大于corePoolSize时，keepAliveTime才会起作用，直到线程池中的线程数不大于corePoolSize，即当线程池中的线程数大于corePoolSize时，如果一个线程空闲的时间达到keepAliveTime，则会终止，直到线程池中的线程数不超过corePoolSize。但是如果调用了allowCoreThreadTimeOut(boolean)方法，在线程池中的线程数不大于corePoolSize时，keepAliveTime参数也会起作用，直到线程池中的线程数为0；
  • unit：参数keepAliveTime的时间单位，有7种取值，在TimeUnit类中有7种静态属性： TimeUnit.DAYS; //天 TimeUnit.HOURS; //小时 TimeUnit.MINUTES; //分钟
    TimeUnit.SECONDS; //秒 TimeUnit.MILLISECONDS; //毫秒
    TimeUnit.MICROSECONDS; //微妙 TimeUnit.NANOSECONDS; //纳秒
  • workQueue：一个阻塞队列，用来存储等待执行的任务，这个参数的选择也很重要，会对线程池的运行过程产生重大影响，一般来说，这里的阻塞队列有以下几种选择：
    ArrayBlockingQueue; LinkedBlockingQueue; SynchronousQueue;
    ArrayBlockingQueue和PriorityBlockingQueue使用较少，一般使用LinkedBlockingQueue和Synchronous。线程池的排队策略与BlockingQueue有关。
  • threadFactory：线程工厂，主要用来创建线程； handler：表示当拒绝处理任务时的策略，有以下四种取值：
    ThreadPoolExecutor.AbortPolicy:丢弃任务并抛出RejectedExecutionException异常。
    ThreadPoolExecutor.DiscardPolicy：也是丢弃任务，但是不抛出异常。
    ThreadPoolExecutor.DiscardOldestPolicy：丢弃队列最前面的任务，然后重新尝试执行任务（重复此过程）
    ThreadPoolExecutor.CallerRunsPolicy：由调用线程处理该任务

在ThreadPoolExecutor类中有几个非常重要的方法：

  • execute()
  • submit()
  • shutdown()
  • shutdownNow()

execute()方法实际上是Executor中声明的方法，在ThreadPoolExecutor进行了具体的实现，这个方法是ThreadPoolExecutor的核心方法，通过这个方法可以向线程池提交一个任务，交由线程池去执行。
submit()方法是在ExecutorService中声明的方法，在AbstractExecutorService就已经有了具体的实现，在ThreadPoolExecutor中并没有对其进行重写，这个方法也是用来向线程池提交任务的，但是它和execute()方法不同，它能够返回任务执行的结果，去看submit()方法的实现，会发现它实际上还是调用的execute()方法，只不过它利用了Future来获取任务执行结果。
shutdown()和shutdownNow()是用来关闭线程池的,调用shutdown()方法，不会立即终止线程池，而是要等所有任务缓存队列中的任务都执行完后才终止，但再也不会接受新的任务；调用了shutdownNow()方法，立即终止线程池，并尝试打断正在执行的任务，并且清空任务缓存队列，返回尚未执行的任务。
还有很多其他的方法，比如：getQueue() 、getPoolSize() 、getActiveCount()、getCompletedTaskCount()等获取与线程池相关属性的方法，有兴趣的朋友可以自行查阅API。

三. ThreadPoolExecutor、AbstractExecutorService、ExecutorService和Executor之间的关系

AbstractExecutorService 类源码

public abstract class AbstractExecutorService implements ExecutorService {    protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) { };    protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) { };    public Future<?> submit(Runnable task) {};    public <T> Future<T> submit(Runnable task, T result) { };    public <T> Future<T> submit(Callable<T> task) { };    private <T> T doInvokeAny(Collection<? extends Callable<T>> tasks,                            boolean timed, long nanos)        throws InterruptedException, ExecutionException, TimeoutException {    };    public <T> T invokeAny(Collection<? extends Callable<T>> tasks)        throws InterruptedException, ExecutionException {    };    public <T> T invokeAny(Collection<? extends Callable<T>> tasks,                           long timeout, TimeUnit unit)        throws InterruptedException, ExecutionException, TimeoutException {    };    public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)        throws InterruptedException {    };    public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks,                                         long timeout, TimeUnit unit)        throws InterruptedException {    };}
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27

ExecutorService 类源码

public interface ExecutorService extends Executor {    void shutdown();    boolean isShutdown();    boolean isTerminated();    boolean awaitTermination(long timeout, TimeUnit unit)        throws InterruptedException;    <T> Future<T> submit(Callable<T> task);    <T> Future<T> submit(Runnable task, T result);    Future<?> submit(Runnable task);    <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)        throws InterruptedException;    <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks,                                  long timeout, TimeUnit unit)        throws InterruptedException;    <T> T invokeAny(Collection<? extends Callable<T>> tasks)        throws InterruptedException, ExecutionException;    <T> T invokeAny(Collection<? extends Callable<T>> tasks,                    long timeout, TimeUnit unit)        throws InterruptedException, ExecutionException, TimeoutException;}
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22

Executor 类源码

public interface Executor {    /**     * Executes the given command at some time in the future.  The command     * may execute in a new thread, in a pooled thread, or in the calling     * thread, at the discretion of the <tt>Executor</tt> implementation.     *     * @param command the runnable task     * @throws RejectedExecutionException if this task cannot be     * accepted for execution.     * @throws NullPointerException if command is null     */    void execute(Runnable command);}
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14

Executor是一个顶层接口，在它里面只声明了一个方法execute(Runnable)，返回值为void，参数为Runnable类型，从字面意思可以理解，就是用来执行传进去的任务的；
然后ExecutorService接口继承了Executor接口，并声明了一些方法：submit、invokeAll、invokeAny以及shutDown等；
抽象类AbstractExecutorService实现了ExecutorService接口，基本实现了ExecutorService中声明的所有方法；
然后ThreadPoolExecutor继承了类AbstractExecutorService。

四. 深入线程池

1. 线程池状态

在ThreadPoolExecutor中定义了一个volatile变量，另外定义了几个static final变量表示线程池的各个状态：

volatile int runState;static final int RUNNING    = 0;static final int SHUTDOWN   = 1;static final int STOP       = 2;static final int TERMINATED = 3;
  • 1
  • 2
  • 3
  • 4
  • 5

runState表示当前线程池的状态，它是一个volatile变量用来保证线程之间的可见性；
下面的几个static final变量表示runState可能的几个取值。
当创建线程池后，初始时，线程池处于RUNNING状态；
如果调用了shutdown()方法，则线程池处于SHUTDOWN状态，此时线程池不能够接受新的任务，它会等待所有任务执行完毕；
如果调用了shutdownNow()方法，则线程池处于STOP状态，此时线程池不能接受新的任务，并且会去尝试终止正在执行的任务；
当线程池处于SHUTDOWN或STOP状态，并且所有工作线程已经销毁，任务缓存队列已经清空或执行结束后，线程池被设置为TERMINATED状态。

2. 线程池中的线程初始化

默认情况下，创建线程池之后，线程池中是没有线程的，需要提交任务之后才会创建线程。
在实际中如果需要线程池创建之后立即创建线程，可以通过以下两个方法办到：
prestartCoreThread()：初始化一个核心线程；
prestartAllCoreThreads()：初始化所有核心线程
下面是这2个方法的实现：

public boolean prestartCoreThread() {    return addIfUnderCorePoolSize(null); //注意传进去的参数是null}public int prestartAllCoreThreads() {    int n = 0;    while (addIfUnderCorePoolSize(null))//注意传进去的参数是null        ++n;    return n;}
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9

注意上面传进去的参数是null，根据源码的分析可知如果传进去的参数为null，则最后执行线程会阻塞在getTask方法中的r = workQueue.take();即等待任务队列中有任务。

3. 任务缓存队列及排队策略

在前面我们多次提到了任务缓存队列，即workQueue，它用来存放等待执行的任务。
workQueue的类型为BlockingQueue，通常可以取下面三种类型：
　　1）SynchronousQueue：直接提交策略，它不会保存提交的任务，而是将直接新建一个线程来执行新来的任务。它将任务直接提交给线程而不保持它们。如果不存在可用于立即运行任务的线程，则试图把任务加入队列将失败，因此会构造一个新的线程。
　　2）LinkedBlockingQueue：基于链表的先进先出队列，如果创建时没有指定此队列大小，则默认为Integer.MAX_VALUE；
　　3）ArrayBlockingQueue：基于数组的先进先出队列，此队列创建时必须指定大小，JDK不推荐使用；

4. 任务拒绝策略

当线程池的任务缓存队列已满并且线程池中的线程数目达到maximumPoolSize，如果还有任务到来就会采取任务拒绝策略，通常有以下四种策略：

  • ThreadPoolExecutor.AbortPolicy:丢弃任务并抛出
  • RejectedExecutionException异常。
  • ThreadPoolExecutor.DiscardPolicy：也是丢弃任务，但是不抛出异常。
  • ThreadPoolExecutor.DiscardOldestPolicy：丢弃队列最前面的任务，然后重新尝试执行任务（重复此过程）
  • ThreadPoolExecutor.CallerRunsPolicy：由调用线程处理该任务

5. 线程池容量的动态调整

ThreadPoolExecutor提供了动态调整线程池容量大小的方法：setCorePoolSize()和setMaximumPoolSize()，
setCorePoolSize：设置核心池大小
setMaximumPoolSize：设置线程池最大能创建的线程数目大小
当上述参数从小变大时，ThreadPoolExecutor进行线程赋值，还可能立即创建新的线程来执行任务。

五.使用示例

创建线程池可以使用ThreadPoolExecutor executor = new ThreadPoolExecutor(size, maxSize, 1, TimeUnit.DAYS, queue);这种new的方式创建，要配置一个线程池是比较复杂的，尤其是对于线程池的原理不是很清楚的情况下，很有可能配置的线程池不是较优的，因此在Executors类里面提供了一些静态工厂，生成一些常用的线程池。

  • newSingleThreadExecutor
    创建一个单线程的线程池。这个线程池只有一个线程在工作，也就是相当于单线程串行执行所有任务。如果这个唯一的线程因为异常结束，那么会有一个新的线程来替代它。此线程池保证所有任务的执行顺序按照任务的提交顺序执行。
  • newFixedThreadPool
    创建固定大小的线程池。每次提交一个任务就创建一个线程，直到线程达到线程池的最大大小。线程池的大小一旦达到最大值就会保持不变，如果某个线程因为执行异常而结束，那么线程池会补充一个新线程。
    固定大小线程池默认使用无限制的链表。注意，这可能引起资源耗尽问题，但只要线程处理的速度大于队列增长的速度就不会发生。
  • newCachedThreadPool
    创建一个可缓存的线程池。如果线程池的大小超过了处理任务所需要的线程，那么就会回收部分空闲（60秒不执行任务）的线程，当任务数增加时，此线程池又可以智能的添加新线程来处理任务。此线程池不会对线程池大小做限制，线程池大小完全依赖于操作系统（或者说JVM）能够创建的最大线程大小。
  • newScheduledThreadPool
    创建一个大小无限的线程池。此线程池支持定时以及周期性执行任务的需求。
//Executeors中的实现public static ExecutorService newCachedThreadPool() {        return new ThreadPoolExecutor(0, Integer.MAX_VALUE,                                      60L, TimeUnit.SECONDS,                                      new SynchronousQueue<Runnable>());    }
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6

在JDK帮助文档中，有如此一段话：“强烈建议程序员使用较为方便的Executors工厂方法Executors.newCachedThreadPool()（无界线程池，可以进行自动线程回收）、Executors.newFixedThreadPool(int)（固定大小线程池）、Executors.newSingleThreadExecutor()（单个后台线程），它们均为大多数使用场景预定义了设置。”

public class Test4{    public static void main(String[] args)    {        ExecutorService threadPool = Executors.newCachedThreadPool();         for(int i = 0; i< 100; i++)        {            WorkerThread thread = new WorkerThread(i+"");            threadPool.submit(thread);        }        threadPool.shutdown();    }}class WorkerThread implements Runnable {    private String command;    public WorkerThread(String s){        this.command=s;    }    @Override    public void run() {        System.out.println(Thread.currentThread().getName()+" Start. Command = "+command);        processCommand();        System.out.println(Thread.currentThread().getName()+" "+command + " End.");    }    private void processCommand() {        try {            Thread.sleep(1000);        } catch (InterruptedException e) {            e.printStackTrace();        }    }    @Override    public String toString(){        return this.command;    }}

Java线程池


联系
我们