Java多线程之线程池结构

Java多线程之线程池结构

 

线程池框架图

 

 

Executor

 

这是线程池的最顶层接口，我们一般不会用它，就只有一个方法void execute(Runnable command)。

 

 

 

ExecutorService

 

第二层接口，继承了Executor，如果是用工具类Executors创建的线程池，就可以用这种类型来修饰。

增加的方法如下：

public interface ExecutorService extends Executor {  
    void shutdown();  
    List<Runnable> shutdownNow();  
    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;
}  

 

 

 

Executors

 

这是线程池的工具类，内部实现就是通过ThreadPoolExecutor来实现

 

包含下面这几种线程池：

 

newCachedThreadPool

 

• 该线程池比较适合没有固定大小并且比较快速就能完成的小任务，它将为每个任务创建一个线程。
• 与每次都新建线程的区别，就是如果任务的执行时间很短，就能在一定时间内重用线程。

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

 

newFixedThreadPool

 

使用的Thread对象的数量是有限的,如果提交的任务数量大于限制的最大线程数，那么这些任务讲排队，然后当有一个线程的任务结束之后，将会根据调度策略继续等待执行下一个任务。

 

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

 

 

newSingleThreadExecutor

 

线程数量为1的FixedThreadPool,如果提交了多个任务，那么这些任务将会排队，每个任务都会在下一个任务开始之前运行结束，所有的任务将会使用相同的线程。

 

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

 

 

newScheduledThreadPool

 

创建一个固定长度的线程池，而且以延迟或定时的方式来执行任务。 

 

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

 

 

 

ThreadPoolExecutor

 

上面工具类的几种类型的线程池都是通过ThreadPoolExecutor实现的，下面了解下里面的配置

private final BlockingQueue<Runnable> workQueue;              // 阻塞队列  
private final ReentrantLock mainLock = new ReentrantLock();   // 互斥锁  
private final HashSet<Worker> workers = new HashSet<Worker>();// 线程集合.一个Worker对应一个线程  
private final Condition termination = mainLock.newCondition();// 终止条件  
private int largestPoolSize;           // 线程池中线程数量曾经达到过的最大值。  
private long completedTaskCount;       // 已完成任务数量  
private volatile ThreadFactory threadFactory;     // ThreadFactory对象，用于创建线程。  
private volatile RejectedExecutionHandler handler;// 拒绝策略的处理句柄  
private volatile long keepAliveTime;   // 线程池维护线程所允许的空闲时间  
private volatile boolean allowCoreThreadTimeOut;  
private volatile int corePoolSize;     // 线程池维护线程的最小数量，哪怕是空闲的  
private volatile int maximumPoolSize;  // 线程池维护的最大线程数量  

 

注意：如果用ThreadPoolExecutor创建线程池时，没有定义拒绝策略，那默认的拒绝策略是

AbortPolicy（不执行此任务，而且直接抛出一个运行时异常！）

 

 

corePoolSize（核心线程数）与maximumPoolSize（最大线程数） 

 

1. 如果运行的线程少于 corePoolSize，则创建新线程来处理请求，即使其他辅助线程是空闲的；
2. 如果设置的corePoolSize 和 maximumPoolSize相同，则创建的线程池是大小固定的，如果运行的线程与corePoolSize相同，当有新请求过来时，若workQueue未满，则将请求放入workQueue中，等待有空闲的线程去从workQueue中取任务并处理 
3. 如果运行的线程多于 corePoolSize 而少于 maximumPoolSize，则仅当队列满时才创建新线程才创建新的线程去处理请求； 
4. 如果运行的线程多于corePoolSize 并且等于maximumPoolSize，若队列已经满了，则通过handler所指定的策略来处理新请求；
5. 如果将 maximumPoolSize 设置为基本的无界值（如 Integer.MAX_VALUE），则允许池适应任意数量的并发任务 

也就是说，处理任务的优先级为：

 

