java线程池的工作原理_Java 线程池的介绍以及工作原理
在什么情況下使用線程池?
1.單個任務(wù)處理的時間比較短
2.將需處理的任務(wù)的數(shù)量大
使用線程池的好處:
1. 降低資源消耗: 通過重復(fù)利用已創(chuàng)建的線程降低線程創(chuàng)建和銷毀造成的消耗。
2. 提高響應(yīng)速度: 當(dāng)任務(wù)到達(dá)時,任務(wù)可以不需要等到線程創(chuàng)建就能立即執(zhí)行。
3. 提高線程的可管理性: 線程是稀缺資源,如果無限制的創(chuàng)建。不僅僅會降低系統(tǒng)的穩(wěn)定性,使用線程池可以統(tǒng)一分配,調(diào)優(yōu)和監(jiān)控。但是要做到合理的利用線程池。必須對于其實現(xiàn)原理了如指掌。
一個線程池包括以下四個基本組成部分:
1、線程池管理器(ThreadPool):用于創(chuàng)建并管理線程池,包括 創(chuàng)建線程池,銷毀線程池,添加新任務(wù);
2、工作線程(PoolWorker):線程池中線程,在沒有任務(wù)時處于等待狀態(tài),可以循環(huán)的執(zhí)行任務(wù);
3、任務(wù)接口(Task):每個任務(wù)必須實現(xiàn)的接口,以供工作線程調(diào)度任務(wù)的執(zhí)行,它主要規(guī)定了任務(wù)的入口,任務(wù)執(zhí)行完后的收尾工作,任務(wù)的執(zhí)行狀態(tài)等;
4、任務(wù)隊列(taskQueue):用于存放沒有處理的任務(wù)。提供一種緩沖機制。
在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表示當(dāng)前線程池的狀態(tài),它是一個volatile變量用來保證線程之間的可見性;
當(dāng)創(chuàng)建線程池后,初始時,線程池處于RUNNING狀態(tài);
如果調(diào)用了shutdown()方法,則線程池處于SHUTDOWN狀態(tài),此時線程池不能夠接受新的任務(wù),它會等待所有任務(wù)執(zhí)行完畢;
如果調(diào)用了shutdownNow()方法,則線程池處于STOP狀態(tài),此時線程池不能接受新的任務(wù),并且會去嘗試終止正在執(zhí)行的任務(wù);
當(dāng)線程池處于SHUTDOWN或STOP狀態(tài),并且所有工作線程已經(jīng)銷毀,任務(wù)緩存隊列已經(jīng)清空或執(zhí)行結(jié)束后,線程池被設(shè)置為TERMINATED狀態(tài)。
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方法檢查如果當(dāng)前線程池的大小小于配置的核心線程數(shù),說明還可以創(chuàng)建新線程,則啟動新的線程執(zhí)行這個任務(wù)。
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中的內(nèi)部類
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:
判斷如果當(dāng)前狀態(tài)不是RUNING,則當(dāng)前任務(wù)不加入到任務(wù)隊列中,判斷如果狀態(tài)是停止,線程數(shù)小于允許的最大數(shù),且任務(wù)隊列還不空,則加入一個新的工作線程到線程池來幫助處理還未處理完的任務(wù)。
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檢查如果線程池的大小小于配置的最大線程數(shù),并且任務(wù)隊列已經(jīng)滿了(就是execute方法試圖把當(dāng)前線程加入任務(wù)隊列時不成功),
說明現(xiàn)有線程已經(jīng)不能支持當(dāng)前的任務(wù)了,但線程池還有繼續(xù)擴充的空間,就可以創(chuàng)建一個新的線程來處理提交的任務(wù)。
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、如果線程池的當(dāng)前大小還沒有達(dá)到基本大小(poolSize < corePoolSize),那么就新增加一個線程處理新提交的任務(wù);
2、如果當(dāng)前大小已經(jīng)達(dá)到了基本大小,就將新提交的任務(wù)提交到阻塞隊列排隊,等候處理workQueue.offer(command);
3、如果隊列容量已達(dá)上限,并且當(dāng)前大小poolSize沒有達(dá)到maximumPoolSize,那么就新增線程來處理任務(wù);
4、如果隊列已滿,并且當(dāng)前線程數(shù)目也已經(jīng)達(dá)到上限,那么意味著線程池的處理能力已經(jīng)達(dá)到了極限,此時需要拒絕新增加的任務(wù)。至于如何拒絕處理新增的任務(wù),取決于線程池的飽和策略RejectedExecutionHandler。
================================================
設(shè)置合適的線程池大小:
如果是CPU密集型的任務(wù),那么良好的線程個數(shù)是實際CPU處理器的個數(shù)的1倍;
如果是I/O密集型的任務(wù),那么良好的線程個數(shù)是實際CPU處理器個數(shù)的1.5倍到2倍
線程池中線程數(shù)量:
View Code
為什么+1,與CPU核數(shù)相等,表示滿核運行,+1的話表示在CPU上存在競爭,兩者的競爭力不一樣。稍微高一點負(fù)荷是不影響的。
==================================================================================
Java中提供了幾個Executors類的靜態(tài)方法:
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創(chuàng)建的線程池corePoolSize和maximumPoolSize值是相等的,它使用的LinkedBlockingQueue;
newSingleThreadExecutor將corePoolSize和maximumPoolSize都設(shè)置為1,也使用的LinkedBlockingQueue;
newCachedThreadPool將corePoolSize設(shè)置為0,將maximumPoolSize設(shè)置為Integer.MAX_VALUE,使用的SynchronousQueue,也就是說來了任務(wù)就創(chuàng)建線程運行,當(dāng)線程空閑超過60秒,就銷毀線程。
任務(wù)拒絕策略:
當(dāng)線程池的任務(wù)緩存隊列已滿并且線程池中的線程數(shù)目達(dá)到maximumPoolSize,如果還有任務(wù)到來就會采取任務(wù)拒絕策略,通常有以下四種策略:
ThreadPoolExecutor.AbortPolicy:丟棄任務(wù)并拋出RejectedExecutionException異常。
ThreadPoolExecutor.DiscardPolicy:也是丟棄任務(wù),但是不拋出異常。
ThreadPoolExecutor.DiscardOldestPolicy:丟棄隊列最前面的任務(wù),然后重新嘗試執(zhí)行任務(wù)(重復(fù)此過程)
ThreadPoolExecutor.CallerRunsPolicy:由調(diào)用線程處理該任務(wù)
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("線程池中線程數(shù)目:"+executor.getPoolSize()+",隊列中等待執(zhí)行的任務(wù)數(shù)目:"+executor.getQueue().size()+",已執(zhí)行玩別的任務(wù)數(shù)目:"+executor.getCompletedTaskCount());
}
executor.shutdown();
}
}class MyTask implementsRunnable {private inttaskNum;public MyTask(intnum) {this.taskNum =num;
}
@Overridepublic voidrun() {
System.out.println("正在執(zhí)行task "+taskNum);try{
Thread.currentThread().sleep(0);
}catch(InterruptedException e) {
e.printStackTrace();
}
System.out.println("task "+taskNum+"執(zhí)行完畢");
}
}
線程池中線程數(shù)目:1,隊列中等待執(zhí)行的任務(wù)數(shù)目:0,已執(zhí)行玩別的任務(wù)數(shù)目:0線程池中線程數(shù)目:2,隊列中等待執(zhí)行的任務(wù)數(shù)目:0,已執(zhí)行玩別的任務(wù)數(shù)目:0線程池中線程數(shù)目:3,隊列中等待執(zhí)行的任務(wù)數(shù)目:0,已執(zhí)行玩別的任務(wù)數(shù)目:0正在執(zhí)行task0線程池中線程數(shù)目:4,隊列中等待執(zhí)行的任務(wù)數(shù)目:0,已執(zhí)行玩別的任務(wù)數(shù)目:0正在執(zhí)行task3正在執(zhí)行task1task 3執(zhí)行完畢
task 1執(zhí)行完畢
線程池中線程數(shù)目:5,隊列中等待執(zhí)行的任務(wù)數(shù)目:0,已執(zhí)行玩別的任務(wù)數(shù)目:0task 0執(zhí)行完畢
正在執(zhí)行task5線程池中線程數(shù)目:5,隊列中等待執(zhí)行的任務(wù)數(shù)目:1,已執(zhí)行玩別的任務(wù)數(shù)目:2線程池中線程數(shù)目:5,隊列中等待執(zhí)行的任務(wù)數(shù)目:1,已執(zhí)行玩別的任務(wù)數(shù)目:3線程池中線程數(shù)目:5,隊列中等待執(zhí)行的任務(wù)數(shù)目:2,已執(zhí)行玩別的任務(wù)數(shù)目:3線程池中線程數(shù)目:5,隊列中等待執(zhí)行的任務(wù)數(shù)目:3,已執(zhí)行玩別的任務(wù)數(shù)目:3線程池中線程數(shù)目:5,隊列中等待執(zhí)行的任務(wù)數(shù)目:4,已執(zhí)行玩別的任務(wù)數(shù)目:3線程池中線程數(shù)目:5,隊列中等待執(zhí)行的任務(wù)數(shù)目:5,已執(zhí)行玩別的任務(wù)數(shù)目:3task 5執(zhí)行完畢
正在執(zhí)行task6task 6執(zhí)行完畢
正在執(zhí)行task7task 7執(zhí)行完畢
正在執(zhí)行task8task 8執(zhí)行完畢
正在執(zhí)行task9task 9執(zhí)行完畢
正在執(zhí)行task10task 10執(zhí)行完畢
線程池中線程數(shù)目:6,隊列中等待執(zhí)行的任務(wù)數(shù)目:0,已執(zhí)行玩別的任務(wù)數(shù)目:9線程池中線程數(shù)目:6,隊列中等待執(zhí)行的任務(wù)數(shù)目:1,已執(zhí)行玩別的任務(wù)數(shù)目:9線程池中線程數(shù)目:6,隊列中等待執(zhí)行的任務(wù)數(shù)目:2,已執(zhí)行玩別的任務(wù)數(shù)目:9線程池中線程數(shù)目:6,隊列中等待執(zhí)行的任務(wù)數(shù)目:3,已執(zhí)行玩別的任務(wù)數(shù)目:9正在執(zhí)行task12正在執(zhí)行task14正在執(zhí)行task13task 14執(zhí)行完畢
task 13執(zhí)行完畢
task 12執(zhí)行完畢
正在執(zhí)行task2task 2執(zhí)行完畢
正在執(zhí)行task4task 4執(zhí)行完畢
正在執(zhí)行task11task 11執(zhí)行完畢
View Code
總結(jié)
以上是生活随笔為你收集整理的java线程池的工作原理_Java 线程池的介绍以及工作原理的全部內(nèi)容,希望文章能夠幫你解決所遇到的問題。
- 上一篇: java kafka 集群消费_kafk
- 下一篇: Java怎么定义图片公共路径_【Java