您现在的位置：程式碼之家 >> 編程語言 > >> Java 程式設計 >> Java 相關知識

Java 多執行緒基礎總結（九）：Mina 窺探 (1)

編輯：Java 相關知識

一直以來的多執行緒基礎總結都是脫離應用的，但是要說多執行緒的應用就不能不說 Mina。Apache Mina 作為一個高效能的 Java 非同步併發網路通訊框架，其內部的多執行緒設計和實作可謂是學習多執行緒的良藥。手上的 Mina 原始碼是 svn 剪下來的最新的程式碼，mvn 轉換成 eclipse 專案後導入 mina-core 的原始碼看看多執行緒的應用吧。

首先簡單介紹在 org.apache.mina.core.service 套件裡的核心介面之一：IoService。這個介面是對於伺服器端接收連接和客戶端發起連接這兩種服務的頂層抽象，所以就有了 IoAcceptor 和 IoConnector 兩個子介面的繼承與隔離。很拍馬屁的說，從這個小細節可以看到 mina 架構的精細。這種程度的介面隔離最重要的就是對介面內抽象行為的準確劃分，錯誤的劃分介面的職責將使得後面得實作顯得不合理甚至是錯誤。只是不知道負責抽象的設計人員是否是一次性的抽象成功，如果是只能說很強。至於互動會話 IoSession 和實際的 I/O 操作處理器 IoProcessor 以及底層處理 I/O 事件的 IoHandle 這些介面就不廢話了，今天看的是與 IoServiceListener 有關的多執行緒應用。IoServiceListener 主要是用來監聽 IoService 相關的事件，而今日主角——IoServiceListenerSupport 則是用來把 IoService 和對應的 IoServiceListener 包裝在一起進行管理的輔助類別。先看看其原始碼：

public class IoServiceListenerSupport {
    /** The {@link IoService} that this instance manages. */
    private final IoService service;

    /** A list of {@link IoServiceListener}s. */
    private final List<IoServiceListener> listeners = new CopyOnWriteArrayList<IoServiceListener>();

    /** Tracks managed sessions. */
    private final ConcurrentMap<Long, IoSession> managedSessions = new ConcurrentHashMap<Long, IoSession>();

    /** Read only version of {@link #managedSessions}. */
    private final Map<Long, IoSession> readOnlyManagedSessions = Collections.unmodifiableMap(managedSessions);

    private final AtomicBoolean activated = new AtomicBoolean();

    /** Time this listenerSupport has been activated */
    private volatile long activationTime;

    /** A counter used to store the maximum sessions we managed since the listenerSupport has been activated */
    private volatile int largestManagedSessionCount = 0;

    /** A global counter to count the number of sessions managed since the start */
    private volatile long cumulativeManagedSessionCount = 0;

    /**
     * Adds a new listener.
     * @param listener The added listener
     */
    public void add(IoServiceListener listener) {
        if (listener != null) {
            listeners.add(listener);
        }
    }

    /**
     * @return true if the instance is active
     */
    public boolean isActive() {
        return activated.get();
    }

    /**
     * Calls {@link IoServiceListener#serviceActivated(IoService)}
     * for all registered listeners.
     */
    public void fireServiceActivated() {
        if (!activated.compareAndSet(false, true)) {
            // The instance is already active
            return;
        }

        activationTime = System.currentTimeMillis();

        // Activate all the listeners now
        for (IoServiceListener listener : listeners) {
            try {
                listener.serviceActivated(service);
            } catch (Throwable e) {
                ExceptionMonitor.getInstance().exceptionCaught(e);
            }
        }
    }

    /**
     * Calls {@link IoServiceListener#sessionCreated(IoSession)} for all registered listeners.
     * @param session The session which has been created
     */
    public void fireSessionCreated(IoSession session) {
        boolean firstSession = false;

        if (session.getService() instanceof IoConnector) {
            synchronized (managedSessions) {
                firstSession = managedSessions.isEmpty();
            }
        }
        ...

        cumulativeManagedSessionCount++;
        ...
    }
}

這裡為了說明多執行緒的應用，我僅僅節選相關的方法和方法片段。從這個類別中我們還是能看到豐富的多執行緒應用的。下面就開始這盤菜吧！

全域變數與併發集合分析

首先從全域變數開始看看：CopyOnWriteArrayList、ConcurrentMap、AtomicBoolean、volatile 堪稱多執行緒小動員了。後兩者之前都有介紹暫時省略，後面的方法分析時會看到使用情境。前兩者都是 java.util.concurrent 套件下的併發集合框架成員。ConcurrentMap 是高效的併發 Map 實作，主要是採取分段加鎖的機制，預設是 16 段鎖，所以多執行緒的競爭大大的降低。只有 key 的 hash 分布在同一段位上執行緒之間存在競爭。

而 CopyOnWriteArrayList 是 ArrayList 的一個執行緒安全的變體，變態之處在於對其所有的可變操作都是對底層的陣列進行一次新的複製。看看其實作原來是使用 ReentrantLock 來保證可變時的執行緒安全，又多了一個多執行緒的成員啊。當然使用這個併發集合實作是需要特殊的情況的（要知道它的每次的可變都是全盤複製，這意味著很大的效能成本）：如果遍歷集合的操作次數大大超過了可變操作時，這時候它的效能優勢就體現出來了。因為內部的陣列使用的是 volatile 的，所以遍歷查找時都不用同步而能保證可見性，這是鎖同步無法比擬的。要說 mina 重視效能從這兒可見一斑，畢竟每個全域變數都是為了效能去選擇相應的同步機制，而不是 synchronized 通吃天下。

