java线程池的工作原理_Java 线程池的介绍以及工作原理

在什么情况下使用线程池？

1.单个任务处理的时间比较短

2.将需处理的任务的数量大

使用线程池的好处:

1. 降低资源消耗：　　　　　　通过重复利用已创建的线程降低线程创建和销毁造成的消耗。

2. 提高响应速度：　　　　　　当任务到达时，任务可以不需要等到线程创建就能立即执行。

3. 提高线程的可管理性：　　　线程是稀缺资源，如果无限制的创建。不仅仅会降低系统的稳定性，使用线程池可以统一分配，调优和监控。但是要做到合理的利用线程池。必须对于其实现原理了如指掌。

一个线程池包括以下四个基本组成部分：

1、线程池管理器(ThreadPool)：用于创建并管理线程池，包括 创建线程池，销毁线程池，添加新任务；

2、工作线程(PoolWorker)：线程池中线程，在没有任务时处于等待状态，可以循环的执行任务；

3、任务接口(Task)：每个任务必须实现的接口，以供工作线程调度任务的执行，它主要规定了任务的入口，任务执行完后的收尾工作，任务的执行状态等；

4、任务队列(taskQueue)：用于存放没有处理的任务。提供一种缓冲机制。

在JDK1.6中研究ThreadPoolExecutor类:

volatile intrunState;static final int RUNNING = 0;static final int SHUTDOWN = 1;static final int STOP = 2;static final int TERMINATED = 3;

runState表示当前线程池的状态，它是一个volatile变量用来保证线程之间的可见性；

当创建线程池后，初始时，线程池处于RUNNING状态；

如果调用了shutdown()方法，则线程池处于SHUTDOWN状态，此时线程池不能够接受新的任务，它会等待所有任务执行完毕；

如果调用了shutdownNow()方法，则线程池处于STOP状态，此时线程池不能接受新的任务，并且会去尝试终止正在执行的任务；

当线程池处于SHUTDOWN或STOP状态，并且所有工作线程已经销毁，任务缓存队列已经清空或执行结束后，线程池被设置为TERMINATED状态。

execute方法：

public voidexecute(Runnable command) {if (command == null)throw newNullPointerException();if (poolSize >= corePoolSize || !addIfUnderCorePoolSize(command)) {if (runState == RUNNING &&workQueue.offer(command)) {if (runState != RUNNING || poolSize == 0)

ensureQueuedTaskHandled(command);

}else if (!addIfUnderMaximumPoolSize(command))

reject(command);//is shutdown or saturated

}

}

addIfUnderCorePoolSize方法检查如果当前线程池的大小小于配置的核心线程数，说明还可以创建新线程，则启动新的线程执行这个任务。

private booleanaddIfUnderCorePoolSize(Runnable firstTask) {

Thread t= null;final ReentrantLock mainLock = this.mainLock;

mainLock.lock();try{if (poolSize < corePoolSize && runState ==RUNNING)

t=addThread(firstTask);

}finally{

mainLock.unlock();

}return t != null;

}

addThread:

privateThread addThread(Runnable firstTask) {

Worker w= newWorker(firstTask);

Thread t=threadFactory.newThread(w);boolean workerStarted = false;if (t != null) {if (t.isAlive()) //precheck that t is startable

throw newIllegalThreadStateException();

w.thread=t;

workers.add(w);int nt = ++poolSize;if (nt >largestPoolSize)

largestPoolSize=nt;try{

t.start();

workerStarted= true;

}finally{if (!workerStarted)

workers.remove(w);

}

}returnt;

}

Worker，在ThreadPoolExecutor中的内部类

