Java併發編程基礎

本篇博客主要是紀錄對<<Java并发编程的艺术>>的學習心得以及一些知識點的重新整理,因此文章中會有許多引用自該書的文字。同時也借鑑其他相關的技術博客,以完善整個知識框架。

線程簡介

什麼是線程

現代操作系統在運行一個程序時,會為其創建一個進程。例如,啟動一個Java程序,操作系統就會創建一個Java進程。現代操作系統調度的最小單元是線程,也叫輕量級進程(Light Weight Process),在一個進程裡可以創建多個線程,這些線程都擁有各自的技術器、堆棧和局部變量等屬性,並且能夠訪問共享的內存變量。處理器在這些線程上高速切換,讓使用者感覺到這些線程在同時執行。

詳情可見本博客關於操作系統的文章:

https://jack870131.github.io/2020/04/19/%E9%80%B2%E7%A8%8B%E8%88%87%E7%B7%9A%E7%A8%8B/

為什麼要使用多線程

  1. 更多的處理器核心: 線程是大多數操作系統調度的基本單元,一個程序作為一個進程來運行,程序運行過程中能創建多個線程,而一個線程在一個時刻只能運行在一個處理器核心上。是想一下,一個單線程程序在運行時只能使用一個處理器核心,那麼再多的處理器核心加入也無法顯著提升該程序的執行效率。相反,如果該程序使用多線程技術,將計算邏輯分配到多個處理器核心上,就會顯著減少程序的處理時間,並且隨著更多處理器核心的加入而變得更有效率。
  2. 更快的響應時間: 多線程技術可以將數據一致性不強的操作派發給其他線程處理(也可以使用消息隊列),如生成訂單快照、發送郵件等。這樣做的好處是響應用戶請求的線程能夠盡可能快遞處理完成,縮短了響應時間,提升用戶體驗。
  3. 更好的編程模型: Java為多線程編程提供了良好且一致的編程模型,使開發人員能夠更加專注於問題解決,而不是去考慮如何將其多線程化。

線程優先級

現代操作系統基本採用時分的形式調度運行的線程,操作系統會分出一個個時間片,線程會分配到若干時間片,當線程的時間片用完了就會發生線程調度,並等待下一次的分配。線程分配到的時間片決定了線程使用處理器資源的多少,而線程優先級決定線程需要多或少分配一些除禮器資源的線程屬性。

在Java線程中,通過一個整型城元變量priority來控制優先級,優先級的範圍從1~10,在線程構建的時候可以通過setPriority(int)方法來修改,默認優先級是5,優先級高的線程分配時間片的數量要多於優先級低的線程。設置優先級時,針對頻繁阻塞(休眠或者I/O操作)的線程需要設置較高優先級,而偏重計算(需要較多CPU時間或者偏運算)的線程則設置較低優先級,確保處理器不會被獨佔。

線程的狀態

Java線程運行的生命週期有如下圖,6種不同的狀態,在給定的一個時刻,線程只能處於其中的某一個狀態。

狀態名稱 說明
NEW 初始狀態,線程被構建,但是還沒有調用start()方法
RUNNABLE 運行狀態,Java線程將操作系統中的就緒和運行兩種狀態籠統地稱作”運行中”
BLOCKED 阻塞狀態,表示線程阻塞於鎖
WAITING 等待狀態,表示線程進入等待狀態,進入該狀態表示當前線程需要等待其他線程做出一些特定動作(通知或中斷)
TIME_WAITING 超時等待狀態,該狀態不同於WAITING,它是可以在指定的時間自行返回的
TERMINATED 終止狀態,表示當前線程已經執行完畢

線程在自身的生命周期中,並不是固定地處在某個狀態,而是隨著代碼的執行在不同的狀態之間進行切換,Java線程狀態變遷如下圖所示:

Java的線程狀態變遷

由上圖可以看到,線程創建以後,調用start()方法開始運行。當線程執行wait()方法之後,線程進入等待狀態。進入等待狀態的線程需要依靠其他線程的通知才能夠返回到運行狀態,而超時等待狀態相當於在等待狀態的基礎上增加了超時限制,也就是超時時間到達時將會返回到運行狀態。當線程調用同步法時,在沒有獲取到鎖的情況下,線程將會進入到阻塞狀態。線程在執行Runnable的run()方法之後會進入到終止狀態。

Daemon 線程

Daemon線程是一種支持型線程,因為它主要被用作程序中後台調度以及支持性工作。這意味著,當一個Java虛擬機中不存在非Daemon線程的時候,Java虛擬機將會退出。可以通過調用Thread.setDaemon(true)將線程設置為Daemon線程(需要在啟動線程之前設置)。

Daemon線程被用作完成支持性工作,但是在Java虛擬機退出時Daemon線程中的finally塊並不一定會執行,以下為示例代碼:

public class Daemon {
    public static void main(String[] args) {
        Thread thread = new Thread(new DaemonRunner(), "DaemonRunner");
        thread.setDaemon(true);
        thread.start();
    }
    static class DaemonRunner implements Runnable {
        @Override
        public void run() {
            try {
                SleepUtils.second(10);
            } finally {
                System.out.println("DaemonThread finally run.");
            }
        }
    }
}

運行Daemon程序,可以看到在終端或者命令行上沒有任何輸出。main線程(非Daemon線程)在啟動了線程DaemonRunner之後隨著main方法執行完畢而終止,而此時Java虛擬機中沒有非Daemon線程,虛擬機需要退出。Java虛擬機中的所有Daemon線程都需要立即終止,因此DaemonRunner立即終止,但是DaemonRunner中的finally塊並沒有執行(在構建Daemon線程時,不能依靠finally塊中的內容來確保執行關閉或清理資源的邏輯)。

啟動和終止線程

接下來將詳細介紹線程的啟動和終止。

構造線程

在運行線程之前首先要構造一個線程對象,線程對象在構造的時候需要提供線程所需要的屬性,如線程所屬的線程組、線程優先級、是否是Daemon線程等信息。以下代碼是java.lang.Thread中對線程進行初始化的部分:

private void init(ThreadGroup g, Runnable target, String name, long stackSize, AccessControlContext acc) {
    if (name == null) {
        throw new NullPointerException("name connot be null");
    }
    // 當前線程就是該線程的父線程
    Thread parent = currentThread();
    this.group = g;
    // 將daemon、priority屬性設置為父線程的對應屬性
    this.daemon = parent.isDaemon();
    this.priority = parent.isDaemon();
    this.name = name.toCharArray();
    this.target = target;
    setPriority(priority);
    // 將父線程的InheritableThreadLocal複製過來
    if (parent.inheritableThreadLocals != null) {
        this.inheritableThreadLocals = ThreadLocal.createInheritedMap(Parent.inheritableThreadLocals);
        // 分配一個線程ID
        tid = nextThreadID();
    }
}

在上述過程中,一個新構造的線程對向是由其parent線程來進行空間分配的,而child線程繼承了parent是否為Daemon、優先級和加載資源的contextClassLocader以及可繼承的ThreadLocal,同時還會分配一個為一個ID來標示這個child線程。至此,一個能夠運行的線程對向就初始化好了,在堆內存等待著運行。

啟動線程

線程對象在初始化完成之後,調用start()方法就可以啟動該線程。線程start()方法的含意是: 當前線程(即parent線程)同步告知Java虛擬機,只要線程規劃器空閒,應該立即啟動調用start()方法的線程。

中斷

中斷可以理解為線程的一個標示位屬性,它表示一個運行中的線程是否被其他線程進行了中斷操作。中斷好比其他線程對該線程打了個招呼,其他線程通過調用該線程的interrupt()方法對其進行中斷操作。

線程通過檢查自身是否被中斷來進行響應,麲程通過方法isInterrupted()方法來進行判斷是否被中斷,也可以調用靜態方法Thread.interrupted()對當前線程的中斷標示位進行復位。如果該線程已經處於終結狀態,即使線程被中斷過,在調用該線程對向的isInterrupted()時依舊會返回false。

從Java的API中可以看到,許多聲明拋出InterruptedException的方法(例如Thread.sleep(long millis)方法),這些方法在拋出InterruptedException之前,Java虛擬機會先將該線程的中斷標示位清除,然後拋出InterruptedEcxception,此時調用isInterrupted()方法將會返回false。

