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

是一個線程池實現類,管理線程,減少線程開銷,可以用來提高任務執行效率,

構造方法中的參數有

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用來在接收到遠程請求後,將每次請求單獨作為一個任務去處理,每次調用execute(Runnable)

初始化

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);
 }

在線程池創建時,調用prestartAllCoreThreads(), 初始化核心工作線程worker,並啟動

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) 提交任務:

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)實際代碼:

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 高並發相關類

線程池使用中,有平凡用到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,告訴操作者此次原子性修改失敗。

阻塞和喚醒線程

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 類中

通過鏈表存放線程信息

// 添加一個阻塞線程
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);
 }

到此這篇關於Tomcat使用線程池處理遠程並發請求的方法的文章就介紹到這瞭,更多相關Tomcat線程池處理遠程並發請求內容請搜索WalkonNet以前的文章或繼續瀏覽下面的相關文章希望大傢以後多多支持WalkonNet!

推薦閱讀: