Tomcat如何使用執行緒池處理遠端併發請求
阿新 • • 發佈:2020-12-25
# Tomcat如何使用執行緒池處理遠端併發請求
通過了解學習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
是一個執行緒池實現類,管理執行緒,減少執行緒開銷,可以用來提高任務執行效率,
構造方法中的引數有
```java
public ThreadPoolExecutor(
int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue workQueue,
ThreadFactory threadFactory,
RejectedExecutionHandler handler) {
}
```
corePoolSize 是核心執行緒數
maximumPoolSize 是最大執行緒數
keepAliveTime 非核心執行緒最大空閒時間(超過時間終止)
unit 時間單位
workQueue 佇列,當任務過多時,先存放在佇列
threadFactory 執行緒工廠,建立執行緒的工廠
handler 決絕策略,當任務數過多,佇列不能再存放任務時,該如何處理,由此物件去處理。這是個介面,你可以自定義處理方式
## ThreadPoolExecutor在Tomcat中http請求的應用
此執行緒池是tomcat用來在接收到遠端請求後,將每次請求單獨作為一個任務去處理,每次呼叫execute(Runnable)
### 初始化
`org.apache.tomcat.util.net.NioEndpoint`
NioEndpoint初始化的時候,建立了執行緒池
```java
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);
}
```
線上程池建立時,呼叫prestartAllCoreThreads(), 初始化核心工作執行緒worker,並啟動
```java
public int prestartAllCoreThreads() {
int n = 0;
while (addWorker(null, true))
++n;
return n;
}
```
當addWorker 數量等於corePoolSize時,addWorker(null,ture)會返回false,停止worker工作執行緒的建立
### 提交任務到佇列
每次客戶端過來請求(http),就會提交一次處理任務,
worker 從佇列中獲取任務執行,下面是任務放入佇列的邏輯程式碼
ThreadPoolExecutor.execute(Runnable) 提交任務:
```java
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))
reject(command);
}
```
workQueue.offer(command) 完成了任務的提交(在tomcat處理遠端http請求時)。
#### workQueue.offer
TaskQueue 是 BlockingQueue 具體實現類,workQueue.offer(command)實際程式碼:
```java
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 node = new Node(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 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方法:
```java
/** Delegates main run loop to outer runWorker */
public void run() {
runWorker(this);
}
```
#### 迴圈獲取佇列中的任務
runWorker(worker)方法 迴圈部分程式碼:
```java
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.獲取佇列的任務,保證其他執行緒不能同時操作佇列
#### 給佇列新增任務上鎖
```java
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 node = new Node(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;
}
```
### 獲取佇列任務上鎖
```java
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 高併發相關類
執行緒池使用中,有平凡用到Unsafe類,這個類在高併發中,能做一些原子CAS操作,鎖執行緒,釋放執行緒等。
`sun.misc.Unsafe` 類是底層類,openjdk原始碼中有
### 原子操作資料
java.util.concurrent.locks.AbstractQueuedSynchronizer 類中就有保證原子操作的程式碼
```java
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
//對應的java底層,實際是native方法,對應C++程式碼
/**
* Atomically update Java variable to x if it is currently
* holding expected.
* @return true 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,告訴操作者此次原子性修改失敗。
### 阻塞和喚醒執行緒
public native void park(boolean isAbsolute, long time); //阻塞當前執行緒
執行緒池的worker角色迴圈獲取佇列任務,如果佇列中沒有任務,worker.run 還是在等待的,不會退出執行緒,程式碼中用了`notEmpty.await() ` 中斷此worker執行緒,放入一個等待執行緒佇列(區別去任務佇列);當有新任務需要時,再`notEmpty.signal()`喚醒此執行緒
底層分別是
unsafe.park() 阻塞當前執行緒
public native void park(boolean isAbsolute, long time);
unsafe.unpark() 喚醒執行緒
public native void unpark(Object thread);
這個操作是對應的,阻塞時,先將thread放入佇列,喚醒時,從佇列拿出被阻塞的執行緒,unsafe.unpark(thread)喚醒指定執行緒。
`java.util.concurrent.locks.AbstractQueuedLongSynchronizer.ConditionObject` 類中
通過連結串列存放執行緒資訊
```java
// 新增一個阻塞執行緒
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);