在以下代碼中,首先創建了兩個線程,SleepThread和BusyThread,前者不停地睡眠,後者一直運行,然後對這兩個線程分別進行中斷操作,觀察二者的中斷標示位。

public class Interrupted {
    public static void main(String[] args) throws Exception {
        // sleepThread不停地嘗試睡眠
        sleepThread.setDaemon(true);
        // busyThread不停的運行
        Thread busyThread = new THread(new BusyRunner(), "BusyThread");
        busyThread.setDaemon(true);
        sleepThread.start();
        busyThread.start();
        // 休眠5秒,讓sleepThread和busyThread充分運行
        TimeUnit.SECONDS.sleep(5);
        sleepThread.interrupt();
        busyThread.interrupt();
        System.out.println("SleepThread interrupted is " + sleepThread.isInterrupted());
        System.out.println("BusyThread interrupted is " + busyThread.isInterrupted());
        // 防止sleepThread和busyThread立刻退出
        SleepUtils.second(2);
    }
    static class SleepRunner implements Runnable {
        @Override
        public void run() {
            while (true) {
                SleepUtils.second(10);
            }
        }
    }
    static class BusyRunner implements Runnable {
        @Override
        public void run() {
            while (true) {}
        }
    }
}

輸出如下:

SleepThread interrupted is false
BusyThread interrupted is true

從結果可以看出,拋出InterruptedException的線程SleepThread,其中斷標示位被清除了,而一直忙碌運作的線程BusyThread,中斷標示位沒有被清除。

安全地終止線程

在上一節提到的中斷狀態是線程的標示位,而中斷操作是一種簡便的線程交互方式,而這種交互方式最適合用來取消或停止任務。除了中斷以外,還可以利用一個boolean變量來控制是否需要停止任務並終止該線程。

在以下代碼示例中,創建了一個線程CountThread,它不斷地進行變量累加,而主線程常是對其進行中斷操作和停止操作。

public class Shutdown {
    public static void main(String[] args) {
        Runner one = new Runner();
        Thread countThread = new Thread(one, "CountThread");
        countThread.start();
        // 睡眠1秒,main線程對CountThread進行中斷,使CountThread能夠感知中斷而結束
        TimeUnit.SECONDS.sleep(1);
        countThread.interrupt();
        Runner two = new Runner();
        countThread = new Thread(two, "CountThread");
        countThread.start();
        // 睡眠1秒,main線程對Runner two進行取消,使CountThread能夠感知on為false而結束
        TimeUnit.SECONDS.sleep(1);
        two.cancel();
    }
    private static class Runner implements Runnable {
        private long i;
        private volatile boolean on = true;
        @Override
        public void run() {
            while (on && !Thread.currentThread().isInterrupted()) {
                i++;
            }
            System.out.println("Count i = " + i);
        }
        public void cancel() {
            on = false;
        }
    }
}

輸出結果如下所是(輸出內容可能不同):

Count i = 543487324
Count i = 540898082

以上代碼在執行過程中,main線程通過中斷操作和cancel()方法均可使CountThread得以終止。這種通過標示為或者中斷操作的方式能夠使線程在終止時有機會去清理資源,而不是武斷地將線程停止,因此這種終止線程的作法顯得更加安全和優雅。

線程間通信

接下來介紹線程間是如何相互配合工作創造價值。

volatile和synchronized關鍵字

Java支持多個線程同時訪問一個對象或者對象的成員變量,由於每個線程可以擁有這個變量的拷貝(雖然對象以及成員變量分配的內存是在共享內存中的,但是每個執行的線程還是可以擁有一份拷貝,這樣做的目的是加速程序的執行,這是現代多核處理器的一個顯著特性),所以程序在執行過程中,一個線程的變量不一定是最新的。

關鍵字volatile可以用來修飾字段(成員變量),就是告知程序任何對該變量的訪問均需要從共享內存中獲取,而對它的改變必須同步刷新回共享內存,它能保證所有線程對變量訪問的可見性。

