线程的创建与线程池ThreadPoolExecutor，Executors

本文主要是介绍线程的创建与线程池ThreadPoolExecutor，Executors，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

线程的创建与线程池及线程池工具类

1.线程的创建方式

1.1继承Thread类重写run方法

public class Test {public static void main(String[] args) {Thread thread = new MyThread();thread.setName("mythread001");thread.start();  }}
class MyThread extends Thread{@Overridepublic void run() {// TODO Auto-generated method stubSystem.out.println(Thread.currentThread().getName() + ",running ....");}
}

1.2实现Runnable接口

public class Test {public static void main(String[] args) {Thread thread = new Thread(new MyRunnable());thread.setName("myRunnable001");thread.start();  }
}
class MyRunnable implements Runnable{@Overridepublic void run() {System.out.println("my runnable...");}
}

其实以上两种方式本质上是一样的，都是重写了接口Runnable的run方法，因为Thread类实现了Runnable接口。Runnable接口是JDK1.0提供的，在java.lang包下，但是Runnable接口的run方法有的场景下并不合适，比如不能返回运算结果和抛出检查异常，这时我们可以使用JDK1.5提供的在java.util.concurrent包下的Callable接口：

@FunctionalInterface
public interface Callable<V> {/*** Computes a result, or throws an exception if unable to do so.** @return computed result* @throws Exception if unable to compute a result*/V call() throws Exception;
}

1.3Callable接口类型的线程创建

因为该接口可以返回一个结果，所以我们需要获取返回结果，这时系统提供了一个Future接口用于获取返回的结果，但是该接口代表的是一个异步计算的结果，所以我们需要等到计算完成时才能获取结果，否则就会一直阻塞直到结果返回。Future接口内的方法如下：

