Tomcat用線程池處理http並發請求
通過了解學習tomcat如何處理並發請求，了解到線程池，鎖，隊列，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
是一個線程池實現類，管理線程，減少線程開銷，可以用來提高任務執行效率，
構造方法中的參數有
public ThreadPoolExecutor(
int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue,
ThreadFactory threadFactory,
RejectedExecutionHandler handler) {
}
corePoolSize 是核心線程數
maximumPoolSize 是最大線程數
keepAliveTime 非核心線程最大空閑時間（超過時間終止）
unit 時間單位
workQueue 隊列，當任務過多時，先存放在隊列
threadFactory 線程工廠，創建線程的工廠
handler 拒絕策略，當任務數過多，隊列不能再存放任務時，該如何處理，由此對象去處理。這是個接口，你可以自定義處理方式
ThreadPoolExecutor在Tomcat中http請求的應用
tomcat有一個自己的線程池類：org.apache.tomcat.util.threads.ThreadPoolExecutor，繼承原先java.util.concurrent.ThreadPoolExecutor類，此線程池是tomcat用來在接收到遠程請求后，將每次請求單獨作為一個任務去處理使用，即調用execute(Runnable)，此類重寫了execute方法，做了一點功能擴展，有一個功能是為了判斷worker數量是否足夠，判斷不足夠時，添加非核心線程worker
org.apache.tomcat.util.threads.ThreadPoolExecutor 部分功能擴展代碼：
private final AtomicInteger submittedCount = new AtomicInteger(0); //提交任務總數
// 重寫 execute(Runnable command)
public void execute(Runnable command) {
execute(command,0,TimeUnit.MILLISECONDS);
}
public void execute(Runnable command, long timeout, TimeUnit unit) {
submittedCount.incrementAndGet(); // 提交任務之前，總數 + 1
try {
super.execute(command);
} catch (RejectedExecutionException rx) {
}
}
//重寫 afterExecute 添加任務完成后的邏輯
@Override
protected void afterExecute(Runnable r, Throwable t) {
if (!(t instanceof StopPooledThreadException)) {
submittedCount.decrementAndGet(); // 完成任務后 總數 -1
}
if (t == null) {
stopCurrentThreadIfNeeded();
}
}
上面是tomcat自己的線程池判斷是否需要添加非核心線程關鍵部分，在workQueue.offer時，會拿submittedCount這個數作為是否添加woker的一個依據。
workQueue.offer見下文
初始化
org.apache.tomcat.util.net.NioEndpoint
創建線程池
NioEndpoint初始化的時候，創建了線程池
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);
}
創建工作線程worker
在線程池創建時，調用prestartAllCoreThreads(), 初始化核心工作線程worker，並啟動
public int prestartAllCoreThreads() {
int n = 0;
while (addWorker(null, true))
++n;
return n;
}
當addWorker 數量等於corePoolSize時，addWorker(null,ture)會返回false,停止worker工作線程的創建
addWorker時，會啟動worker線程
private boolean addWorker(Runnable firstTask, boolean core) {
//......省去判斷代碼（是否需要添加worker的判斷）
boolean workerStarted = false;
boolean workerAdded = false;
Worker w = null;
try {
w = new Worker(firstTask);//1 創建worker線程
final Thread t = w.thread;
if (t != null) {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
workers.add(w);
workerAdded = true;
}
} finally {
mainLock.unlock();
}
if (workerAdded) {
t.start(); //2 如果worker創建成功，啟動這個工作線程
workerStarted = true; //返回true
}
}
} finally {
if (! workerStarted)
addWorkerFailed(w);
}
return workerStarted;
}
接收任務放入隊列
每次客戶端過來請求（http），就會提交一次處理任務，
poller對象的run方法中開始 -> processKey() -> processSocket() -> executor.execute()
//org.apache.tomcat.util.net.NioEndpoint.Poller.run()
@Override
public void run() {
// Loop until destroy() is called
while (true) {
//...............
NioSocketWrapper socketWrapper = (NioSocketWrapper) sk.attachment();
if (socketWrapper != null) {
//1調用processKey方法
processKey(sk, socketWrapper);
}
//.............
}
}
//org.apache.tomcat.util.net.NioEndpoint.Poller.processKey(SelectionKey, NioSocketWrapper)
protected void processKey(SelectionKey sk, NioSocketWrapper socketWrapper) {
try {
//....................
// 2調用processSocket方法
processSocket(socketWrapper, SocketEvent.OPEN_WRITE, true))
//..................
}
}
//org.apache.tomcat.util.net.AbstractEndpoint.processSocket(SocketWrapperBase<S>, SocketEvent, boolean)
public boolean processSocket(SocketWrapperBase<S> socketWrapper,
SocketEvent event, boolean dispatch) {
try {
//...............
Executor executor = getExecutor();
if (dispatch && executor != null) {
executor.execute(sc); // 3調用ThreadPoolExecutor.execute提交新請求任務
} else {
sc.run();
}
//.....................
return true;
}
ThreadPoolExecutor.execute
worker 從隊列中獲取任務運行，下面是將任務放入隊列的邏輯代碼
ThreadPoolExecutor.execute(Runnable) 提交任務：
public void execute(Runnable command) {
if (command == null)
throw new NullPointerException();
int c = ctl.get();
// worker數 是否小於 核心線程數 tomcat中初始化后，一般不滿足第一個條件，不會addWorker
if (workerCountOf(c) < corePoolSize) {
if (addWorker(command, true))
return;
c = ctl.get();
}
// workQueue.offer(command)，將任務添加到隊列
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)) //workQueue.offer 返回false時，添加非核心線程
reject(command);
}
workQueue.offer(command) 最終完成了任務的提交(在tomcat處理遠程http請求時)。
workQueue.offer
TaskQueue 是 BlockingQueue 具體實現類，TaskQueue在offer時，首先會判斷一些條件，如果TaskQueue覺得worker數量不夠，會添加worker，但不是核心線程；
corePoolSize = 10， maximumPoolSize=200 時，並發量小，一般線程數10（核心線程數），若並發非常大，最多也只能創建200個worker線程，190個線程在任務處理完后，閑時狀態下會被回收，worker數回到10的數量；
workQueue.offer(command)實際代碼：
//TaskQueue
@Override
public boolean offer(Runnable o) {
if (parent.getSubmittedCount()<=(parent.getPoolSize())) return super.offer(o);
if (parent.getPoolSize()<parent.getMaximumPoolSize()) return false;
// 當任務提交過多：未處理任務數(SubmittedCount) > 線程數，並且 poolSize < maximumPoolSize
// 返回false ThreadPoolExecutor會 addWorker(command, false) 添加worker線程
return super.offer(o);
}
//super.offer LinkedBlockingQueue
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;
}
// 添加任務到隊列
/**
* 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; //鏈表結構 last.next = node; last = node
}
之后是worker的工作，worker在run方法中通過去getTask()獲取此處提交的任務，並執行完成任務。
線程池如何處理新提交的任務
添加worker之后，提交任務，因為worker數量達到corePoolSize，任務都會將放入隊列，而worker的run方法則是循環獲取隊列中的任務（不為空時），
worker run方法：
/** Delegates main run loop to outer runWorker */
public void run() {
runWorker(this);
}
循環獲取隊列中的任務
runWorker(worker)方法 循環部分代碼：
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) { //循環獲取隊列中的任務
w.lock(); // 上鎖
try {
// 運行前處理
beforeExecute(wt, task);
// 隊列中的任務開始執行
task.run();
// 運行后處理
afterExecute(task, thrown);
} finally {
task = null;
w.completedTasks++;
w.unlock(); // 釋放鎖
}
}
completedAbruptly = false;
} finally {
processWorkerExit(w, completedAbruptly);
}
}
task.run()執行任務
鎖運用
鎖用於保證過程的有序，一般一段代碼上鎖后，同一時間只允許一個線程去操作
ThreadPoolExecutor 使用鎖主要保證兩件事情，
1.給隊列添加任務，釋放鎖之前，保證其他線程不能操作隊列-添加隊列任務）
2.獲取隊列的任務，釋放鎖之前，保證其他線程不能操作隊列-取出隊列任務）
在高並發情況下，鎖能有效保證請求的有序處理，不至於混亂
給隊列添加任務時上鎖
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;
}
獲取隊列任務時上鎖
private Runnable getTask() {
boolean timedOut = false; // Did the last poll() time out?
// ...省略
for (;;) {
try {
Runnable r = timed ?
workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
workQueue.take(); //獲取隊列中一個任務
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(); //如果隊列中沒有任務，等待
}
x = dequeue();
c = count.getAndDecrement();
if (c > 1)
notEmpty.signal();
} finally {
takeLock.unlock(); // 釋放鎖
}
if (c == capacity)
signalNotFull();
return x;
}
其他
volatile
在並發場景這個關鍵字修飾成員變量很常見，
主要目的公共變量在被某一個線程修改時，對其他線程可見（實時）
sun.misc.Unsafe 高並發相關類API
線程池使用中，有平凡用到Unsafe類，這個類在高並發中，能做一些原子CAS操作，鎖線程，釋放線程等。
sun.misc.Unsafe 類是底層類，openjdk源碼中有
原子操作數據
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);
}
對應Unsafe類的代碼:
//對應的java底層，實際是native方法，對應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);
方法的作用簡單來說就是 更新一個值，保證原子性操作
當你要操作一個對象o的一個成員變量offset時,修改o.offset，
高並發下為保證准確性，你在操作o.offset的時候，讀應該是正確的值，並且中間不能被別的線程修改來保證高並發的環境數據操作有效。
即 expected 期望值與內存中的值比較是一樣的expected == 內存中的值 ，則更新值為 x，返回true代表修改成功
否則，期望值與內存值不同，說明值被其他線程修改過，不能更新值為x，並返回false，告訴操作者此次原子性修改失敗。
注意一下能知道這是locks包下的類，ReentrantLock鎖的底層原理就與unsafe類有關，以及下面的park，unpark。線程可以通過這個原子操作放回true或者false的機制，定義自己獲取鎖成功還是失敗。
阻塞和喚醒線程
ThreadPoolExecute設計在請求隊列任務為空時，worker線程可以是等待或者中斷的（非銷毀狀態）。
這種做法避免了沒必要的循環，節省了硬件資源，提高線程使用效率，
線程池的worker角色循環獲取隊列任務，如果隊列中沒有任務，worker.run 還是在等待的，不會退出線程，代碼中用了notEmpty.await() 中斷此worker線程，放入一個等待線程隊列（區別去任務隊列）；當有新任務需要時，再notEmpty.signal()喚醒此線程
底層分別是
park
unsafe.park() 阻塞(停止)當前線程
public native void park(boolean isAbsolute, long time);
unpark
unsafe.unpark() 喚醒(取消停止)線程
public native void unpark(Object thread);
這個操作是對應的，
阻塞時，先將thread放入隊列,再park，
喚醒時，從隊列拿出被阻塞的線程，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);
}
這里要區分park 和 compareAndSwapInt是兩個完全不同的東西，可以單獨或者組合使用，
比如ReentrantLock實現鎖功能這兩個都需要