舉個例子,定義一個表示程序是否運行的成員變量boolean on = true,那麼另一個線程可能對它執行關閉動作(on = false),這裡涉及多個線程對變量的執行關閉動作(on = false),這裡涉及多個線程對變量訪問,因此需要將其定義為volatile boolean on = true,這樣其他線程對它進行改變時,可以讓所有線程感知道變化,因此所有對on變量的訪問和修改都需要以共享內存為準。但是過多地使用volatile是不必要的,因為它會降低程序執行的效率。

關鍵字synchronized可以修飾方法或者以同步塊的形式來進行使用,它主要確保多個線程在同一個時刻,只能有一個線程處於方法或者同步塊中,它保證了線程對變量訪問的可見性和排他性。

任意一個對象通擁有自己的監視器,當這個對象由同步塊或者這個對象的同步方法調用時,執行方法的線程必須先或取道該對象的監視器才能進入同步塊或者同步方法,而沒有獲取到監視器(執行該方法)的線程將會被阻塞在同步塊和同步方法的入口處,進入BLOCKED的狀態。

下圖描述了對象、對象監視器、同步隊列和執行線程之間的關係:

對象、監視器、同步隊列和執行線程之間的關係

從上圖可以看到,任意線程對Object(Object由synchronized保護)的訪問,首先要獲得Object的監視器。如果獲取失敗,線程進入同步隊列,現成狀態變為BLOCKED。當訪問Object的前驅(獲得了鎖的線程)釋放了鎖,則該釋放操作喚醒阻塞在同步隊列中的線程,使其重新嘗試對監視器的獲取。

等待/通知機制

一個線程修改了一個對象的值,而另一個線程感知到了變化,然後進行相應的操作,整個過程開始於一個線程,而最終執行又是另一個線程。前者是生產者,後者是消費者,這種模式隔離了”做甚麼(what)”和”怎麼做”,在功能層面上實現了解藕,體系結構上具備了良好的伸縮性。接下來介紹在Java語言中如何實現類似的功能。

簡單的辦法是讓消費者線程不斷地循環檢查變量是否符合預期,如以下代碼所示,在while循環中設置不滿足調建,如果條件滿足則退出while循環,從而完成消費者的工作。

while (value != desire) {
  Thread.sleep(1000);
}
doSomething();

上面這段代碼在條件不滿足時就睡眠一段時間,這樣做的目的是防止過快的”無效”城市,這種方式看似能夠解時線所需的功能,但卻存在如下問題:

  1. 難以確保及時性: 在睡眠時,基本不消耗處理器資源,但是如果睡得過久,就不能及時發現條件已經變化,也就是及時性難保證。
  2. 難以降低開銷: 如果降低睡眠時間,比如休眠1毫秒,這樣消費者能更加迅速地發現條件變化,但是卻可能消耗更多的處理器資源,造成無端的浪費。

Java會通過內痔的等待/通知機制很好地解決以上所說的矛盾並實現所需的功能。等待/通知的相關方法是任意Java對象都具備的,因為這些方法被定義在所有對象的超類java.lang.Object上,方法和描述如下表所示:

方法名稱 描述
notify() 通知一個在對象上等待的線程,使其從wait()方法返回,而俺回的前提是該線程火取道了對象的鎖
notifyAll() 通知所有等待在該對象上的線程
wait() 調用該方法的線程進入WAITING狀態,只有等待另外線程的通知或被中斷才會返回,需要注意,調用wait()方法後,會釋放對象的鎖
wait(long) 超時等待一段時間,這裡的參數時間是毫秒,也就是等待長達n毫秒,如果沒有通知就超時返回
wait(long, int) 對於超時時間更細粒度的控制,可以達到milisecond

等待/通知機制,是指一個線程調用了對象O的wait()方法進入等待狀態,而另一線程B調用了對象O的notifyAll()方法,線程A收到通知後從對象O的wait()方法返回,進而執行後續操作。

在以下代碼示例中,創建了兩個線程 – WaitThread和NotifyThread,前者檢查flag值是否為false,如果符合要求,進行後續操作,否則在lock上等待,後者睡眠了一段時間後對lock進行通知:

public clas WaitNotify {
  static boolean flag = true;
  static Object lock = new Object();
  public static void main(String[] args) throws Exception {
    Thread waitThread = new Thread(new Wait(), "WaitThread");
    watTHread.starty();
    TimeUnit.SECONDS.sleep(1);
    Thread notifyTHread = new Thread(new Notify(), "NotifyThread");
    notifyThread.start();
  }
  static class Wait implements Runnable {
    public void run() {
      // 加鎖,擁有lock的Monitor
      synchronized (lock) {
        // 當條件不滿足時,繼續wait,同時釋放lock的鎖
        while (flag) {
          tru {
            System.out.println(Thread.currentThread() + " flag is true. wait @ " + new SimpleDateFormat("HH:mm:ss").format(new Date()));
            lock.wait();
          } catch (InterruptedException e) {}
        }
        // 條件滿足時,完成工作
        System.out.println(Thread.currentThread() + " flag is false. running @ " + new SimpleDateFormat("HH:mm:ss") .format(new Date()));
      }
    }
  }
  static class Notify implements Runnable {
    public void run() {
      // 加鎖,擁有lock的Monitor
      sunchronized (lock) {
        // 獲取lock的鎖,然後進行通知,通知時不會釋放lock的鎖
        // 直到當前線程釋放了lock後,WaitThread才能從wait方法中返回
        System.out.println(Thread.currentThread() + " hold lock. notify @ " + new SimpleDateFormat("HH:mm:ss").format(new Date()));
        lock.notifyAll();
        flag = false;
        SleepUtils.second(5);
      }
      // 再次加鎖
      synchronized (lock) {
        Sustem.out.println(Thread.currentThread() + " hold lock again. sleep @ " + new SimpleDateFormat("HH:mm:ss").format(new Date()));
        SleepUtils.second(5);
      }
    }
  }
}

輸出:

Thread[WaitTHread,5,main] flag i true. wait @ 22:23:03
Thread[NotifyThread],5,main] hold lock. notify @ 22:23:04
Thread[NotifyThread,5,main] hold lock again. sleep @ 22:23:09
Thread[WaitThread,5,main] flag is false. running @ 22:23:14

上述第3行和第4行輸出的順序可能會互換,而上述例子主要說明了調用wait()、notify()以及notifyAll()時所需要注意的細節:

  1. 使用wait()、notify()和notifyAll()時需要先對調用對象加鎖。
  2. 調用wait()方法後,線程狀態由RUNNING變為WAITING,並將當前線程放置到對象的等待隊列。
  3. notify()或notifyAll()方法調用後,等待線程依舊不會從wait()返回,需要調用notify()或notifyAll()的線程釋放鎖之後,等待線程才有機會從wait()方法返回。
  4. notify()方法將等待隊列中的一個等待線程從等待隊列中移到同步隊列中,而notifyAll()方法則是將等待隊列中所有的線程全部移到同步隊列,被移動的線程狀態由WAITING變為BLOCKED。
  5. 從wait()方法返回的前提是獲得了調用對象的鎖。

從上述細節中可以看到,等待/通知機制依託於同步機制,其目的就是確保等待線程從wait()方法返回時能夠感知到通知現成對變量做出的修改。

下圖為上述示例的過程:

WaitNotify.java的運行過程

上途中,WaitThread首先獲取了對象的鎖,然後調用對象的wait()方法,從而放棄了鎖並進入了對象的等待隊列WaitQueue中,進入等待狀態。由於WaitThread釋放了對象的鎖,NotifyThread隨後獲取了對象的鎖,並調用對象的notify()方法,將WaitThread從WaitQueue移到SynchronizedQueue中,此時WaitThread的狀態變為阻塞狀態。NotifyThread釋放了鎖之後,WaitThread再次獲取到鎖並從wait()方法返回繼續執行。

等待/通知的經典范式

從WaitNotify勢力中可以提煉出等待/通知的經典范式,該范式分為兩部分,分別針對等待方(消費者)和通知方(生產者)。

等待方遵循如下原則:

  1. 獲取對象的鎖。
  2. 如果條件不滿足,那麼調用對象的wait()方法,被通知後仍要檢查條件。
  3. 條件滿足則執行對應的邏輯。

代碼如下:

