Tomcat如何使用執行緒池處理遠程並發請求

2020 年 12 月 24 日
筆記
JAVA, tomcat

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(unnable)

初始化

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

Tags: JAVA tomcat