Effective Java 第三版——81. 優先使用併發實用程序替代wait和notify

Tips
書中的源代碼地址：https://github.com/jbloch/effective-java-3e-source-code
注意，書中的有些代碼裏方法是基於Java 9 API中的，因此JDK 最好下載 JDK 9以上的版本。

81. 優先使用併發實用程序替代wait和notify

本書的初版專門用一個條目來介紹正確使用wait和notify方法[Bloch01，Item 50]。 它的建議仍然有效，並在本條目末尾進行了總結，但這個建議遠不如之前那麼重要了。 這是由於沒有太多理由再使用wait和notify了。 從Java 5開始，該平臺提供了更高級別的併發實用程序，能夠執行之前必須在wait和notify時手動編寫代碼的各類操做。 鑑於正確使用wait和notify的困難，應該使用更高級別的併發實用程序。

java.util.concurrent包中的高級實用程序分爲三類：Executor Framework，在條目 80中簡要介紹了它；併發集合（concurrent collections） 和同步器（synchronizers）。 本條目簡要介紹後二者。

併發集合是標準集合接口（如List，Queue和Map）的高性能併發實現。 爲了提供高併發性，這些實如今內部管理本身的同步（條目 79）。 所以，不可能從併發集合中排除併發活動; 鎖定它只會使程序變慢。

由於不能排除併發集合上的併發活動，因此也不能以原子方式組合對它們的方法調用。 所以，併發集合接口配備了依賴於狀態的修改操做，這些操做將幾個基本操做組合成單個原子操做。 事實證實，這些操做對併發集合很是有用，它們使用默認方法（條目 21）添加到Java 8中相應的集合接口中。

例如，Map的putIfAbsent(key, value)方法插入鍵的映射（若是不存在）並返回與鍵關聯的以前的值，若是沒有則返回null。
這樣能夠輕鬆實現線程安全的規範化Map。 此方法模擬String.intern`方法的行爲：

// Concurrent canonicalizing map atop ConcurrentMap - not optimal
private static final ConcurrentMap<String, String> map =
        new ConcurrentHashMap<>();

public static String intern(String s) {
    String previousValue = map.putIfAbsent(s, s);
    return previousValue == null ? s : previousValue;
}

事實上，你能夠作得更好。ConcurrentHashMap針對get等檢索操做進行了優化。所以，只有在get代表有必要時，才首先調用get並調用putIfAbsent方法:

// Concurrent canonicalizing map atop ConcurrentMap - faster!
public static String intern(String s) {
    String result = map.get(s);
    if (result == null) {
        result = map.putIfAbsent(s, s);
        if (result == null)
            result = s;
    }
    return result;
}

除了提供出色的併發性外，ConcurrentHashMap很是快。 在個人機器上，上面的intern方法比String.intern快6倍（但請記住，String.intern必須採用一些策略來防止在長期運行的應用程序中泄漏內存）。 併發集合使基於同步的集合在很大程度上已通過時了。 例如，使用ConcurrentHashMap優先於Collections.synchronizedMap。 簡單地用併發Map替換同步Map以顯着提升併發應用程序的性能。

一些集合接口使用阻塞操做進行擴展，這些操做等待（或阻塞）直到能夠成功執行。 例如，BlockingQueue擴展了Queue並添加了幾個方法，包括take，它從隊列中刪除並返回head元素，等待隊列爲空。 這容許阻塞隊列用於工做隊列（也稱爲生產者——消費者隊列），一個或多個生產者線程將工做項入隊，而且一個或多個消費者線程從哪一個隊列變爲可用時出隊並處理項目。 正如所指望的那樣，大多數ExecutorService實現（包括ThreadPoolExecutor）都使用BlockingQueue（條目 80）。

同步器是使線程可以彼此等待的對象，容許它們協調各自的活動。 最經常使用的同步器是CountDownLatch和Semaphore。 不太經常使用的是CyclicBarrier和Exchanger。 最強大的同步器是Phaser。

倒計時鎖存器（CountDownLatch）是一次性使用的屏障，容許一個或多個線程等待一個或多個其餘線程執行某些操做。 CountDownLatch的惟一構造方法接受一個int類型的參數，它是在容許全部等待的線程繼續以前，必須在latch上調用countDown方法的次數。

在這個簡單的原語上構建有用的東西很是容易。例如，假設想要構建一個簡單的框架來爲一個操做的併發執行計時。這個框架由一個方法組成，該方法使用一個執行器executor來執行操做，一個表示要併發執行的操做數量併發級別，以及一個表示該操做的runnable組成。全部工做線程都準備在計時器線程啓動時鐘以前運行操做。當最後一個工做線程準備好運行該操做時，計時器線程「發號施令（fires the starting gun）」，容許工做線程執行該操做。一旦最後一個工做線程完成該操做，計時器線程就中止計時。在wait和notify的基礎上直接實現這種邏輯至少會有點麻煩，可是在CountDownLatch的基礎上實現起來卻很是簡單:

// Simple framework for timing concurrent execution
public static long time(Executor executor, int concurrency,
            Runnable action) throws InterruptedException {
    CountDownLatch ready = new CountDownLatch(concurrency);
    CountDownLatch start = new CountDownLatch(1);
    CountDownLatch done  = new CountDownLatch(concurrency);

    for (int i = 0; i < concurrency; i++) {
        executor.execute(() -> {
            ready.countDown(); // Tell timer we're ready
            try {
                start.await(); // Wait till peers are ready
                action.run();
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            } finally {
                done.countDown();  // Tell timer we're done
            }
        });
    }
    ready.await();     // Wait for all workers to be ready
    long startNanos = System.nanoTime();
    start.countDown(); // And they're off!
    done.await();      // Wait for all workers to finish
    return System.nanoTime() - startNanos;
}

請注意，該方法使用三個倒計時鎖存器。 第一個ready，由工做線程來告訴計時器線程什麼時候準備就緒。 工做線程而後等待第二個鎖存器，即start。 當最後一個工做線程調用ready.countDown時，計時器線程記錄開始時間並調用start.countDown，容許全部工做線程繼續。 而後，計時器線程等待第三個鎖存器完成，直到最後一個工做線程完成運行並調用done.countDown。 一旦發生這種狀況，計時器線程就會喚醒並記錄結束時間。

還有一些細節值得注意。傳遞給time方法的executor必須容許建立至少與給定併發級別相同數量的線程，不然測試將永遠不會結束。這被稱爲線程飢餓死鎖（thread starvation deadlock）[Goetz06, 8.1.1]。若是工做線程捕捉到InterruptedException異常，它使用習慣用法thread.currentthread ().interrupt()從新斷言中斷，並從它的run方法返回。這容許執行程序按照它認爲合適的方式處理中斷。System.nanoTime用於計算活動的時間。**對於間隔計時，請始終使用System.nanoTime而不是System.currentTimeMillis。 System.nanoTime更準確，更精確，不受系統實時時鐘調整的影響。最後，請注意，本例中的代碼不會產生準確的計時，除非action作了至關多的工做，好比一秒鐘或更長時間。準確的微基準測試是很是困難的，最好是藉助諸如jmh [JMH]這樣的專業框架來完成。

這個條目只涉及使用併發實用程序作一些皮毛的事情。 例如，前一個示例中的三個倒計時鎖存器能夠由單個CyclicBarrier或Phaser實例替換。 結果代碼會更簡潔，但可能更難理解。

雖然應該始終優先使用併發實用程序來等替換wait和notify方法，但你可能必須維護使用wait和notify的舊代碼。 wait方法用於使線程等待某些條件。 必須在同步區域內調用它，該區域鎖定調用它的對象。 下面是使用wait方法的標準習慣用法：

// The standard idiom for using the wait method
synchronized (obj) {
    while (<condition does not hold>)
        obj.wait(); // (Releases lock, and reacquires on wakeup)
    ... // Perform action appropriate to condition
}

始終要在循環中調用wait方法；永遠不要在循環以外調用它。循環用於測試wait先後的條件。

若是條件已經存在，則在wait以前測試條件並跳過等待以確保活性（liveness）。 若是條件已經存在而且在線程等待以前已經調用了notify（或notifyAll）方法，則沒法保證線程將從等待中喚醒。

爲了確保安全，須要在等待以後再測試條件，若是條件不成立，則再次等待。若是線程在條件不成立的狀況下繼續執行該操做，它可能會破壞由鎖保護的不變式（invariant）。當條件不成立時，如下幾個緣由能夠把線程喚醒：

• 另外一個線程能夠得到鎖並在線程調用notify和等待線程醒來之間改變了保護狀態。
• 當條件不成立時，另外一個線程可能意外地或惡意地調用notify方法。類經過等待公共可訪問的對象來暴露本身。公共可訪問對象的同步方法中的任何wait方法都容易受到這個問題的影響。
• 通知線程在喚醒等待線程時可能過於「慷慨」。例如，即便只有一些等待線程的知足條件，通知線程也可能調用notifyAll。
• 在沒有通知的狀況下，等待的線程可能(不多)被喚醒。這被稱爲虛假的喚醒（spurious wakeup）[POSIX, 11.4.3.6.1;Java9-api]。

一個相關的問題是，爲了喚醒等待的線程，是使用notify仍是notifyAll。(回想一下notify喚醒一個等待線程，假設存在這樣一個線程，notifyAll喚醒全部等待線程)。有時人們會說，應該始終使用notifyAll。這是合理的、保守的建議。它老是會產生正確的結果，由於它保證喚醒全部須要被喚醒的線程。可能還會喚醒其餘一些線程，但這不會影響程序的正確性。這些線程將檢查它們正在等待的條件，若是發現爲false，將繼續等待。

做爲一種優化，若是全部線程都在等待相同的條件，而且每次只有一個線程能夠從條件變爲true中喚醒，那麼能夠選擇調用notify而不是notifyAll。

即便知足了這些先決條件，也可能有理由使用notifyAll來代替notify。正如將wait方法調用放在循環中能夠防止公共訪問對象上的意外或惡意通知同樣，使用notifyAll代替notify能夠防止不相關線程的意外或惡意等待。不然，這樣的等待可能會「吞下」一個關鍵通知，讓預期的接收者無限期地等待。

總之，與java.util.concurrent提供的高級語言相比，直接使用wait和notify就像在「併發彙編語言」中編程同樣。在新代碼中基本上不存在使用wait和notify的理由。 若是正在維護使用wait和notify的代碼，請確保它始終使用標準慣用法在while循環內調用wait方法。 一般應優先使用notifyAll方法進行通知。 若是使用notify，必須很是當心以確保程序的活性。