sunchronized(對象) {
  while(條件不滿足) {
    對象.wait();
  }
  處理邏輯
}

通知方遵循如下原則:

  1. 獲得對象的鎖。
  2. 改變條件。
  3. 通知所有等待在對象上的線程。

對應的代碼如下:

synchronized(對象) {
  改變條件
  對象.notifyAll();
}

管道輸入/輸出流

管道輸入/輸出流和普通的文件輸入/輸出流或者網絡輸入/輸出流不同之處在於,它主要用於線程之間的數據傳輸,而傳輸的媒介為內存。

管道輸入/輸出流主要包括了如下4中具體實現: PipedOutputStream、PipedInputStream、PipedReader和PipedWriter,前兩種面向字節,而後兩種面向字符。

ThreadLocal的使用

ThreadLocal,即線程變量、是一個以ThreadLocal對象為鍵、任意對象為值的存儲結構。這個結構被附帶在線程上,也就是說一個線程可以根據一個ThreadLocal對象查詢到綁訂在這個線程上的一個值。

可以通過set(T)方法來設置一個值,在當前線程下再通過get()方法獲取到原先設置的值。

在以下示例中,構建了一個常用的Profiler類,它具有begin()和end()兩個方法,而end()方法返回從begin()方法調用開始到end()方法被調用時的時間差,單位是毫秒:

public class Profiler {
  // 第一次get()方法調用時會進行初始化(如果set方法沒有調用),每個線程會調用一次
  private static final ThraedLocal<Long> TIME_THREADLOCAL = new ThreadLocal<Long>() {
    protected Long initialValue() {
      return System.currentTimeMillis();
    }
  };
  public static final void begin() {
    TIME_THREADLOCAL.set(System.currentTimeMillis());
  }
  public static final void end() {
    return System.currentTimeMillis() - TIME_THREADLOCAL.get();
  }
  public static void main(String[] args) throws Exception {
    Profiler.begin();
    TimeUnit.SECONDS.sleep(1);
    System.out.println("Cost: " + Profiler.end() + " milis");
  }
}

輸出:

Cost: 1001 mills

Profiler可以被複用在方法調用耗時統計的功能上,在方法的入口前執行begin()方法,在方法調用後進行end()方法,好處是兩個方法的調用不用在一個方法或者類中,比如在AOP(面向方面編程)中,可以在方法調用前的切入點執行begin()方法,而在方法調用後的切入點執行end()方法,這樣依舊可以獲得方法的執行耗時。

線程應用

等待超時模式

開發人員經常會遇到這樣的方法調用場景: 調用一個方法時等待一段時間(一般來說是給定一個時間段),如果該方法能夠在給定的時間段內得到結果,那麼將結果立刻返回,反之,超時返回默認結果。

前面的章節介紹了等待/通知的經典范式,以及加鎖、條件循環和處理邏輯3個步驟,而這種范式無法做到超時等待。而超時等待的加入,只需要對經典范式做出非常小的改動,改動內容如下所示:

假設超時時間段是T,那麼可以推斷出在當前時間now + T之後就會超時。

定義如下變量:

  • 等待持續時間: REMAINING=T。
  • 超時時間: FUTURE=now=T。

這時僅需要wait(REMAINING)即可,在wait(REMAINING)返回之後將會執行REMAINING=FUTURE-now。如果REMAINING小於等於0,表示已經超時,直接退出,否則將繼續執行wait(REMAINING)。

// 對當前對象加鎖
public synchronized Object get(long mills) throws InterruptedException {
  long future = System..currentTimeMillis() + millis;
  long remaining = mills;
  // 當超時大於0並且result返回值不滿足要求
  while ((result == null) && remaining > 0) {
    wait (remaining);
    remaining = future - System.currentTimeMillis();
  }
  return result;
}

可以看出,等待超時模式就是在等待/通知范式基礎上增加了超時控制,這使得該模式相比原有范式更具有靈活性,因為即使方法執行時間過長,也不會”永久”阻塞調用者,而是會按照調用者的要求”按時”返回。

數據庫連接池