1. 核心线程corePoolSize > 任务队列workQueue > 最大线程maximumPoolSize，如果三者都满了，使用handler处理被拒绝的任务。
2. 当池中的线程数大于corePoolSize的时候，多余的线程会等待keepAliveTime长的时间，如果无请求可处理就自行销毁。 

 

workQueue（工作队列）

 

workQueue的类型为BlockingQueue；线程池所使用的缓冲队列，该缓冲队列的长度决定了能够缓冲的最大数量，缓冲队列有三种通用策略：

 

1. ArrayBlockingQueue：基于数组的FIFO队列，是有界的，创建时必须指定大小
2. LinkedBlockingQueue： 基于链表的FIFO队列，是无界的，默认大小是 Integer.MAX_VALUE
3. synchronousQueue: 一个比较特殊的队列，虽然它是无界的，但它不会保存任务，每一个新增任务的线程必须等待另一个线程取出任务，也可以把它看成容量为0的队列；也就是说当任务来时，核心线程数已经满了，就不会看缓冲队列，而是直接看maxPoolSize。

 

ThreadFactory 

 

使用 ThreadFactory 创建新线程。如果没有另外说明，则在同一个 ThreadGroup 中一律使用 Executors.defaultThreadFactory() 创建线程，并且这些线程具有相同的 NORM_PRIORITY 优先级和非守护进程状态。通过提供不同的 ThreadFactory，可以改变线程的名称、线程组、优先级、守护进程状态等等。如果从 newThread 返回 null 时 ThreadFactory 未能创建线程，则执行程序将继续运行，但不能执行任何任务。

 

DefaultThreadFactory类中实现了ThreadFactory接口并对其中定义的方法进行了实现

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) {  
        // 线程对应的任务是Runnable对象r  
        Thread t = new Thread(group, r,namePrefix + threadNumber.getAndIncrement(), 0);  
        // 设为非守护线程  
        if (t.isDaemon())  
            t.setDaemon(false);  
        // 将优先级设为Thread.NORM_PRIORITY  
        if (t.getPriority() != Thread.NORM_PRIORITY)  
            t.setPriority(Thread.NORM_PRIORITY);  
        return t;  
    }  
} 

 

 

ThreadGroup

 

对线程进行分组管理，为了方便统一管理，线程组可以进行复制，快速定位到一个线程，统一进行配置

 

 

ThreadGroup g1= new ThreadGroup("group 1");
Thread t1 = new Thread(g1, myThread);

 

 

 

RejectedExecutionHandler 

 

这是线程池的拒绝策略，当缓冲队列已经满了，并且最大线程数已满，再收到请求就会启动拒绝策略。

 

1. 在默认的 ThreadPoolExecutor.AbortPolicy 处理程序遭到拒绝将抛出运行时 RejectedExecutionException; 
2. 在 ThreadPoolExecutor.CallerRunsPolicy 线程调用运行该任务的 execute 本身。此策略提供简单的反馈控制机制，能够减缓新任务的提交速度 
3. 在 ThreadPoolExecutor.DiscardPolicy 不能执行的任务将被删除; 
4. 在 ThreadPoolExecutor.DiscardOldestPolicy 如果执行程序尚未关闭，则位于工作队列头部的任务将被删除，然后重试执行程序（如果再次失败，则重复此过程）

 

 

拒绝策略的几个示例

 

AbortPolicy（抛异常）

public class AbortPolicyDemo {  

    private static final int THREADS_SIZE = 1;  
    private static final int CAPACITY = 1;  

