17站長網(wǎng) › 首頁 ›網(wǎng)站› 服務(wù)器 › 查看內(nèi)容

Tomcat使用線程池處理遠(yuǎn)程并發(fā)請求的方法

2022-12-3 14:49| 查看: 2240 |來源: 互聯(lián)網(wǎng)

大家好，現(xiàn)在小編來為大家解答以上問題。Tomcat使用線程池處理遠(yuǎn)程并發(fā)請求的方法相信很多人還不知道，現(xiàn)在讓我們一起來看看吧！通過了解學(xué)習(xí)tomcat如何處理并發(fā)請求，了解到線程池，鎖，隊(duì)列，unsafe類，下面的主要

大家好，現(xiàn)在小編來為大家解答以上問題。Tomcat使用線程池處理遠(yuǎn)程并發(fā)請求的方法相信很多人還不知道，現(xiàn)在讓我們一起來看看吧！

通過了解學(xué)習(xí)tomcat如何處理并發(fā)請求，了解到線程池，鎖，隊(duì)列，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

是一個(gè)線程池實(shí)現(xiàn)類，管理線程，減少線程開銷，可以用來提高任務(wù)執(zhí)行效率，

構(gòu)造方法中的參數(shù)有


public ThreadPoolExecutor(
 int corePoolSize,
 int maximumPoolSize,
 long keepAliveTime,
 TimeUnit unit,
 BlockingQueue workQueue,
 ThreadFactory threadFactory,
 RejectedExecutionHandler handler) {
 
}

corePoolSize 是核心線程數(shù)
maximumPoolSize 是最大線程數(shù)
keepAliveTime 非核心線程最大空閑時(shí)間（超過時(shí)間終止）
unit 時(shí)間單位
workQueue 隊(duì)列，當(dāng)任務(wù)過多時(shí)，先存放在隊(duì)列
threadFactory 線程工廠，創(chuàng)建線程的工廠
handler 決絕策略，當(dāng)任務(wù)數(shù)過多，隊(duì)列不能再存放任務(wù)時(shí)，該如何處理，由此對象去處理。這是個(gè)接口，你可以自定義處理方式

ThreadPoolExecutor在Tomcat中http請求的應(yīng)用

此線程池是tomcat用來在接收到遠(yuǎn)程請求后，將每次請求單獨(dú)作為一個(gè)任務(wù)去處理，每次調(diào)用execute(Runnable)

初始化

org.apache.tomcat.util.net.NioEndpoint

NioEndpoint初始化的時(shí)候，創(chuàng)建了線程池


public void createExecutor() {
 internalExecutor = true;
 TaskQueue taskqueue = new TaskQueue();
 //TaskQueue無界隊(duì)列，可以一直添加，因此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)建時(shí)，調(diào)用prestartAllCoreThreads(), 初始化核心工作線程worker，并啟動(dòng)


public int prestartAllCoreThreads() {
 int n = 0;
 while (addWorker(null, true))
  ++n;
 return n;
 }

當(dāng)addWorker 數(shù)量等于corePoolSize時(shí)，addWorker(null,ture)會(huì)返回false,停止worker工作線程的創(chuàng)建

提交任務(wù)到隊(duì)列

每次客戶端過來請求（http），就會(huì)提交一次處理任務(wù)，

worker 從隊(duì)列中獲取任務(wù)運(yùn)行，下面是任務(wù)放入隊(duì)列的邏輯代碼

ThreadPoolExecutor.execute(Runnable) 提交任務(wù)：


public void execute(Runnable command) {
 if (command == null)
  throw new NullPointerException();
 
 int c = ctl.get();
 	// worker數(shù) 是否小于 核心線程數(shù) tomcat中初始化后，一般不滿足第一個(gè)條件，不會(huì)addWorker
 if (workerCountOf(c) < corePoolSize) {
  if (addWorker(command, true))
  return;
  c = ctl.get();
 }
 	// workQueue.offer(command)，將任務(wù)添加到隊(duì)列，
 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處理遠(yuǎn)程http請求時(shí))。

workQueue.offer

TaskQueue 是 BlockingQueue 具體實(shí)現(xiàn)類，workQueue.offer(command)實(shí)際代碼：


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); //此處將任務(wù)添加到隊(duì)列
  c = count.getAndIncrement();
  if (c + 1 < capacity)
  notFull.signal();
 }
 } finally {
 putLock.unlock();
 }
 if (c == 0)
 signalNotEmpty();
 return c >= 0;
}

// 添加任務(wù)到隊(duì)列
/**
 * 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; //鏈表結(jié)構(gòu) last.next = node; last = node
}

之后是worker的工作，worker在run方法中通過去getTask()獲取此處提交的任務(wù)，并執(zhí)行完成任務(wù)。

線程池如何處理新提交的任務(wù)

添加worker之后，提交任務(wù)，因?yàn)閣orker數(shù)量達(dá)到corePoolSize，任務(wù)都會(huì)將放入隊(duì)列，而worker的run方法則是循環(huán)獲取隊(duì)列中的任務(wù)（不為空時(shí)），

worker run方法：


/** Delegates main run loop to outer runWorker */
 public void run() {
  runWorker(this);
 }

循環(huán)獲取隊(duì)列中的任務(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)獲取隊(duì)列中的任務(wù)
  w.lock(); // 上鎖
  try {
   // 運(yùn)行前處理
   beforeExecute(wt, task);
   // 隊(duì)列中的任務(wù)開始執(zhí)行
   task.run();
   // 運(yùn)行后處理
   afterExecute(task, thrown);
  } finally {
   task = null;
   w.completedTasks++;
   w.unlock(); // 釋放鎖
  }
  }
  completedAbruptly = false;
 } finally {
  processWorkerExit(w, completedAbruptly);
 }
 }