我們使用等待超時模式來構造一個簡單的數據庫連接池,在示例中模擬從連接池中獲取、使用和釋放連接的過程,而客戶端獲取連接的過程被設定為等待超時的模式,也就是在1000毫秒內無法獲取到可用連接,將會返回給客戶端一個null。設定連接池的大小為10個,然後通過調節客戶端的線程數來模擬無法獲取連接的場景。

首先看下連接池的定義。它通過構造函數初始化連接的最大上限,通過一個雙向隊列來維護連接,調用方需要先調用fetchConnection(long)方法來只訂在多少毫秒內超活獲取連接,當連接使用完成後,需要調用releaseConnection(Connection)方法將連接放回線程池,代碼示例如下:

public class ConnectionPool {
  private LinkedList<Connection> pool = new LinkedList<Connection>();
  public ConnectionPool(int initialSize) {
    if (initialSize > 0) {
      for (int i = 0; i < initialSize; i++) {
        pool.addLast(ConnectionDriver.createConnection());
      }
    }
  }
  public void releaseConnection(Connection connection) {
    if (connection != null) {
      sunchronized (pool) {
        // 連接釋放後需要進行通知,這樣其他消費者能夠感知道連接池中已經歸還了一個連接
        pool.addLast(connection);
        pool.notifyAll();
      }
    }
  }
  // 在mills內無法獲取道連接,將會返回null
  public Connection fetchConnection(long mills) throws InterruptedException {
    synchronized (pool) {
      // 完全超時
      if (mills <= 0) {
        while (pool.isEmpty()) {
          pool.wait();
        }
        return pool.removeFirst();
      } else {
        long future = Sustem.currentTimeMillis() + mills;
        long remaining = mills;
        while (pool.isEmpty() && remaining > 0) {
          pool.wait(remaining);
          remaining = future - System.currentTimeMillis();
        }
        Connection result = null;
        if (!pool.isEmpty()) {
          result = pool.removeFirst();
        }
        return result;
      }
    }
  }
}

由於java.sql.Connection是一個接口,最終的實現是由數據庫驅動提供方來實現的,考慮到只是個示例,我們通過動態代理構造了一個Connection,該Connection的代理實現僅僅是在commit()方法調用時休眠100毫秒,代碼如下所示:

public class ConnectionDriver {
  static class ConnectionHandler implements InvocationHandler {
    public Object invoke(Object proxy, Method method, Object[] args) throws Exception {
      if (method.getName().equals("commit")) {
        TimeUnit.MILLISECONDS.sleep(100);
      }
      return null;
    }
  }
  // 創建一個Connection代理,在commit時休眠100毫秒
  public static final Connection createConnection() {
    return (Connection) Proxy.newProxyInstance(ConnectionDriver.class.getClasLoader(), new class<>[] { Connection.class }, new ConnectionHandler());
  }
}

線程池技術

對於服務端的程序,經常面對的是客戶端傳入的短小(執行時間短、工作內容較為單一)任務,需要服務端快速並返回結果。如果服務端每次接受到一個任務,創建一個線程,然後進行執行,這在圓形階段是個不錯的選擇,但是面對成千上萬的任務遞交進服務器時,如果還是採用一個任務一個線程的方式,那麼將會創建數以萬計的線程,這不是一個好的選擇。因為這會使操作系統頻繁的進行線程上下切換,無故增加系統的負載,而線程的創建和消亡都是需要耗費系統資源的,也無疑浪費了系統資源。

線程池技術能夠很好地解決這個問題,它預先創建了若干數量的線程,並且不能由用戶直接對線程的創建進行控制,在這個前提下重複使用固定或較為固定數目的線程來完成任務的執行。這樣做的好處是,一方面,消除了頻繁創建和消亡線程的系統資源開銷,另一方面,面對過量任務的提交能夠平緩的劣化。

線程池的本質就是使用了一個線程安全的工作隊列連接工作者線程和客戶端線程,客戶端線程放入工作隊列後便返回,而工作者線程則不斷地從工作隊列上取出工作並執行。當工作隊列為空時,所有的工作者線程均等待再工作隊列上,當有客戶端提交了一個任務之後會通知任意一個工作者線程,隨著大量的任務被提交,更多的工作者線程會被喚醒。