 /*** Attempts to cancel execution of this task.  This attempt will* fail if the task has already completed, has already been cancelled,* or could not be cancelled for some other reason. If successful,* and this task has not started when {@code cancel} is called,* this task should never run.  If the task has already started,* then the {@code mayInterruptIfRunning} parameter determines* whether the thread executing this task should be interrupted in* an attempt to stop the task.** <p>After this method returns, subsequent calls to {@link #isDone} will* always return {@code true}.  Subsequent calls to {@link #isCancelled}* will always return {@code true} if this method returned {@code true}.** @param mayInterruptIfRunning {@code true} if the thread executing this* task should be interrupted; otherwise, in-progress tasks are allowed* to complete* @return {@code false} if the task could not be cancelled,* typically because it has already completed normally;* {@code true} otherwise*/boolean cancel(boolean mayInterruptIfRunning);/*** Returns {@code true} if this task was cancelled before it completed* normally.** @return {@code true} if this task was cancelled before it completed*/boolean isCancelled();/*** Returns {@code true} if this task completed.** Completion may be due to normal termination, an exception, or* cancellation -- in all of these cases, this method will return* {@code true}.** @return {@code true} if this task completed*/boolean isDone();/*** Waits if necessary for the computation to complete, and then* retrieves its result.** @return the computed result* @throws CancellationException if the computation was cancelled* @throws ExecutionException if the computation threw an* exception* @throws InterruptedException if the current thread was interrupted* while waiting*/V get() throws InterruptedException, ExecutionException;/*** Waits if necessary for at most the given time for the computation* to complete, and then retrieves its result, if available.** @param timeout the maximum time to wait* @param unit the time unit of the timeout argument* @return the computed result* @throws CancellationException if the computation was cancelled* @throws ExecutionException if the computation threw an* exception* @throws InterruptedException if the current thread was interrupted* while waiting* @throws TimeoutException if the wait timed out*/V get(long timeout, TimeUnit unit)throws InterruptedException, ExecutionException, TimeoutException;

所以我们获取结果要在调用isDone()方法返回true时才能调用get()方法。该接口有一个系统提供的实现类FutureTask，FutureTask类实现了RunnableFuture<V>，RunnableFuture<V>接口继承了Runnable, Future<V>两个接口。FutureTask中实现的Runnable中的run方法调用了Callable接口中的call方法。所以创建Callable接口的线程的方法与Runnable类似：

public class Test {public static void main(String[] args) throws InterruptedException, ExecutionException {FutureTask<String> future = new FutureTask<String>(new Callable<String>() {@Overridepublic String call() throws Exception {return "hello";}});Thread thread = new Thread(future);thread.setName("myCallable001");thread.start();while (!future.isDone()) {try {System.out.println("myCallable001 is not done");//等待线程运行结束Thread.currentThread().sleep(1000);} catch (InterruptedException e) {// TODO Auto-generated catch blocke.printStackTrace();}}String result = future.get();System.out.println("result : " + result);}
}

运行结果如图：

2.线程池

线程的创建我们已经知道了，但是需要让线程启动运行我们需要调用start()方法，该方法会为线程准备必要的系统资源，比如线程的栈内存，程序计数器和本地栈等，所以一个线程是需要消耗系统资源的，也是耗时的。为了降低系统资源的消耗和提高系统的运行效率我们需要对线程复用，就像数据库连接池一样，使用线程池技术，来达到这个目的。JDK为了我们提供了ThreadPoolExecutor类，接下来我们学习下这个线程池类：

2.1Executor

该接口源码如下：

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 {@code Executor} 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);
}

该接口声明了一个方法，void execute(Runnable command);用来执行一个线程。

2.2ExecutorService

该接口继承自Executor，并声明了以下方法：

该接口种的方法更加实用，shutdown()方法与shutdownNow()方法用于停掉线程池，源码中写了如下的例子：

首先调用shutdown()阻止新的任务提交到线程池，调用awaitTermination等待正在执行的线程执行完毕，或者超时或者被打断执行，然后再调用shutdownNow()取消当前正在执行的任务。其他的方法见名知意。

2.3AbstractExecutorService

该抽象类实现了ExecutorService接口，其方法如图：

这些方法中要说的就是<T> RunnableFuture<T> newTaskFor(Runnable runnable, T value)，该方法将Runnable接口的任务转换为Callable相关的FutureTask<T>。其代码实现如图：

在FutureTask<T>中的构造方法源码如下：

再到Executors.callable(runnable, result)查看源码如下：

2.4ThreadPoolExecutor

构造函数

线程池实现类，我们先看该类的核心构造函数：

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;}

参数：corePoolSize。the number of threads to keep in the pool, even if they are idle, unless {@code allowCoreThreadTimeOut} is set。线程池中保持的线程数量，尽管这些线程是空闲的，除非设置allowCoreThreadTimeOut参数为true，则在空闲时间超过keepAliveTime时，会被终止掉。allowCoreThreadTimeOut默认为false。

参数：maximumPoolSize。线程池中允许的最大线程数量。

参数：keepAliveTime。保持活跃的时间，也就是当线程池中的线程数量超过corePoolSize时，这些超量的线程在等待被新任务使用前的最大等待时间，超过找个时间就要被终止掉了。

参数：unit。保持活跃时间的单位。可选为：NANOSECONDS，MICROSECONDS，MILLISECONDS，SECONDS，MINUTES，HOURS，DAYS等。

参数：workQueue。工作队列。这队列用来保持那些execute()方法提交的还没有执行的任务。常用的队列有SynchronousQueue,LinkedBlockingQueue,ArrayBlockingQueue。一般我们需要根据自己的实际业务需求选择合适的工作队列。

SynchronousQueue：直接传递。对于一个好的默认的工作队列选择是SynchronousQueue，该队列传递任务到线程而不持有它们。在这一点上，试图向该队列压入一个任务，如果没有可用的线程立刻运行任务，那么就会入列失败，所以一个新的线程就会被创建。当处理那些内部依赖的任务集合时，这个选择可以避免锁住。直接接传递通常需要无边界的最大线程数来避免新提交任务被拒绝处理。当任务以平均快于被处理的速度提交到线程池时，它依次地确认无边界线程增长的可能性；

LinkedBlockingQueue：无界队列。没有预先定义容量的无界队列，在核心线程数都繁忙的时候会使新提交的任务在队列中等待被执行，所以将不会创建更多的线程，因此，最大线程数的值将不起作用。当每个任务之间是相互独立的时比较适合该队列，所以任务之间不能互相影响执行。例如，在一个WEB页面服务器，当平滑的出现短暂的请求爆发时这个类型的队列是非常有用的，当任务以快于平均处理速度被提交时该队列会确认无边界队列增长的可能性。

ArrayBlockingQueue：有界阻塞队列，遵循FIFO原则，一旦创建容量不能改变，当向一个已经满了的该队列中添加元素和向一个已经为空的该队列取出元素都会导致阻塞；当线程池使用有限的最大线程数时该队列可以帮助保护资源枯竭，但它更难协调和控制。队列大小和最大线程数在性能上可以互相交换：使用大队列和小线程池会降低CPU使用和OS资源与上下文切换开销，但会导致人为降低吞吐量，如果任务频繁阻塞，系统的线程调度时间会超过我们的允许值；如果使用小队列大池，这将会使CPU较为繁忙但会出现难以接受的调度开销，这也会导致降低吞吐量。

参数：threadFactory。线程工厂。当线程池需要创建线程的时候用来创建线程。默认是Executors类的静态内部类DefaultThreadFactory。源码如下：

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;}}