private final class Worker implementsRunnable {/*** The runLock is acquired and released surrounding each task

* execution. It mainly protects against interrupts that are

* intended to cancel the worker thread from instead

* interrupting the task being run.*/

private final ReentrantLock runLock = newReentrantLock();/*** Initial task to run before entering run loop. Possibly null.*/

privateRunnable firstTask;/*** Per thread completed task counter; accumulated

* into completedTaskCount upon termination.*/

volatile longcompletedTasks;/*** Thread this worker is running in. Acts as a final field,

* but cannot be set until thread is created.*/Thread thread;/*** Records that the thread assigned to this worker has actually

* executed our run() method. Such threads are the only ones

* that will be interrupted.*/

volatile boolean hasRun = false;

Worker(Runnable firstTask) {this.firstTask =firstTask;

}booleanisActive() {returnrunLock.isLocked();

}/*** Interrupts thread if not running a task.*/

voidinterruptIfIdle() {final ReentrantLock runLock = this.runLock;if(runLock.tryLock()) {try{if (hasRun && thread !=Thread.currentThread())

thread.interrupt();

}finally{

runLock.unlock();

}

}

}/*** Interrupts thread even if running a task.*/

voidinterruptNow() {if(hasRun)

thread.interrupt();

}/*** Runs a single task between before/after methods.*/

private voidrunTask(Runnable task) {final ReentrantLock runLock = this.runLock;

runLock.lock();try{/** If pool is stopping ensure thread is interrupted;

* if not, ensure thread is not interrupted. This requires

* a double-check of state in case the interrupt was

* cleared concurrently with a shutdownNow -- if so,

* the interrupt is re-enabled.*/

if ((runState >= STOP ||(Thread.interrupted()&& runState >= STOP)) &&hasRun)

thread.interrupt();/** Track execution state to ensure that afterExecute

* is called only if task completed or threw

* exception. Otherwise, the caught runtime exception

* will have been thrown by afterExecute itself, in

* which case we don't want to call it again.*/

boolean ran = false;

beforeExecute(thread, task);try{

task.run();

ran= true;

afterExecute(task,null);++completedTasks;

}catch(RuntimeException ex) {if (!ran)

afterExecute(task, ex);throwex;

}

}finally{

runLock.unlock();

}

}/*** Main run loop*/

public voidrun() {try{

hasRun= true;

Runnable task=firstTask;

firstTask= null;while (task != null || (task = getTask()) != null) {

runTask(task);

task= null;

}

}finally{

workerDone(this);

}

}

}

View Code

ensureQueuedTaskHandled:

判断如果当前状态不是RUNING，则当前任务不加入到任务队列中，判断如果状态是停止，线程数小于允许的最大数，且任务队列还不空，则加入一个新的工作线程到线程池来帮助处理还未处理完的任务。

private voidensureQueuedTaskHandled(Runnable command) {final ReentrantLock mainLock = this.mainLock;

mainLock.lock();boolean reject = false;

Thread t= null;try{int state =runState;if (state != RUNNING &&workQueue.remove(command))

reject= true;else if (state < STOP &&poolSize< Math.max(corePoolSize, 1) &&

!workQueue.isEmpty())

t= addThread(null);

}finally{

mainLock.unlock();

}if(reject)

reject(command);

}

voidreject(Runnable command) {

handler.rejectedExecution(command,this);

}

addIfUnderMaximumPoolSize:

addIfUnderMaximumPoolSize检查如果线程池的大小小于配置的最大线程数，并且任务队列已经满了(就是execute方法试图把当前线程加入任务队列时不成功)，

说明现有线程已经不能支持当前的任务了，但线程池还有继续扩充的空间，就可以创建一个新的线程来处理提交的任务。

private booleanaddIfUnderMaximumPoolSize(Runnable firstTask) {

Thread t= null;final ReentrantLock mainLock = this.mainLock;

mainLock.lock();try{if (poolSize < maximumPoolSize && runState ==RUNNING)

t=addThread(firstTask);

}finally{

mainLock.unlock();

}return t != null;

}

整个流程：

1、如果线程池的当前大小还没有达到基本大小(poolSize < corePoolSize)，那么就新增加一个线程处理新提交的任务；

2、如果当前大小已经达到了基本大小，就将新提交的任务提交到阻塞队列排队，等候处理workQueue.offer(command)；

3、如果队列容量已达上限，并且当前大小poolSize没有达到maximumPoolSize，那么就新增线程来处理任务；

4、如果队列已满，并且当前线程数目也已经达到上限，那么意味着线程池的处理能力已经达到了极限，此时需要拒绝新增加的任务。至于如何拒绝处理新增的任务，取决于线程池的饱和策略RejectedExecutionHandler。

================================================

设置合适的线程池大小：

如果是CPU密集型的任务，那么良好的线程个数是实际CPU处理器的个数的1倍；

如果是I/O密集型的任务，那么良好的线程个数是实际CPU处理器个数的1.5倍到2倍

线程池中线程数量：

View Code

为什么+1,与CPU核数相等，表示满核运行,+1的话表示在CPU上存在竞争，两者的竞争力不一样。稍微高一点负荷是不影响的。

==================================================================================

Java中提供了几个Executors类的静态方法：

