Java中的线程池

1.简介

 

使用线程池可以避免线程的频繁创建以及销毁。

java中提供的用于实现线程池的api：

 

 

 

executor、executorservice、abstractexecutorservice、threadpoolexecutor、forkjoinpool都位于java.util.concurrent包下。

*threadpoolexecutor、forkjoinpool为线程池的实现类。 

 

 

2.executor

 

public interface executor {

    /**
     * 向线程池提交一个任务,交由线程池去执行
     */
    void execute(runnable command);

}

 

*该接口声明了execute(runnable command)方法，负责向线程池中提交一个任务。

 

 

3.executorservice接口

 

public interface executorservice extends executor {

    /**
     * 关闭线程池(直到队列中的任务被执行完才会进行关闭)
     */
    void shutdown();

    /**
     * 立刻关闭线程池(不执行队列中的任务，且尝试中断当前执行的任务)
     */
    list<runnable> shutdownnow();

    /**
     * 判断线程池是否处于shutdown状态.
     */
    boolean isshutdown();

    /**
     * 判断线程池是否处于terminated状态.
     */
    boolean isterminated();

    /**
     * 若在指定时间内线程池处于terminated状态则立即返回true,否则超过时间后仍未为terminated状态则返回false.
     */
    boolean awaittermination(long timeout, timeunit unit) throws interruptedexception;

    /**
     * 向线程池提交一个任务并返回包含指定类型的future(根据callable的泛型)
     */
    <t> future<t> submit(callable<t> task);

    /**
     * 向线程池提交一个任务并指定任务执行结果的类型,返回包含指定类型的future.
     */
    <t> future<t> submit(runnable task, t result);

    /**
     * 向线程池提交一个任务并返回未知类型的future.
     */
    future<?> submit(runnable task);

    /**
     * 向线程池提交多个任务并返回指定类型的future列表.
     */
    <t> list<future<t>> invokeall(collection<? extends callable<t>> tasks) throws interruptedexception;

    /**
     * 向线程池提交多个任务并返回指定类型的future列表,如果在指定时间内没有执行完毕则直接返回.
     */
    <t> list<future<t>> invokeall(collection<? extends callable<t>> tasks, long timeout, timeunit unit) throws interruptedexception;
     
    /**
     * 向线程池提交多个任务,当任意一个任务执行完毕后返回指定类型的future.
     */
    <t> t invokeany(collection<? extends callable<t>> tasks) throws interruptedexception, executionexception;

    /**
     * 向线程池提交多个任务,在指定时间内,当任意一个任务执行完毕后返回指定类型的future,若超时则抛出异常.
     */
    <t> t invokeany(collection<? extends callable<t>> tasks, long timeout, timeunit unit) throws interruptedexception, executionexception, timeoutexception;
}

 

public interface future<v> {

    /**
     * 中断任务的执行
     */
    boolean cancel(boolean mayinterruptifrunning);

    /**
     * 判断任务是否中断成功
     */
    boolean iscancelled();

    /**
     * 判断任务是否执行完成
     */
    boolean isdone();
    
    /**
     * 获取任务的执行结果直到任务执行完毕(阻塞线程)
     */
    v get() throws interruptedexception, executionexception;
  
    /**
     * 获取任务的执行结果,若在指定时间内任务仍然没有执行完毕则抛出timeoutexception
     */
    v get(long timeout, timeunit unit) throws interruptedexception, executionexception, timeoutexception;
}

 

*execute()方法不能获取任务的执行结果，而submit()方法能够根据返回的future实例获取任务的执行结果。

 

 

4.threadpoolexecutor

 

 

threadpoolexecutor声明的核心属性

 

corepoolsize：线程池中核心线程的数量。

maximumpoolsize：线程池中最大线程数。

keepalivetime：线程的空闲时间。

unit：修饰线程空闲时间的单位。

workqueue：任务队列。

threadfactory：线程工厂，用于创建线程。

handler：当队列已满且当前线程数已达到所允许的最大值时的处理策略。

*线程池中的线程包括核心线程以及普通线程，核心线程一旦创建后直到线程池被关闭前都就不会销毁，而普通线程会因为到达空闲时间而被销毁。

 

构造方法：

public threadpoolexecutor(int corepoolsize,
                          int maximumpoolsize,
                          long keepalivetime,
                          timeunit unit,
                          blockingqueue<runnable> workqueue,
                          threadfactory threadfactory,
                          rejectedexecutionhandler handler)

 

 

blockingqueue的类型 

 

blockingqueue提供了arrayblockingqueue、linkedblockingqueue、synchronousqueue等实现类。

1.arrayblockingqueue：使用顺序表的结构进行存储，在使用时需要指定其长度，支持公平锁/非公平锁进行操作。

2.linkedblockingqueue：使用链表的结构进行存储，在使用时不需要指定其长度，队列的长度为integer.max_value。

3.synchronousqueue：一个不存储元素的队列，每一个put操作必须等待take操作，否则不能添加元素，支持公平锁和非公平锁。

 

*这些实现类在进行入队和出队操作时都会进行加锁，以保证在并发访问时数据的完整性。

 

 

队列已满且线程数已达到所允许的最大值时的处理策略

 

rejectedexecutionhandler提供了abortpolicy、discardpolicy、discardolderstpolicy、callerunspolicy四个策略，这四个策略都是threadpoolexecutor的静态内部类。

1.abortpolicy：放弃任务并抛出rejectedexecutionexception异常。

2.discardpolicy：放弃任务但不抛出异常。

3.discardolderstpolicy： 放弃队头中的任务，然后重新尝试执行新任务。

4.callerunspolicy： 由调用线程来处理该任务。

 

 

线程池的状态

 

private static final int running    = -1；
private static final int shutdown   =  0； 
private static final int stop       =  1；
private static final int tidying    =  2； 
private static final int terminated =  3；

 

1.runing：线程池处于运行状态，此时可以接受新的任务请求，并且执行队列中的任务。

2.shutdown：线程池处于关闭状态，此时不接受新的任务请求，但会继续执行队列中的任务。

3.stop：线程池处于禁用状态，此时不接受新的任务请求，并且不执行队列中的任务。

4.tidying：线程池处于整理状态，此时没有正在执行的任务

5.terminated ：线程池处于终止状态。

 

线程池状态变化过程

1.当线程池创建后处于running状态。

2.1 若此时调用了shutdown()方法，那么线程池将处于shutdown状态，不接受新的任务请求，但会继续执行队列中的任务，当队列中的任务为空且没有正在执行的任务时，线程池的状态为tidying。

2.2 若此时调用了shutdownnow()方法，那么线程池将处于stop状态，不接受新的任务请求且不执行队列中的任务，此时线程池的状态为tidying。

3.当线程池的状态为tidying时，当terminated()方法处理完毕后，线程池将处于trrminated状态。

 

 

任务的执行流程

 

1.当调用了execute()或者submit()方法向线程池提交任务后，首先判断当前线程池中的线程个数是否大于核心线程数。

2.如果当前线程池的线程个数小于核心线程数，则创建一个核心线程来处理任务。

3.如果当前线程池的线程个数大于核心线程数，则将任务放入到队列中，如果放入队列成功，那么该任务将等待被空闲的线程处理，如果放入队列失败(队满)，则判断当前线程池中的线程个数是否达到所允许的线程最大值，若没达到则创建一个普通线程去处理任务，否则根据预定义的处理策略去进行处理。

 

 

 

 

 

 

5.executors工具类

 

java中提供了executors工具类，用于直接创建executor。

 

cachethreadpool

public static executorservice newcachedthreadpool() {
    return new threadpoolexecutor(0, integer.max_value,60l, timeunit.seconds,new synchronousqueue<runnable>());
}

cachethreadpool创建的都是普通线程(其核心线程数为0)、线程池最大的线程数为integer.max_value、线程的空闲时间为60秒，此方式适合大量耗时短的任务、不适合大量耗时长的任务。

*由于创建的都是普通线程，且空闲时间为60秒，则仍有可能会频繁的创建线程。

 

fixedthreadpool

public static executorservice newfixedthreadpool(int nthreads) {
    return new threadpoolexecutor(nthreads, nthreads,0l, timeunit.milliseconds,new linkedblockingqueue<runnable>());
}

fixedthreadpool创建的都是核心线程，其线程个数由构建方法的入参决定，线程不会因为空闲时间而被销毁，适合预知任务数量的业务。

 

singlethreadexector

public static executorservice newsinglethreadexecutor() {
    return new finalizabledelegatedexecutorservice(new threadpoolexecutor(1, 1,new linkedblockingqueue<runnable>()));
}

singlethreadexector使用一个核心线程来处理任务。

 

scheduledthreadpool

public static scheduledexecutorservice newscheduledthreadpool(int corepoolsize) {
        return new scheduledthreadpoolexecutor(corepoolsize);
}

*scheduledthreadpool支持定时执行任务以及固定间隔执行任务。

 

singlethreadscheduledexecutor

public static scheduledexecutorservice newsinglethreadscheduledexecutor() {
        return new delegatedscheduledexecutorservice(new scheduledthreadpoolexecutor(1));
}

*singlethreadscheduledexecutor支持一个线程的定时执行任务以及固定间隔执行任务。

 

public interface scheduledexecutorservice extends executorservice {

    /**
     * 在指定的延迟时间到达后执行任务一次
     */
    public scheduledfuture<?> schedule(runnable command,long delay, timeunit unit);

    /**
     * 在指定的延迟时间到达后执行任务一次
     */
    public <v> scheduledfuture<v> schedule(callable<v> callable,long delay, timeunit unit);

    /**
     * 在指定的初始化延迟时间到达后执行任务一次,往后每隔period时间执行任务一次.
     */
    public scheduledfuture<?> scheduleatfixedrate(runnable command,long initialdelay,long period,timeunit unit);

    /**
     * 在指定的初始化延迟时间到达后执行任务一次,往后每次任务执行完毕后相隔delay时间执行任务一次.
     */
    public scheduledfuture<?> schedulewithfixeddelay(runnable command,long initialdelay,long delay,timeunit unit);

}

 

 workstealingpool

public static executorservice newworkstealingpool(int parallelism) {
        return new forkjoinpool(parallelism,forkjoinpool.defaultforkjoinworkerthreadfactory,null, true);
}

 

workstealingpool创建一个并行级别的线程池，同一时刻最多只能有指定个数个线程正在执行任务，创建时直接指定最多允许并行执行的线程个数即可(如果不传则使用cpu的核数)

newworkstealingpool方法内部返回一个forkjoinpool实例，forkjoinpool是java7新提供的线程池，同样继承abstactexecutorservice。