task.run()執(zhí)行任務(wù)

鎖運(yùn)用

ThreadPoolExecutor 使用鎖主要保證兩件事情，
1.給隊(duì)列添加任務(wù)，保證其他線程不能操作隊(duì)列
2.獲取隊(duì)列的任務(wù)，保證其他線程不能同時(shí)操作隊(duì)列

給隊(duì)列添加任務(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 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;
 }

獲取隊(duì)列任務(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(); //獲取隊(duì)列中一個(gè)任務(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(); //如果隊(duì)列中沒有任務(wù)，等待
  }
  x = dequeue();
  c = count.getAndDecrement();
  if (c > 1)
  notEmpty.signal();
 } finally {
  takeLock.unlock(); // 釋放鎖
 }
 if (c == capacity)
  signalNotFull();
 return x;
 }

volatile

在并發(fā)場景這個(gè)關(guān)鍵字修飾成員變量很常見，

主要目的公共變量在被某一個(gè)線程修改時(shí)，對其他線程可見（實(shí)時(shí)）

sun.misc.Unsafe 高并發(fā)相關(guān)類

線程池使用中，有平凡用到Unsafe類，這個(gè)類在高并發(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底層，實(shí)際是native方法，對應(yīng)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);

方法的作用簡單來說就是更新一個(gè)值，保證原子性操作
當(dāng)你要操作一個(gè)對象o的一個(gè)成員變量offset時(shí),修改o.offset，
高并發(fā)下為保證準(zhǔn)確性，你在操作o.offset的時(shí)候，讀應(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)獲取隊(duì)列任務(wù)，如果隊(duì)列中沒有任務(wù)，worker.run 還是在等待的，不會(huì)退出線程，代碼中用了notEmpty.await() 中斷此worker線程，放入一個(gè)等待線程隊(duì)列（區(qū)別去任務(wù)隊(duì)列）；當(dāng)有新任務(wù)需要時(shí)，再notEmpty.signal()喚醒此線程

底層分別是
unsafe.park() 阻塞當(dāng)前線程
public native void park(boolean isAbsolute, long time);

unsafe.unpark() 喚醒線程
public native void unpark(Object thread);

這個(gè)操作是對應(yīng)的，阻塞時(shí)，先將thread放入隊(duì)列，喚醒時(shí)，從隊(duì)列拿出被阻塞的線程，unsafe.unpark(thread)喚醒指定線程。

java.util.concurrent.locks.AbstractQueuedLongSynchronizer.ConditionObject 類中

通過鏈表存放線程信息


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

// 拿出一個(gè)被阻塞的線程
 public final void signal() {
  if (!isHeldExclusively())
  throw new IllegalMonitorStateException();
  Node first = firstWaiter; //鏈表中第一個(gè)阻塞的線程
  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使用線程池處理遠(yuǎn)程并發(fā)請求的方法就是小編為大家收集整理的全部內(nèi)容了，希望對大家有所幫助。如果您喜歡這篇文章，可以收藏或分享給您的小伙伴們吧！歡迎持續(xù)關(guān)注我們的后續(xù)更新。

本文最后更新于 2022-12-3 14:49，某些文章具有時(shí)效性，若有錯(cuò)誤或已失效，請?jiān)诰W(wǎng)站留言或聯(lián)系站長：[email protected]

·END·

站長網(wǎng)微信號：w17tui，關(guān)注站長、創(chuàng)業(yè)、關(guān)注互聯(lián)網(wǎng)人 - 互聯(lián)網(wǎng)創(chuàng)業(yè)者營銷服務(wù)中心

免責(zé)聲明：本站部分文章和圖片均來自用戶投稿和網(wǎng)絡(luò)收集，旨在傳播知識，文章和圖片版權(quán)歸原作者及原出處所有，僅供學(xué)習(xí)與參考，請勿用于商業(yè)用途，如果損害了您的權(quán)利，請聯(lián)系我們及時(shí)修正或刪除。謝謝！

下一篇：Tomcat中修改server.xml和content.xml后自動(dòng)還原如何解決上一篇：一次tomcat源碼啟動(dòng)控制臺(tái)中文亂碼的調(diào)試過程記錄

17站長網(wǎng)微信二維碼

始終以前瞻性的眼光聚焦站長、創(chuàng)業(yè)、互聯(lián)網(wǎng)等領(lǐng)域，為您提供最新最全的互聯(lián)網(wǎng)資訊，幫助站長轉(zhuǎn)型升級，為互聯(lián)網(wǎng)創(chuàng)業(yè)者提供更加優(yōu)質(zhì)的創(chuàng)業(yè)信息和品牌營銷服務(wù)，與站長一起進(jìn)步！讓互聯(lián)網(wǎng)創(chuàng)業(yè)者不再孤獨(dú)！

掃一掃，關(guān)注站長網(wǎng)微信

精品免费在线观看-精品欧美-精品欧美成人bd高清在线观看-精品欧美高清不卡在线-精品欧美日韩一区二区

Tomcat使用線程池處理遠(yuǎn)程并發(fā)請求的方法

ThreadPoolExecutor

ThreadPoolExecutor在Tomcat中http請求的應(yīng)用

原子操作數(shù)據(jù)

阻塞和喚醒線程

大家都在看

相關(guān)分類

熱門排行

最近更新