    public static void main(String[] args) throws Exception {  

        // 创建线程池。线程池的"最大池大小"和"核心池大小"都为1(THREADS_SIZE)，"线程池"的阻塞队列容量为1(CAPACITY)。  
        ThreadPoolExecutor pool = new ThreadPoolExecutor(THREADS_SIZE, THREADS_SIZE, 0, TimeUnit.SECONDS,  
                new ArrayBlockingQueue<Runnable>(CAPACITY));  
        // 设置线程池的拒绝策略为"抛出异常"  
        pool.setRejectedExecutionHandler(new ThreadPoolExecutor.AbortPolicy());  

        try {  

            // 新建10个任务，并将它们添加到线程池中。  
            for (int i = 0; i < 10; i++) {  
                Runnable myrun = new MyRunnable("task-"+i);  
                pool.execute(myrun);  
            }  
        } catch (RejectedExecutionException e) {  
            e.printStackTrace();  
            // 关闭线程池  
            pool.shutdown();  
        }  
    }  
}  

 

CallerRunsPolicy（让调用线程自己执行）

public class CallerRunsPolicyDemo {  

    private static final int THREADS_SIZE = 1;  
    private static final int CAPACITY = 1;  

    public static void main(String[] args) throws Exception {  

        // 创建线程池。线程池的"最大池大小"和"核心池大小"都为1(THREADS_SIZE)，"线程池"的阻塞队列容量为1(CAPACITY)。  
        ThreadPoolExecutor pool = new ThreadPoolExecutor(THREADS_SIZE, THREADS_SIZE, 0, TimeUnit.SECONDS,  
                new ArrayBlockingQueue<Runnable>(CAPACITY));  
        // 设置线程池的拒绝策略为"CallerRunsPolicy"  
        pool.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());  

        // 新建10个任务，并将它们添加到线程池中。  
        for (int i = 0; i < 10; i++) {  
            Runnable myrun = new MyRunnable("task-"+i);  
            pool.execute(myrun);  
        }  

        // 关闭线程池  
        pool.shutdown();  
    }  
} 

 

DiscardPolicy（丢弃任务）

public class DiscardPolicyDemo {  

    private static final int THREADS_SIZE = 1;  
    private static final int CAPACITY = 1;  

    public static void main(String[] args) throws Exception {  

        // 创建线程池。线程池的"最大池大小"和"核心池大小"都为1(THREADS_SIZE)，"线程池"的阻塞队列容量为1(CAPACITY)。  
        ThreadPoolExecutor pool = new ThreadPoolExecutor(THREADS_SIZE, THREADS_SIZE, 0, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(CAPACITY));  
        // 设置线程池的拒绝策略为"丢弃"  
        pool.setRejectedExecutionHandler(new ThreadPoolExecutor.DiscardPolicy());  

        // 新建10个任务，并将它们添加到线程池中。  
        for (int i = 0; i < 10; i++) {  
            Runnable myrun = new MyRunnable("task-"+i);  
            pool.execute(myrun);  
        }  
        // 关闭线程池  
        pool.shutdown();  
    }  
} 

 

DiscardOldestPolicy（丢弃最老的一个任务）

public class DiscardOldestPolicyDemo {  

    private static final int THREADS_SIZE = 1;  
    private static final int CAPACITY = 1;  

    public static void main(String[] args) throws Exception {  

        // 创建线程池。线程池的"最大池大小"和"核心池大小"都为1(THREADS_SIZE)，"线程池"的阻塞队列容量为1(CAPACITY)。  
        ThreadPoolExecutor pool = new ThreadPoolExecutor(THREADS_SIZE, THREADS_SIZE, 0, TimeUnit.SECONDS,  
                new ArrayBlockingQueue<Runnable>(CAPACITY));  
        // 设置线程池的拒绝策略为"DiscardOldestPolicy"  
        pool.setRejectedExecutionHandler(new ThreadPoolExecutor.DiscardOldestPolicy());  

        // 新建10个任务，并将它们添加到线程池中。  
        for (int i = 0; i < 10; i++) {  
            Runnable myrun = new MyRunnable("task-"+i);  
            pool.execute(myrun);  
        }  
        // 关闭线程池  
        pool.shutdown();  
    }  
}