Tomcat使用線程池處理遠程并發(fā)請求的方法
通過了解學(xué)習(xí)tomcat如何處理并發(fā)請求,了解到線程池,鎖,隊列,unsafe類,下面的主要代碼來自
java-jre:
sun.misc.Unsafe java.util.concurrent.ThreadPoolExecutor java.util.concurrent.ThreadPoolExecutor.Worker java.util.concurrent.locks.AbstractQueuedSynchronizer java.util.concurrent.locks.AbstractQueuedLongSynchronizer java.util.concurrent.LinkedBlockingQueue
tomcat:
org.apache.tomcat.util.net.NioEndpoint org.apache.tomcat.util.threads.ThreadPoolExecutor org.apache.tomcat.util.threads.TaskThreadFactory org.apache.tomcat.util.threads.TaskQueue
ThreadPoolExecutor
是一個線程池實現(xiàn)類,管理線程,減少線程開銷,可以用來提高任務(wù)執(zhí)行效率,
構(gòu)造方法中的參數(shù)有
public ThreadPoolExecutor( int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler) { }
corePoolSize 是核心線程數(shù)
maximumPoolSize 是最大線程數(shù)
keepAliveTime 非核心線程最大空閑時間(超過時間終止)
unit 時間單位
workQueue 隊列,當(dāng)任務(wù)過多時,先存放在隊列
threadFactory 線程工廠,創(chuàng)建線程的工廠
handler 決絕策略,當(dāng)任務(wù)數(shù)過多,隊列不能再存放任務(wù)時,該如何處理,由此對象去處理。這是個接口,你可以自定義處理方式
ThreadPoolExecutor在Tomcat中http請求的應(yīng)用
此線程池是tomcat用來在接收到遠程請求后,將每次請求單獨作為一個任務(wù)去處理,每次調(diào)用execute(Runnable)
初始化
org.apache.tomcat.util.net.NioEndpoint
NioEndpoint初始化的時候,創(chuàng)建了線程池
public void createExecutor() { internalExecutor = true; TaskQueue taskqueue = new TaskQueue(); //TaskQueue無界隊列,可以一直添加,因此handler 等同于無效 TaskThreadFactory tf = new TaskThreadFactory(getName() + "-exec-", daemon, getThreadPriority()); executor = new ThreadPoolExecutor(getMinSpareThreads(), getMaxThreads(), 60, TimeUnit.SECONDS,taskqueue, tf); taskqueue.setParent( (ThreadPoolExecutor) executor); }
在線程池創(chuàng)建時,調(diào)用prestartAllCoreThreads(), 初始化核心工作線程worker,并啟動
public int prestartAllCoreThreads() { int n = 0; while (addWorker(null, true)) ++n; return n; }
當(dāng)addWorker 數(shù)量等于corePoolSize時,addWorker(null,ture)會返回false,停止worker工作線程的創(chuàng)建
提交任務(wù)到隊列
每次客戶端過來請求(http),就會提交一次處理任務(wù),
worker 從隊列中獲取任務(wù)運行,下面是任務(wù)放入隊列的邏輯代碼
ThreadPoolExecutor.execute(Runnable) 提交任務(wù):
public void execute(Runnable command) { if (command == null) throw new NullPointerException(); int c = ctl.get(); // worker數(shù) 是否小于 核心線程數(shù) tomcat中初始化后,一般不滿足第一個條件,不會addWorker if (workerCountOf(c) < corePoolSize) { if (addWorker(command, true)) return; c = ctl.get(); } // workQueue.offer(command),將任務(wù)添加到隊列, if (isRunning(c) && workQueue.offer(command)) { int recheck = ctl.get(); if (! isRunning(recheck) && remove(command)) reject(command); else if (workerCountOf(recheck) == 0) addWorker(null, false); } else if (!addWorker(command, false)) reject(command); }
workQueue.offer(command) 完成了任務(wù)的提交(在tomcat處理遠程http請求時)。
workQueue.offer
TaskQueue 是 BlockingQueue 具體實現(xiàn)類,workQueue.offer(command)實際代碼:
public boolean offer(E e) { if (e == null) throw new NullPointerException(); final AtomicInteger count = this.count; if (count.get() == capacity) return false; int c = -1; Node<E> node = new Node<E>(e); final ReentrantLock putLock = this.putLock; putLock.lock(); try { if (count.get() < capacity) { enqueue(node); //此處將任務(wù)添加到隊列 c = count.getAndIncrement(); if (c + 1 < capacity) notFull.signal(); } } finally { putLock.unlock(); } if (c == 0) signalNotEmpty(); return c >= 0; } // 添加任務(wù)到隊列 /** * Links node at end of queue. * * @param node the node */ private void enqueue(Node<E> node) { // assert putLock.isHeldByCurrentThread(); // assert last.next == null; last = last.next = node; //鏈表結(jié)構(gòu) last.next = node; last = node }
之后是worker的工作,worker在run方法中通過去getTask()獲取此處提交的任務(wù),并執(zhí)行完成任務(wù)。
線程池如何處理新提交的任務(wù)
添加worker之后,提交任務(wù),因為worker數(shù)量達到corePoolSize,任務(wù)都會將放入隊列,而worker的run方法則是循環(huán)獲取隊列中的任務(wù)(不為空時),
worker run方法:
/** Delegates main run loop to outer runWorker */ public void run() { runWorker(this); }
循環(huán)獲取隊列中的任務(wù)
runWorker(worker)方法 循環(huán)部分代碼:
final void runWorker(Worker w) { Thread wt = Thread.currentThread(); Runnable task = w.firstTask; w.firstTask = null; w.unlock(); // allow interrupts boolean completedAbruptly = true; try { while (task != null || (task = getTask()) != null) { //循環(huán)獲取隊列中的任務(wù) w.lock(); // 上鎖 try { // 運行前處理 beforeExecute(wt, task); // 隊列中的任務(wù)開始執(zhí)行 task.run(); // 運行后處理 afterExecute(task, thrown); } finally { task = null; w.completedTasks++; w.unlock(); // 釋放鎖 } } completedAbruptly = false; } finally { processWorkerExit(w, completedAbruptly); } }
task.run()執(zhí)行任務(wù)
鎖運用
ThreadPoolExecutor 使用鎖主要保證兩件事情,
1.給隊列添加任務(wù),保證其他線程不能操作隊列
2.獲取隊列的任務(wù),保證其他線程不能同時操作隊列
給隊列添加任務(wù)上鎖
public boolean offer(E e) { if (e == null) throw new NullPointerException(); final AtomicInteger count = this.count; if (count.get() == capacity) return false; int c = -1; Node<E> node = new Node<E>(e); final ReentrantLock putLock = this.putLock; putLock.lock(); //上鎖 try { if (count.get() < capacity) { enqueue(node); c = count.getAndIncrement(); if (c + 1 < capacity) notFull.signal(); } } finally { putLock.unlock(); //釋放鎖 } if (c == 0) signalNotEmpty(); return c >= 0; }
獲取隊列任務(wù)上鎖
private Runnable getTask() { boolean timedOut = false; // Did the last poll() time out? // ...省略 for (;;) { try { Runnable r = timed ? workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) : workQueue.take(); //獲取隊列中一個任務(wù) if (r != null) return r; timedOut = true; } catch (InterruptedException retry) { timedOut = false; } } } public E take() throws InterruptedException { E x; int c = -1; final AtomicInteger count = this.count; final ReentrantLock takeLock = this.takeLock; takeLock.lockInterruptibly(); // 上鎖 try { while (count.get() == 0) { notEmpty.await(); //如果隊列中沒有任務(wù),等待 } x = dequeue(); c = count.getAndDecrement(); if (c > 1) notEmpty.signal(); } finally { takeLock.unlock(); // 釋放鎖 } if (c == capacity) signalNotFull(); return x; }
volatile
在并發(fā)場景這個關(guān)鍵字修飾成員變量很常見,
主要目的公共變量在被某一個線程修改時,對其他線程可見(實時)
sun.misc.Unsafe 高并發(fā)相關(guān)類
線程池使用中,有平凡用到Unsafe類,這個類在高并發(fā)中,能做一些原子CAS操作,鎖線程,釋放線程等。
sun.misc.Unsafe
類是底層類,openjdk源碼中有
原子操作數(shù)據(jù)
java.util.concurrent.locks.AbstractQueuedSynchronizer 類中就有保證原子操作的代碼
protected final boolean compareAndSetState(int expect, int update) { // See below for intrinsics setup to support this return unsafe.compareAndSwapInt(this, stateOffset, expect, update); }
對應(yīng)Unsafe類的代碼:
//對應(yīng)的java底層,實際是native方法,對應(yīng)C++代碼 /** * Atomically update Java variable to <tt>x</tt> if it is currently * holding <tt>expected</tt>. * @return <tt>true</tt> if successful */ public final native boolean compareAndSwapInt(Object o, long offset, int expected, int x);
方法的作用簡單來說就是 更新一個值,保證原子性操作
當(dāng)你要操作一個對象o
的一個成員變量offset
時,修改o.offset,
高并發(fā)下為保證準(zhǔn)確性,你在操作o.offset的時候,讀應(yīng)該是正確的值,并且中間不能被別的線程修改來保證高并發(fā)的環(huán)境數(shù)據(jù)操作有效。
即 expected 期望值與內(nèi)存中的值比較是一樣的expected == 內(nèi)存中的值 ,則更新值為 x,返回true代表修改成功
否則,期望值與內(nèi)存值不同,說明值被其他線程修改過,不能更新值為x,并返回false,告訴操作者此次原子性修改失敗。
阻塞和喚醒線程
public native void park(boolean isAbsolute, long time); //阻塞當(dāng)前線程
線程池的worker角色循環(huán)獲取隊列任務(wù),如果隊列中沒有任務(wù),worker.run 還是在等待的,不會退出線程,代碼中用了notEmpty.await()
中斷此worker線程,放入一個等待線程隊列(區(qū)別去任務(wù)隊列);當(dāng)有新任務(wù)需要時,再notEmpty.signal()
喚醒此線程
底層分別是
unsafe.park() 阻塞當(dāng)前線程
public native void park(boolean isAbsolute, long time);
unsafe.unpark() 喚醒線程
public native void unpark(Object thread);
這個操作是對應(yīng)的,阻塞時,先將thread放入隊列,喚醒時,從隊列拿出被阻塞的線程,unsafe.unpark(thread)喚醒指定線程。
java.util.concurrent.locks.AbstractQueuedLongSynchronizer.ConditionObject
類中
通過鏈表存放線程信息
// 添加一個阻塞線程 private Node addConditionWaiter() { Node t = lastWaiter; // If lastWaiter is cancelled, clean out. if (t != null && t.waitStatus != Node.CONDITION) { unlinkCancelledWaiters(); t = lastWaiter; } Node node = new Node(Thread.currentThread(), Node.CONDITION); if (t == null) firstWaiter = node; else t.nextWaiter = node; lastWaiter = node; //將新阻塞的線程放到鏈表尾部 return node; } // 拿出一個被阻塞的線程 public final void signal() { if (!isHeldExclusively()) throw new IllegalMonitorStateException(); Node first = firstWaiter; //鏈表中第一個阻塞的線程 if (first != null) doSignal(first); } // 拿到后,喚醒此線程 final boolean transferForSignal(Node node) { LockSupport.unpark(node.thread); return true; } public static void unpark(Thread thread) { if (thread != null) UNSAFE.unpark(thread); }
到此這篇關(guān)于Tomcat使用線程池處理遠程并發(fā)請求的方法的文章就介紹到這了,更多相關(guān)Tomcat線程池處理遠程并發(fā)請求內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家!
相關(guān)文章
解決tomcat部署時war和war exploded導(dǎo)致的問題
war包和warexploded模式的區(qū)別主要在于部署方式,war包是在項目發(fā)布時打包上傳,而warexploded模式則是直接將文件夾結(jié)構(gòu)上傳到服務(wù)器,支持熱部署,常用于開發(fā)階段,這種差異可能導(dǎo)致樣式和文件引用問題,如在不同電腦部署時路徑設(shè)置的不同2024-10-10同一臺服務(wù)器(電腦)運行多個Tomcat的設(shè)置方法步驟
這篇文章主要介紹了同一臺服務(wù)器(電腦)運行多個Tomcat的設(shè)置方法步驟,小編覺得挺不錯的,現(xiàn)在分享給大家,也給大家做個參考。一起跟隨小編過來看看吧2017-04-04