主要是设置了一些守护性和优先级等属性。

参数：handler。拒绝执行处理器。当线程的数量已经到了边界值，并且workQueue中任务也达到最大值，此时需要使用处理器处理多余的任务。该参数有四个值，分别是CallerRunsPolicy，AbortPolicy，DiscardPolicy，DiscardOldestPolicy。这四个处理器是线程池的内部类，都实现了RejectedExecutionHandler接口。

CallerRunsPolicy:在调用execut方法的调用线程中直接执行线程池拒绝的任务；

AbortPolicy：以抛出一个RejectedExecutionException的方式处理拒绝执行的任务；

DiscardPolicy：以直接忽略的方式处理拒绝执行的任务；

DiscardOldestPolicy：忽略掉最老的没有处理的拒绝任务，然后继续尝试执行execute方法，知道线程池关闭，那么放弃找个任务。

线程的管理

首先看看该类中的注释中是如何解释线程池怎么管理线程的：

When a new task is submitted in method {@link #execute(Runnable)},* and fewer than corePoolSize threads are running, a new thread is* created to handle the request, even if other worker threads are* idle.  If there are more than corePoolSize but less than* maximumPoolSize threads running, a new thread will be created only* if the queue is full.  By setting corePoolSize and maximumPoolSize* the same, you create a fixed-size thread pool. By setting* maximumPoolSize to an essentially unbounded value such as {@code* Integer.MAX_VALUE}, you allow the pool to accommodate an arbitrary* number of concurrent tasks. Most typically, core and maximum pool* sizes are set only upon construction, but they may also be changed* dynamically using {@link #setCorePoolSize} and {@link* #setMaximumPoolSize}.

当一个任务通过execute(Runnable)方法被提交时，并且有少于核心线程数的线程正在运行，那么一个新的线程会被创建去处理请求，即使其它的线程正在空闲；如果有多余核心线程数但小于最大线程数的线程在运行，如果任务队列没有满，则将任务压入任务队列等待执行完任务的线程去处理，如果满了则创建新的线程处理该任务；如果正在运行的线程数大于最大线程数，则需要根据拒绝执行处理器的不同进行处理。通过将核心线程数与最大线程数设置为相同的值，你可以创建一个固定大小的线程池。通过设置最大线程数为一个基本上是无边界的值例如Integer.MAX_VALUE,你允许线程池容纳任意数量的并发任务。通常，核心线程数和最大线程数是通过构造函数设置的，但是他们也可以通过使用setCorePoolSize和setMaximumPoolSize动态的改变。

验证如下：

新建一个线程池，代码如下：

public class Main {public static void main(String[] args) throws Exception {// 有界队列BlockingQueue<Runnable> workQueue = new ArrayBlockingQueue<>(5);// 放弃拒绝的任务并抛出异常RejectedExecutionHandler abortPolicyHandler = new ThreadPoolExecutor.AbortPolicy();RejectedExecutionHandler discardPolicyHandler = new ThreadPoolExecutor.DiscardPolicy();ThreadPoolExecutor threadPool = new ThreadPoolExecutor(5, 10, 30, TimeUnit.SECONDS, workQueue, discardPolicyHandler);long start = System.currentTimeMillis();for (int i = 0; i < 40; i++) {threadPool.execute(new MyTask());System.out.println("核心线程数" + threadPool.getCorePoolSize());System.out.println("最大线程数" + threadPool.getMaximumPoolSize());System.out.println("线程池数" + threadPool.getPoolSize());System.out.println("队列任务数" + threadPool.getQueue().size());System.out.println(">>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>");}System.out.println(System.currentTimeMillis()-start);threadPool.shutdown();if (threadPool.awaitTermination(6, TimeUnit.SECONDS)) {threadPool.shutdownNow();}}}class MyTask implements Runnable {@Overridepublic void run() {try {Thread.currentThread().sleep(5000);} catch (InterruptedException e) {// TODO Auto-generated catch blocke.printStackTrace();}System.out.println(Thread.currentThread().getName() + ", hello");}}

运行部分日志如下：

核心线程数5
最大线程数10
线程池数1
队列任务数0
>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>
核心线程数5
最大线程数10
线程池数2
队列任务数0
>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>
核心线程数5
最大线程数10
线程池数3
队列任务数0
>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>
核心线程数5
最大线程数10
线程池数4
队列任务数0
>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>
核心线程数5
最大线程数10
线程池数5
队列任务数0
>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>
核心线程数5
最大线程数10
线程池数5
队列任务数1
>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>
核心线程数5
最大线程数10
线程池数5
队列任务数2
>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>
核心线程数5
最大线程数10
线程池数5
队列任务数3
>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>
核心线程数5
最大线程数10
线程池数5
队列任务数4
>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>
核心线程数5
最大线程数10
线程池数5
队列任务数5
>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>
核心线程数5
最大线程数10
线程池数6
队列任务数5
>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>
核心线程数5
最大线程数10
线程池数7
队列任务数5
>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>
核心线程数5
最大线程数10
线程池数8
队列任务数5
>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>
核心线程数5
最大线程数10
线程池数9
队列任务数5
>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>
核心线程数5
最大线程数10
线程池数10
队列任务数5
>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>
核心线程数5
最大线程数10
线程池数10
队列任务数5
>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>
核心线程数5
最大线程数10
线程池数10
队列任务数5
>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>
核心线程数5
最大线程数10
线程池数10
队列任务数5
>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>
核心线程数5
最大线程数10
线程池数10
队列任务数5
>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>
核心线程数5
最大线程数10
线程池数10
队列任务数5
>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>
核心线程数5
最大线程数10
线程池数10
队列任务数5
>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>
核心线程数5
最大线程数10
线程池数10
队列任务数5
>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>
核心线程数5
最大线程数10
线程池数10
队列任务数5
>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>
核心线程数5
最大线程数10
线程池数10
队列任务数5
>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>
核心线程数5
最大线程数10
线程池数10
队列任务数5
>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>
核心线程数5
最大线程数10
线程池数10
队列任务数5
>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>
核心线程数5
最大线程数10
线程池数10
队列任务数5
>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>
核心线程数5
最大线程数10
线程池数10
队列任务数5
>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>
核心线程数5
最大线程数10
线程池数10
队列任务数5
>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>
核心线程数5
最大线程数10
线程池数10
队列任务数5
>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>
核心线程数5
最大线程数10
线程池数10
队列任务数5
>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>
核心线程数5
最大线程数10
线程池数10
队列任务数5
>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>
核心线程数5
最大线程数10
线程池数10
队列任务数5
>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>
核心线程数5
最大线程数10
线程池数10
队列任务数5
>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>
核心线程数5
最大线程数10
线程池数10
队列任务数5
>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>
核心线程数5
最大线程数10
线程池数10
队列任务数5
>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>
核心线程数5
最大线程数10
线程池数10
队列任务数5
>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>
核心线程数5
最大线程数10
线程池数10
队列任务数5
>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>
核心线程数5
最大线程数10
线程池数10
队列任务数5
>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>
核心线程数5
最大线程数10
线程池数10
队列任务数5
>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>

通过以上日志分析可以验证线程的创建过程是正确的。

初始化核心线程数

默认的核心线程只有在任务被提交的时候才会创建和启动，但是当我们的工作队列是一个非空队列的时候我们需要在线程池启动的时候就有核心线程可用，这时候我们可以使用prestartCoreThread()方法和prestartAllCoreThreads()方法。我们只需要在线程池被创建后根据自己的需求调用其中的方法即可。