public static ExecutorService newFixedThreadPool(intnThreads) {return newThreadPoolExecutor(nThreads, nThreads,0L, TimeUnit.MILLISECONDS,new LinkedBlockingQueue());

}public staticExecutorService newSingleThreadExecutor() {return newFinalizableDelegatedExecutorService

(new ThreadPoolExecutor(1, 1,0L, TimeUnit.MILLISECONDS,new LinkedBlockingQueue()));

}public staticExecutorService newCachedThreadPool() {return new ThreadPoolExecutor(0, Integer.MAX_VALUE,60L, TimeUnit.SECONDS,new SynchronousQueue());

}

newFixedThreadPool创建的线程池corePoolSize和maximumPoolSize值是相等的，它使用的LinkedBlockingQueue；

newSingleThreadExecutor将corePoolSize和maximumPoolSize都设置为1，也使用的LinkedBlockingQueue；

newCachedThreadPool将corePoolSize设置为0，将maximumPoolSize设置为Integer.MAX_VALUE，使用的SynchronousQueue，也就是说来了任务就创建线程运行，当线程空闲超过60秒，就销毁线程。

任务拒绝策略：

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

ThreadPoolExecutor.AbortPolicy:丢弃任务并抛出RejectedExecutionException异常。

ThreadPoolExecutor.DiscardPolicy：也是丢弃任务，但是不抛出异常。

ThreadPoolExecutor.DiscardOldestPolicy：丢弃队列最前面的任务，然后重新尝试执行任务(重复此过程)

ThreadPoolExecutor.CallerRunsPolicy：由调用线程处理该任务

demo:

importjava.util.concurrent.ArrayBlockingQueue;importjava.util.concurrent.ThreadPoolExecutor;importjava.util.concurrent.TimeUnit;public classMain {public static voidmain(String[] args) {

ThreadPoolExecutor executor= new ThreadPoolExecutor(5, 10, 200, TimeUnit.MILLISECONDS,new ArrayBlockingQueue(5));for(int i=0;i<15;i++){

MyTask myTask= newMyTask(i);

executor.execute(myTask);

System.out.println("线程池中线程数目："+executor.getPoolSize()+"，队列中等待执行的任务数目："+executor.getQueue().size()+"，已执行玩别的任务数目："+executor.getCompletedTaskCount());

}

executor.shutdown();

}

}class MyTask implementsRunnable {private inttaskNum;public MyTask(intnum) {this.taskNum =num;

}

@Overridepublic voidrun() {

System.out.println("正在执行task "+taskNum);try{

Thread.currentThread().sleep(0);

}catch(InterruptedException e) {

e.printStackTrace();

}

System.out.println("task "+taskNum+"执行完毕");

}

}

线程池中线程数目：1，队列中等待执行的任务数目：0，已执行玩别的任务数目：0线程池中线程数目：2，队列中等待执行的任务数目：0，已执行玩别的任务数目：0线程池中线程数目：3，队列中等待执行的任务数目：0，已执行玩别的任务数目：0正在执行task0线程池中线程数目：4，队列中等待执行的任务数目：0，已执行玩别的任务数目：0正在执行task3正在执行task1task 3执行完毕

task 1执行完毕

线程池中线程数目：5，队列中等待执行的任务数目：0，已执行玩别的任务数目：0task 0执行完毕

正在执行task5线程池中线程数目：5，队列中等待执行的任务数目：1，已执行玩别的任务数目：2线程池中线程数目：5，队列中等待执行的任务数目：1，已执行玩别的任务数目：3线程池中线程数目：5，队列中等待执行的任务数目：2，已执行玩别的任务数目：3线程池中线程数目：5，队列中等待执行的任务数目：3，已执行玩别的任务数目：3线程池中线程数目：5，队列中等待执行的任务数目：4，已执行玩别的任务数目：3线程池中线程数目：5，队列中等待执行的任务数目：5，已执行玩别的任务数目：3task 5执行完毕

正在执行task6task 6执行完毕

正在执行task7task 7执行完毕

正在执行task8task 8执行完毕

正在执行task9task 9执行完毕

正在执行task10task 10执行完毕

线程池中线程数目：6，队列中等待执行的任务数目：0，已执行玩别的任务数目：9线程池中线程数目：6，队列中等待执行的任务数目：1，已执行玩别的任务数目：9线程池中线程数目：6，队列中等待执行的任务数目：2，已执行玩别的任务数目：9线程池中线程数目：6，队列中等待执行的任务数目：3，已执行玩别的任务数目：9正在执行task12正在执行task14正在执行task13task 14执行完毕

task 13执行完毕

task 12执行完毕

正在执行task2task 2执行完毕

正在执行task4task 4执行完毕

正在执行task11task 11执行完毕

View Code