IoServiceListenerSupport 的 fireServiceActivated()、fireServiceDeactivated()、fireSessionCreated()、fireSessionDestroyed() 方法均有遍歷 listeners 的操作，如果這些方法在執行緒之間頻繁使用的話，那麼無疑使用 CopyOnWriteArrayList 是個很好的解決方案。managedSessions 主要是管理 IoSession 的，需要使用併發 Map 的資料結構，那麼 ConcurrentMap 無疑已被證明是相對出色的併發 Map。activated 類似一個開關的設計，看看為什麼使用無鎖得 AtomicBoolean？

AtomicBoolean 的無鎖優勢

public void fireServiceActivated() {
    if (!activated.compareAndSet(false, true)) {
        // The instance is already active
        return;
    }
    ...
}

這裡可以看到 activated 的狀態改變是要依賴其原來的值的，也就是如果使用 volatile 的話，要判斷之前的是否為 false，不滿足則置為 true。這樣的操作對於 volatile 只能使用鎖同步實作。而 AtomicBoolean 的「CAS」輕鬆地使用無鎖同步原語解決了這個問題。compareAndSet(...) 不管有多少個執行緒執行，只有取得 activated 最新值的執行緒才能返回 true，其餘的都會是 false。這就看到 mina 的 committer 還是很糾結效能的。

volatile 變數分析

然後看看那幾個 volatile 的變數吧。首先是 activationTime，這個變數除了有自身的 get() 可以隨時取得最新的值之外，就在 fireServiceActivated() 內出現了：activationTime = System.currentTimeMillis(); 可以看到僅僅是一次性的賦值而且不依賴其自身的值。所以完全滿足執行緒安全的條件。largestManagedSessionCount 的使用和其類似。

再來看看 cumulativeManagedSessionCount，它是一個全域的計數器，負責記錄啟動後所管理得會話數量。除了同樣有 get() 方法之外就是出現在 fireSessionCreated(...) 方法中：cumulativeManagedSessionCount++; 這引起了我的注意，因為這樣的寫法是不能保證執行緒同步的！因為 volatile 的變數根本無法完成「++」的原子操作。「++」是需要依賴其自身的值而進行更改的操作。為此我簡單的寫了個驗證的範例證明了這個鐵律。

驗證 volatile 無法保證 ++ 原子性

public class VolatileTest {
    public static void main(String[] args) {
        final VolatileSample sample = new VolatileSample();
        Runnable runnable = new Runnable() {
            public void run() {
                for(int i = 0; i < 500; i++) {
                    System.out.println(sample.incrementAndGet());
                }
            }
        };
        for(int i = 0; i < 50; i++) {
            new Thread(runnable).start();
        }
    }
}

class VolatileSample {
    private volatile long counter = 0;
    public long incrementAndGet() {
        counter++;
        return get();
    }
    public long get() {
        return counter;
    }
}

使用 50 個執行緒每個執行緒執行 500 次對 counter 進行疊加。正確的答案應該是 25000，但是很遺憾地發現僅僅測試幾次就出現了 24999 和其他的小於 25000 的最大值。所以很遺憾地發現 mina 的這個變數的方案選擇是有問題的！

這裡應該是使用的是 AtomicLong，而對應的 fireServiceActivated(...) 內的程式碼如果不使用 API 提供的話應該是：

for (;;) {
    long current = cumulativeManagedSessionCount.get();
    long next = current + 1;
    if (compareAndSet(current, next))
        break;
}

雖然是 for 迴圈 + CAS 的 atomic 殺手鐧，但是不要怕，一般 for 迴圈 cpu 都是一次搞定，極少情況是多次，效能不會有什麼明顯影響。這個 bug 可能在數量較少的執行緒的情況下很難顯現，或者是 mina 的開發者考慮使用情境故意的寬鬆執行緒機制？從嚴謹性來看是我個人認為是不對的。

巧妙的 synchronized 等待實作

最後就是看看一個有趣的 synchronized 的程式碼片段：

/** Close all the sessions
 * TODO disconnectSessions.
 */
private void disconnectSessions() {
    Object lock = new Object();
    ...
    try {
        synchronized (lock) {
            while (!managedSessions.isEmpty()) {
                lock.wait(500);
            }
        }
    } catch (InterruptedException ie) {
        // Ignored
    }
}

這個方法是關閉所有的會話，但是為了減少執行緒鎖的競爭使用了一個很可愛的方式。這裡的 lock 是個區域變數，所以每個進入這個方法的執行緒都會擁有一個 lock 物件，而 synchronized(lock) 是迷惑的重點。這個 synchronized 不會使得進入方法的執行緒們產生任何的競爭，因為每個執行緒都能獲得屬於自己的 lock。synchronized 的作用就在於 lock.wait(500) 的呼叫，就因為 wait 方法必須要 synchronized 的配合，所以就出現了這個可愛的程式碼。這段程式碼的主要意圖就是不管幾個執行緒去關閉所有會話，每個執行緒都是間隔 500ms 去檢查 managedSessions 是否為空。這裡沒有對 managedSessions 使用 synchronized 的原因就是為了減少執行緒鎖對 managedSessions 的資源獨占，而改用 while 迴圈的機制寧願等待也不為了檢查 managedSessions 而影響其他執行緒的工作。很精巧的一個實作！

寫到這裡，算是這個小菜的結尾了，總感覺意猶未盡。mina 對高效能併發的目標還是從程式碼上得到了一些的體現。當然它的高效能主要是對 NIO 的封裝使用，不過這一切都是建立在多執行緒的併發基礎上的。至於之前無意發現的 bug（個人認為）如果 mina 的 committer 也有相同的認知的話，我想以後會有相應的修改的。