Apache MINA 2 是一個(gè)開發(fā)高性能和高可伸縮性網(wǎng)絡(luò)應(yīng)用程序的網(wǎng)絡(luò)應(yīng)用框架。它提供了一個(gè)抽象的事件驅(qū)動的異步 API，可以使用 TCP/IP、UDP/IP、串口和虛擬機(jī)內(nèi)部的管道等傳輸方式。Apache MINA 2 可以作為開發(fā)網(wǎng)絡(luò)應(yīng)用程序的一個(gè)良好基礎(chǔ)。下面將首先簡單介紹一下 Apache MINA 2。

Apache MINA 2 介紹

Apache MINA 是 Apache 基金會的一個(gè)開源項(xiàng)目，目前最新的版本是 2.0.0-RC1。本文中使用的版本是 2.0.0-M6。從 參考資料 中可以找到相關(guān)的下載信息。下面首先介紹基于 Apache MINA 的網(wǎng)絡(luò)應(yīng)用的一般架構(gòu)。

基于 Apache MINA 的網(wǎng)絡(luò)應(yīng)用的架構(gòu)

基于 Apache MINA 開發(fā)的網(wǎng)絡(luò)應(yīng)用，有著相似的架構(gòu)。圖 1 中給出了架構(gòu)的示意圖。

如 圖 1 所示，基于 Apache MINA 的網(wǎng)絡(luò)應(yīng)用有三個(gè)層次，分別是 I/O 服務(wù)、I/O 過濾器和 I/O 處理器：

• I/O 服務(wù)：I/O 服務(wù)用來執(zhí)行實(shí)際的 I/O 操作。Apache MINA 已經(jīng)提供了一系列支持不同協(xié)議的 I/O 服務(wù)，如 TCP/IP、UDP/IP、串口和虛擬機(jī)內(nèi)部的管道等。開發(fā)人員也可以實(shí)現(xiàn)自己的 I/O 服務(wù)。
• I/O 過濾器：I/O 服務(wù)能夠傳輸?shù)氖亲止?jié)流，而上層應(yīng)用需要的是特定的對象與數(shù)據(jù)結(jié)構(gòu)。I/O 過濾器用來完成這兩者之間的轉(zhuǎn)換。I/O 過濾器的另外一個(gè)重要作用是對輸入輸出的數(shù)據(jù)進(jìn)行處理，滿足橫切的需求。多個(gè) I/O 過濾器串聯(lián)起來，形成 I/O 過濾器鏈。
• I/O 處理器：I/O 處理器用來執(zhí)行具體的業(yè)務(wù)邏輯。對接收到的消息執(zhí)行特定的處理。

創(chuàng)建一個(gè)完整的基于 Apache MINA 的網(wǎng)絡(luò)應(yīng)用，需要分別構(gòu)建這三個(gè)層次。Apache MINA 已經(jīng)為 I/O 服務(wù)和 I/O 過濾器提供了不少的實(shí)現(xiàn)，因此這兩個(gè)層次在大多數(shù)情況下可以使用已有的實(shí)現(xiàn)。I/O 處理器由于是與具體的業(yè)務(wù)相關(guān)的，一般來說都是需要自己來實(shí)現(xiàn)的。

事件驅(qū)動的 API

Apache MINA 提供的是事件驅(qū)動的 API。它把與網(wǎng)絡(luò)相關(guān)的各種活動抽象成事件。網(wǎng)絡(luò)應(yīng)用只需要對其感興趣的事件進(jìn)行處理即可。事件驅(qū)動的 API 使得基于 Apache MINA 開發(fā)網(wǎng)絡(luò)應(yīng)用變得比較簡單。應(yīng)用不需要考慮與底層傳輸相關(guān)的具體細(xì)節(jié)，而只需要處理抽象的 I/O 事件。比如在實(shí)現(xiàn)一個(gè)服務(wù)端應(yīng)用的時(shí)候，如果有新的連接進(jìn)來，I/O 服務(wù)會產(chǎn)生 sessionOpened這樣一個(gè)事件。如果該應(yīng)用需要在有連接打開的時(shí)候，執(zhí)行某些特定的操作，只需要在 I/O 處理器中此事件處理方法 sessionOpened中添加相應(yīng)的代碼即可。

在介紹 Apache MINA 中的基本概念的細(xì)節(jié)之前，首先通過一個(gè)簡單的應(yīng)用來熟悉上面提到的三個(gè)層次的具體職責(zé)。

回頁首

從簡單應(yīng)用開始

在使用 Apache MINA 開發(fā)復(fù)雜的應(yīng)用之前，首先將介紹一個(gè)簡單的應(yīng)用。通過此應(yīng)用可以熟悉上面提到的三個(gè)層次，即 I/O 服務(wù)、I/O 過濾器和 I/O 處理器。該應(yīng)用是一個(gè)簡單的計(jì)算器服務(wù)，客戶端發(fā)送要計(jì)算的表達(dá)式給服務(wù)器，服務(wù)器返回計(jì)算結(jié)果。比如客戶端發(fā)送 2+2，服務(wù)器返回 4.0作為結(jié)果。

在實(shí)現(xiàn)此計(jì)算器的時(shí)候，首先需要考慮的是 I/O 服務(wù)。該計(jì)算器使用 TCP/IP 協(xié)議，需要在指定端口監(jiān)聽，接受客戶端的連接。Apache MINA 提供了基于 Java NIO 的套接字實(shí)現(xiàn)，可以直接使用。其次要考慮的是 I/O 過濾器。I/O 過濾器過濾所有的 I/O 事件和請求，可以用來處理橫切的需求，如記錄日志、壓縮等。最后就是 I/O 處理器。I/O 處理器用來處理業(yè)務(wù)邏輯。具體到該應(yīng)用來說，就是在接收到消息之后，把該消息作為一個(gè)表達(dá)式來執(zhí)行，并把結(jié)果發(fā)送回去。I/O 處理器需要實(shí)現(xiàn) org.apache.mina.core.service.IoHandler接口或者繼承自 org.apache.mina.core.service.IoHandlerAdapter。該應(yīng)用的 I/O 處理器的實(shí)現(xiàn)如 清單 1 所示。

  public class CalculatorHandler extends IoHandlerAdapter {
            private static final Logger LOGGER = LoggerFactory
            .getLogger(CalculatorHandler.class);
            private ScriptEngine jsEngine = null;
            public CalculatorHandler() {
            ScriptEngineManager sfm = new ScriptEngineManager();
            jsEngine = sfm.getEngineByName("JavaScript");
            if (jsEngine == null) {
            throw new RuntimeException("找不到 JavaScript 引擎。");
            }
            }
            public void exceptionCaught(IoSession session, Throwable cause)
            throws Exception {
            LOGGER.warn(cause.getMessage(), cause);
            }
            public void messageReceived(IoSession session, Object message)
            throws Exception {
            String expression = message.toString();
            if ("quit".equalsIgnoreCase(expression.trim())) {
            session.close(true);
            return;
            }
            try {
            Object result = jsEngine.eval(expression);
            session.write(result.toString());
            } catch (ScriptException e) {
            LOGGER.warn(e.getMessage(), e);
            session.write("Wrong expression, try again.");
            }
            }
            }
            

在 清單 1 中，messageReceived 由 IoHandler 接口聲明。當(dāng)接收到新的消息的時(shí)候，該方法就會被調(diào)用。此處的邏輯是如果傳入了“quit”，則通過 session.close關(guān)閉當(dāng)前連接；如果不是的話，就執(zhí)行該表達(dá)式并把結(jié)果通過 session.write發(fā)送回去。此處執(zhí)行表達(dá)式用的是 JDK 6 中提供的 JavaScript 腳本引擎。此處使用到了 I/O 會話相關(guān)的方法，會在下面進(jìn)行說明。

接下來只需要把 I/O 處理器和 I/O 過濾器配置到 I/O 服務(wù)上就可以了。具體的實(shí)現(xiàn)如 清單 2 所示。

public class CalculatorServer {
            private static final int PORT = 10010;
            private static final Logger LOGGER = LoggerFactory
            .getLogger(CalculatorServer.class);
            public static void main(String[] args) throws IOException {
            IoAcceptor acceptor = new NioSocketAcceptor();
            acceptor.getFilterChain().addLast("logger", new LoggingFilter());
            acceptor.getFilterChain().addLast(
            "codec",
            new ProtocolCodecFilter(new TextLineCodecFactory(Charset
            .forName("UTF-8"))));
            acceptor.setHandler(new CalculatorHandler());
            acceptor.bind(new InetSocketAddress(PORT));
            LOGGER.info("計(jì)算器服務(wù)已啟動，端口是" + PORT);
            }
            }
            

清單 2 中，首先創(chuàng)建一個(gè) org.apache.mina.transport.socket.nio.NioSocketAcceptor 的實(shí)例，由它提供 I/O 服務(wù)；接著獲得該 I/O 服務(wù)的過濾器鏈，并添加兩個(gè)新的過濾器，一個(gè)用來記錄相關(guān)日志，另外一個(gè)用來在字節(jié)流和文本之間進(jìn)行轉(zhuǎn)換；最后配置 I/O 處理器。完成這些之后，通過 bind 方法來在特定的端口進(jìn)行監(jiān)聽，接收連接。服務(wù)器啟動之后，可以通過操作系統(tǒng)自帶的 Telnet 工具來進(jìn)行測試，如 圖 2 所示。在輸入表達(dá)式之后，計(jì)算結(jié)果會出現(xiàn)在下面一行。

在介紹了簡單的計(jì)算器服務(wù)這個(gè)應(yīng)用之后，下面說明本文中會使用的復(fù)雜的聯(lián)機(jī)游戲應(yīng)用。

回頁首

聯(lián)機(jī)游戲示例說明

上一節(jié)中給出了一個(gè)簡單的基于 Apache MINA 的網(wǎng)絡(luò)應(yīng)用的實(shí)現(xiàn)，可以用來熟悉基本的架構(gòu)。而在實(shí)際開發(fā)中，網(wǎng)絡(luò)應(yīng)用都是有一定復(fù)雜度的。下面會以一個(gè)比較復(fù)雜的聯(lián)機(jī)游戲作為示例來詳細(xì)介紹 Apache MINA 的概念、API 和典型用法。

該聯(lián)機(jī)游戲支持兩個(gè)人進(jìn)行俄羅斯方塊的對戰(zhàn)。這個(gè)游戲借鑒了 QQ 的“火拼俄羅斯”。用戶在啟動客戶端之后，需要輸入一個(gè)昵稱進(jìn)行注冊。用戶可以在“游戲大廳”中查看當(dāng)前已注冊的所有其它用戶。當(dāng)前用戶可以選擇另外的一 個(gè)用戶發(fā)送游戲邀請。邀請被接受之后就可以開始進(jìn)行對戰(zhàn)。在游戲過程中，當(dāng)前用戶可以看到對方的游戲狀態(tài)，即方塊的情況。該游戲的運(yùn)行效果如 圖 3 所示。

下面開始以這個(gè)應(yīng)用為例來具體介紹 Apache MINA 中的基本概念。先從 I/O 服務(wù)開始。

回頁首

I/O 服務(wù)

I/O 服務(wù)用來執(zhí)行真正的 I/O 操作，以及管理 I/O 會話。根據(jù)所使用的數(shù)據(jù)傳輸方式的不同，有不同的 I/O 服務(wù)的實(shí)現(xiàn)。由于 I/O 服務(wù)執(zhí)行的是輸入和輸出兩種操作，實(shí)際上有兩種具體的子類型。一種稱為“I/O 接受器（I/O acceptor）”，用來接受連接，一般用在服務(wù)器的實(shí)現(xiàn)中；另外一種稱為“I/O 連接器（I/O connector）”，用來發(fā)起連接，一般用在客戶端的實(shí)現(xiàn)中。對應(yīng)在 Apache MINA 中的實(shí)現(xiàn)，org.apache.mina.core.service.IoService是 I/O 服務(wù)的接口，而繼承自它的接口 org.apache.mina.core.service.IoAcceptor 和 org.apache.mina.core.service.IoConnector 則分別表示 I/O 接受器和 I/O 連接器。IoService 接口提供的重要方法如 表 1 所示。

下面具體介紹 I/O 接受器和 I/O 連接器。

I/O 接受器

I/O 接受器用來接受連接，與對等體（客戶端）進(jìn)行通訊，并發(fā)出相應(yīng)的 I/O 事件交給 I/O 處理器來處理。使用 I/O 接受器的時(shí)候，只需要調(diào)用 bind方法并指定要監(jiān)聽的套接字地址。當(dāng)不再接受連接的時(shí)候，調(diào)用 unbind停止監(jiān)聽即可。關(guān)于 I/O 接受器的具體用法，可以參考 清單 2 中給出的計(jì)算器服務(wù)的實(shí)現(xiàn)。

I/O 連接器

I/O 連接器用來發(fā)起連接，與對等體（服務(wù)器）進(jìn)行通訊，并發(fā)出相應(yīng)的 I/O 事件交給 I/O 處理器來處理。使用 I/O 連接器的時(shí)候，只需要調(diào)用 connect方法連接指定的套接字地址。另外可以通過 setConnectTimeoutMillis設(shè)置連接超時(shí)時(shí)間（毫秒數(shù)）。

清單 3 中給出了使用 I/O 連接器的一個(gè)示例。

SocketConnector connector = new NioSocketConnector();
            connector.setConnectTimeoutMillis(CONNECT_TIMEOUT);
            connector.getFilterChain().addLast("logger", new LoggingFilter());
            connector.getFilterChain().addLast("protocol",
            new ProtocolCodecFilter(new TetrisCodecFactory()));
            ConnectFuture connectFuture = connector.connect(new InetSocketAddress(host, port));
            connectFuture.awaitUninterruptibly();
            

在 清單 3 中，首先創(chuàng)建一個(gè) Java NIO 的套接字連接器 NioSocketConnector 的實(shí)例，接著設(shè)置超時(shí)時(shí)間。再添加了 I/O 過濾器之后，通過 connect 方法連接到指定的地址和端口即可。

在介紹完 I/O 服務(wù)之后，下面介紹 I/O 會話。

回頁首

I/O 會話

I/O 會話表示一個(gè)活動的網(wǎng)絡(luò)連接，與所使用的傳輸方式無關(guān)。I/O 會話可以用來存儲用戶自定義的與應(yīng)用相關(guān)的屬性。這些屬性通常用來保存應(yīng)用的狀態(tài)信息，還可以用來在 I/O 過濾器和 I/O 處理器之間交換數(shù)據(jù)。I/O 會話在作用上類似于 Servlet 規(guī)范中的 HTTP 會話。

Apache MINA 中 I/O 會話實(shí)現(xiàn)的接口是 org.apache.mina.core.session.IoSession。該接口中比較重要的方法如 表 2 所示。

在介紹完 I/O 會話之后，下面介紹 I/O 過濾器。

回頁首

I/O 過濾器

從 I/O 服務(wù)發(fā)送過來的所有 I/O 事件和請求，在到達(dá) I/O 處理器之前，會先由 I/O 過濾器鏈中的 I/O 過濾器進(jìn)行處理。Apache MINA 中的過濾器與 Servlet 規(guī)范中的過濾器是類似的。過濾器可以在很多情況下使用，比如記錄日志、性能分析、訪問控制、負(fù)載均衡和消息轉(zhuǎn)換等。過濾器非常適合滿足網(wǎng)絡(luò)應(yīng)用中各種橫切 的非功能性需求。在一個(gè)基于 Apache MINA 的網(wǎng)絡(luò)應(yīng)用中，一般存在多個(gè)過濾器。這些過濾器互相串聯(lián)，形成鏈條，稱為過濾器鏈。每個(gè)過濾器依次對傳入的 I/O 事件進(jìn)行處理。當(dāng)前過濾器完成處理之后，由過濾器鏈中的下一個(gè)過濾器繼續(xù)處理。當(dāng)前過濾器也可以不調(diào)用下一個(gè)過濾器，而提前結(jié)束，這樣 I/O 事件就不會繼續(xù)往后傳遞。比如負(fù)責(zé)用戶認(rèn)證的過濾器，如果遇到未認(rèn)證的對等體發(fā)出的 I/O 事件，則會直接關(guān)閉連接。這可以保證這些事件不會通過此過濾器到達(dá) I/O 處理器。

Apache MINA 中 I/O 過濾器都實(shí)現(xiàn) org.apache.mina.core.filterchain.IoFilter接口。一般來說，不需要完整實(shí)現(xiàn) IOFilter接口，只需要繼承 Apache MINA 提供的適配器 org.apache.mina.core.filterchain.IoFilterAdapter，并覆寫所需的事件過濾方法即可，其它方法的默認(rèn)實(shí)現(xiàn)是不做任何處理，而直接把事件轉(zhuǎn)發(fā)到下一個(gè)過濾器。

IoFilter 接口詳細(xì)說明

IoFilter接口提供了 15 個(gè)方法。這 15 個(gè)方法大致分成兩類，一類是與過濾器的生命周期相關(guān)的，另外一類是用來過濾 I/O 事件的。第一類方法如 表 3 所示。

在 表 3 中給出的方法中，參數(shù) parent 表示包含此過濾器的過濾器鏈，參數(shù) name 表示過濾器的名稱，參數(shù) nextFilter 表示過濾器鏈中的下一個(gè)過濾器。

第二類方法如 表 4 所示。

對于 表 4 中給出的與 I/O 事件相關(guān)的方法，它們都有一個(gè)參數(shù)是 nextFilter，表示過濾器鏈中的下一個(gè)過濾器。如果當(dāng)前過濾器完成處理之后，可以通過調(diào)用 nextFilter 中的方法，把 I/O 事件傳遞到下一個(gè)過濾器。如果當(dāng)前過濾器不調(diào)用 nextFilter 中的方法的話，該 I/O 事件就不能繼續(xù)往后傳遞。另外一個(gè)共同的參數(shù)是 session，用來表示當(dāng)前的 I/O 會話，可以用來發(fā)送消息給對等體。下面通過具體的實(shí)例來說明過濾器的實(shí)現(xiàn)。

BlacklistFilter

BlacklistFilter是 Apache MINA 自帶的一個(gè)過濾器實(shí)現(xiàn)，其功能是阻止來自特定地址的連接，即所謂的“黑名單”功能。BlacklistFilter繼承自 IoFilterAdapter，并覆寫了 IoHandler相關(guān)的方法。清單 4 中給出了部分實(shí)現(xiàn)。

public void messageReceived(NextFilter nextFilter, IoSession session, Object message) {
            if (!isBlocked(session)) {
            nextFilter.messageReceived(session, message);
            } else {
            blockSession(session);
            }
            }
            private void blockSession(IoSession session) {
            session.close(true);
            }

在 清單 4 中 messageReceived 方法的實(shí)現(xiàn)中，首先通過 isBlocked 來判斷當(dāng)前連接是否應(yīng)該被阻止，如果不是的話，則通過 nextFilter.messageReceived 把該 I/O 事件傳遞到下一個(gè)過濾器；否則的話，則通過 blockSession 來阻止當(dāng)前連接。

使用 ProtocolCodecFilter

ProtocolCodecFilter 用來在字節(jié)流和消息對象之間互相轉(zhuǎn)換。當(dāng)該過濾器接收到字節(jié)流的時(shí)候，需要首先判斷消息的邊界，然后把表示一條消息的字節(jié)提取出來，通過一定的邏輯轉(zhuǎn)換成 消息對象，再把消息對象往后傳遞，交給 I/O 處理器來執(zhí)行業(yè)務(wù)邏輯。這個(gè)過程稱為“解碼”。與“解碼”對應(yīng)的是“編碼”過程。在“編碼”的時(shí)候，過濾器接收到的是消息對象，通過與“解碼”相反的邏 輯，把消息對象轉(zhuǎn)換成字節(jié)，并反向傳遞，交給 I/O 服務(wù)來執(zhí)行 I/O 操作。

在“編碼”和“解碼”中的一個(gè)重要問題是如何在字節(jié)流中判斷消息的邊界。通常來說，有三種辦法解決這個(gè)問題：

• 使用固定長度的消息。這種方式實(shí)現(xiàn)起來比較簡單，只需要每次讀取特定數(shù)量的字節(jié)即可。
• 使用固定長度的消息頭來指明消息主體的長度。比如每個(gè)消息開始的 4 個(gè)字節(jié)的值表示了后面緊跟的消息主體的長度。只需要首先讀取該長度，再讀取指定數(shù)量的字節(jié)即可。
• 使用分隔符。消息之間通過特定模式的分隔符來分隔。每次只要遇到該模式的字節(jié)，就表示到了一個(gè)消息的末尾。

具體到示例應(yīng)用來說，客戶端和服務(wù)器之間的通信協(xié)議比較復(fù)雜，有不同種類的消息。每種消息的格式都不相同，同類消息的內(nèi)容也不盡相同。因此，使用固定長度的消息頭來指明消息主體的長度就成了最好的選擇。

示例應(yīng)用中的每種消息主體由兩部分組成，第一部分是固定長度的消息類別名稱，第二部分是每種消息的主體內(nèi)容。圖 4 中給出了示例應(yīng)用中一條完整的消息的結(jié)構(gòu)。

AbstractTetrisCommand用來描述聯(lián)機(jī)游戲示例應(yīng)用中的消息。它是一個(gè)抽象類，是所有具體消息的基類。其具體實(shí)現(xiàn)如 清單 5 所示。

public abstract class AbstractTetrisCommand implements TetrisCommand {
            public abstract String getName();
            public abstract byte[] bodyToBytes() throws Exception;
            public abstract void bodyFromBytes(byte[] bytes) throws Exception;
            public byte[] toBytes() throws Exception {
            byte[] body = bodyToBytes();
            int commandNameLength = Constants.COMMAND_NAME_LENGTH;
            int len = commandNameLength + body.length;
            byte[] bytes = new byte[len];
            String name = StringUtils.rightPad(getName(), commandNameLength,
            Constants.COMMAND_NAME_PAD_CHAR);
            name = name.substring(0, commandNameLength);
            System.arraycopy(name.getBytes(), 0, bytes, 0, commandNameLength);
            System.arraycopy(body, 0, bytes, commandNameLength, body.length);
            return bytes;
            }
            }

如 清單 5 所示，AbstractTetrisCommand 中定義了 3 個(gè)抽象方法：getName、bodyToBytes 和 bodyFromBytes，分別用來獲取消息的名稱、把消息的主體轉(zhuǎn)換成字節(jié)數(shù)組和從字節(jié)數(shù)組中構(gòu)建消息。bodyToBytes對應(yīng)于前面提到的“編碼”過程，而 bodyFromBytes對應(yīng)于“解碼”過程。每種具體的消息都應(yīng)該實(shí)現(xiàn)這 3 個(gè)方法。AbstractTetrisCommand 中的方法 toBytes 封裝了把消息的主體轉(zhuǎn)換成字節(jié)數(shù)組的邏輯，在字節(jié)數(shù)組中，首先是長度固定為 Constants.COMMAND_NAME_LENGTH的消息類別名稱，緊接著是每種消息特定的主體內(nèi)容，由 bodyToBytes 方法來生成。

在介紹完示例應(yīng)用中的消息格式之后，下面將討論具體的“編碼”和“解碼”過程。“編碼”過程由編碼器來完成，編碼器需要實(shí)現(xiàn) org.apache.mina.filter.codec.ProtocolEncoder 接口，一般來說繼承自 org.apache.mina.filter.codec.ProtocolEncoderAdapter 并覆寫所需的方法即可。清單 6 中給出了示例應(yīng)用中消息編碼器 CommandEncoder 的實(shí)現(xiàn)。

public class CommandEncoder extends ProtocolEncoderAdapter {
            public void encode(IoSession session, Object message,
            ProtocolEncoderOutput out) throws Exception {
            AbstractTetrisCommand command = (AbstractTetrisCommand) message;
            byte[] bytes = command.toBytes();
            IoBuffer buf = IoBuffer.allocate(bytes.length, false);
            buf.setAutoExpand(true);
            buf.putInt(bytes.length);
            buf.put(bytes);
            buf.flip();
            out.write(buf);
            }
            }

在 清單 6 中，encode 方法封裝了編碼的邏輯。由于 AbstractTetrisCommand的 toBytes已經(jīng)完成了到字節(jié)數(shù)組的轉(zhuǎn)換，encode 方法直接使用即可。首先寫入消息主體字節(jié)數(shù)組的長度，再是字節(jié)數(shù)組本身，就完成了編碼的過程。

與編碼過程相比，解碼過程要相對復(fù)雜一些。具體的實(shí)現(xiàn)如 清單 7 所示。

public class CommandDecoder extends CumulativeProtocolDecoder {
            protected boolean doDecode(IoSession session, IoBuffer in,
            ProtocolDecoderOutput out) throws Exception {
            if (in.prefixedDataAvailable(4, Constants.MAX_COMMAND_LENGTH)) {
            int length = in.getInt();
            byte[] bytes = new byte[length];
            in.get(bytes);
            int commandNameLength = Constants.COMMAND_NAME_LENGTH;
            byte[] cmdNameBytes = new byte[commandNameLength];
            System.arraycopy(bytes, 0, cmdNameBytes, 0, commandNameLength);
            String cmdName = StringUtils.trim(new String(cmdNameBytes));
            AbstractTetrisCommand command = TetrisCommandFactory
            .newCommand(cmdName);
            if (command != null) {
            byte[] cmdBodyBytes = new byte[length - commandNameLength];
            System.arraycopy(bytes, commandNameLength, cmdBodyBytes, 0,
            length - commandNameLength);
            command.bodyFromBytes(cmdBodyBytes);
            out.write(command);
            }
            return true;
            } else {
            return false;
            }
            }
            }

在 清單 7 中可以看到，解碼器 CommandDecoder 繼承自 CumulativeProtocolDecoder。這是 Apache MINA 提供的一個(gè)幫助類，它會自動緩存所有已經(jīng)接收到的數(shù)據(jù)，直到編碼器認(rèn)為可以開始進(jìn)行編碼。這樣在實(shí)現(xiàn)自己的編碼器的時(shí)候，就只需要考慮如何判斷消息的邊界即可。如果一條消息的后續(xù)數(shù)據(jù)還沒有接收到，CumulativeProtocolDecoder會自動進(jìn)行緩存。在之前提到過，解碼過程的一個(gè)重要問題是判斷消息的邊界。對于固定長度的消息來說，只需要使用 Apache MINA 的 IoBuffer的 remaining方法來判斷當(dāng)前緩存中的字節(jié)數(shù)目，如果大于消息長度的話，就進(jìn)行解碼；對于使用固定長度消息頭來指明消息主體的長度的情況，IoBuffer提供了 prefixedDataAvailable方法來滿足這一需求。prefixedDataAvailable會檢查當(dāng)前緩存中是否有固定長度的消息頭，并且由此消息頭指定長度的消息主體是否已經(jīng)全部在緩存中。如果這兩個(gè)條件都滿足的話，說明一條完整的消息已經(jīng)接收到，可以進(jìn)行解碼了。解碼的過程本身并不復(fù)雜，首先讀取消息的類別名稱，然后通過 TetrisCommandFactory.newCommand方法來生成一個(gè)該類消息的實(shí)例，接著通過該實(shí)例的 bodyFromBytes方法就可以從字節(jié)數(shù)組中恢復(fù)消息的內(nèi)容，得到一個(gè)完整的消息對象。每次成功解碼一個(gè)消息對象，需要調(diào)用 ProtocolDecoderOutput的 write把此消息對象往后傳遞。消息對象會通過過濾器鏈，最終達(dá)到 I/O 處理器，在 IoHandler.messageReceived中接收到此消息對象。如果當(dāng)前緩存的數(shù)據(jù)不足以用來解碼一條消息的話，doDecode只需要返回 false即可。接收到新的數(shù)據(jù)之后，doDecode會被再次調(diào)用。

過濾器鏈

過濾器只有在添加到過濾器鏈中的時(shí)候才起作用。過濾器鏈?zhǔn)沁^濾器的容器。過濾器鏈與 I/O 會話是一一對應(yīng)的關(guān)系。org.apache.mina.core.filterchain.IoFilterChain是 Apache MINA 中過濾器鏈的接口，其中提供了一系列方法對其中包含的過濾器進(jìn)行操作，包括查詢、添加、刪除和替換等。如 表 5 所示。

在介紹完 I/O 過濾器和過濾器鏈之后，下面介紹 I/O 處理器。

回頁首

I/O 處理器

I/O 事件通過過濾器鏈之后會到達(dá) I/O 處理器。I/O 處理器中與 I/O 事件對應(yīng)的方法會被調(diào)用。Apache MINA 中 org.apache.mina.core.service.IoHandler是 I/O 處理器要實(shí)現(xiàn)的接口，一般情況下，只需要繼承自 org.apache.mina.core.service.IoHandlerAdapter并覆寫所需方法即可。IoHandler接口的方法如 表 6 所示。

對于 表 6 中的方法，有幾個(gè)需要重點(diǎn)的說明一下。首先是 sessionCreated 和 sessionOpened 的區(qū)別。sessionCreated方法是由 I/O 處理線程來調(diào)用的，而 sessionOpened 是由其它線程來調(diào)用的。因此從性能方面考慮，不要在 sessionCreated 方法中執(zhí)行過多的操作。對于 sessionIdle，默認(rèn)情況下，閑置時(shí)間設(shè)置是禁用的，也就是說 sessionIdle 并不會被調(diào)用?？梢酝ㄟ^ IoSessionConfig.setIdleTime(IdleStatus, int) 來進(jìn)行設(shè)置。

Apache MINA 中的基本概念已經(jīng)介紹完了，下面介紹狀態(tài)機(jī)的使用。

回頁首

使用狀態(tài)機(jī)

在 I/O 處理器中實(shí)現(xiàn)業(yè)務(wù)邏輯的時(shí)候，對于簡單的情況，一般只需要在 messageReceived 方法中對傳入的消息進(jìn)行處理。如果需要寫回?cái)?shù)據(jù)到對等體的話，用 IoSession.write 方法即可。在另外的一些情況下，客戶端和服務(wù)器端的通信協(xié)議比較復(fù)雜，客戶端其實(shí)是有狀態(tài)變遷的。這個(gè)時(shí)候可以用 Apache MINA 提供的狀態(tài)機(jī)實(shí)現(xiàn)，可以使得 I/O 處理器的實(shí)現(xiàn)更加簡單。

狀態(tài)機(jī)中兩個(gè)重要的元素是狀態(tài)以及狀態(tài)之間的遷移。示例應(yīng)用中客戶端的狀態(tài)以及遷移如 圖 5 所示。

客戶端初始化的時(shí)候，其狀態(tài)為“未連接”，表示客戶端還沒有在服務(wù)器上面注冊，此時(shí)還不能進(jìn)行游戲；接著用戶需要輸入一個(gè)昵稱來注冊到服務(wù)器 上面，完成之后狀態(tài)遷移到“閑置”。此時(shí)客戶端會接收到當(dāng)前在線的所有其它用戶的列表。當(dāng)前用戶可以邀請其它用戶和他一塊游戲，也可以接收來自其它用戶的 邀請。邀請發(fā)送出去之后，客戶端的狀態(tài)遷移到“邀請已發(fā)送”。如果接受了其它用戶的邀請，客戶端的狀態(tài)遷移到“邀請已接收”。如果某個(gè)用戶的邀請被另外一 個(gè)用戶接受的話，兩個(gè)客戶端的狀態(tài)都會遷移到“游戲中”。

要實(shí)現(xiàn)這樣較為復(fù)雜的狀態(tài)機(jī)的話，只需要在 I/O 處理器中以聲明式的方式定義狀態(tài)和遷移條件就可以了。首先需要聲明狀態(tài)機(jī)中狀態(tài)，如 清單 8 所示。

@State public static final String ROOT = "Root";
            @State(ROOT) public static final String NOT_CONNECTED = "NotConnected";
            @State(ROOT) public static final String IDLE = "Idle";
            @State(ROOT) public static final String INVITATION_SENT = "InvitationSent";
            @State(ROOT) public static final String INVITATION_ACCEPTED = "InvitationAccepted";
            @State(ROOT) public static final String PLAYING = "Playing";
            

如 清單 8 所示，上面定義了一共六個(gè)狀態(tài)。通過標(biāo)注 @State就 聲明了一個(gè)狀態(tài)。需要注意的是狀態(tài)之間是可以繼承的。如果狀態(tài)機(jī)接收到一個(gè)事件的時(shí)候，在當(dāng)前狀態(tài)中找不到對應(yīng)的遷移，就會在其父狀態(tài)上繼續(xù)查找。狀態(tài)的 繼承在某些情況下是很有用的，比如希望為所有的狀態(tài)都增加同樣的遷移邏輯，就可以直接把遷移條件添加在父狀態(tài)上面。一個(gè)典型的場景就是錯(cuò)誤處理，一般來 說，所有的狀態(tài)都需要錯(cuò)誤處理，而錯(cuò)誤處理的邏輯一般都是相同的。把發(fā)生錯(cuò)誤時(shí)候的遷移放在父狀態(tài)中，可以簡潔的描述這一場景。

定義了狀態(tài)之后，下面應(yīng)該聲明狀態(tài)之間的遷移。如 清單 9 所示。

@IoHandlerTransition(on = MESSAGE_RECEIVED, in = NOT_CONNECTED, next = IDLE)
            public void login(TetrisServerContext context, IoSession session, LoginCommand cmd) {
            String nickName = cmd.getNickName();
            context.nickName = nickName;
            session.setAttribute("nickname", nickName);
            session.setAttribute("status", UserStatus.IDLE);
            sessions.add(session);
            users.add(nickName);
            RefreshPlayersListCommand command = createRefreshPlayersListCommand();
            broadcast(command);
            }
            @IoHandlerTransition(on = EXCEPTION_CAUGHT, in = ROOT, weight = 10)
            public void exceptionCaught(IoSession session, Exception e) {
            LOGGER.warn("Unexpected error.", e);
            session.close(true);
            }
            @IoHandlerTransition(in = ROOT, weight = 100)
            public void unhandledEvent() {
            LOGGER.warn("Unhandled event.");
            }
            

清單 9 中，使用標(biāo)注 @IoHandlerTransition聲明了一個(gè)狀態(tài)遷移。每個(gè)狀態(tài)遷移可以有四個(gè)屬性：on、in、next和 weight，其中屬性 in是必須的，其余是可選的。屬性 on表示觸發(fā)此狀態(tài)遷移的事件名稱，如果省略該屬性的話，則默認(rèn)為匹配所有事件的通配符。該屬性的值可以是表中給出的 I/O 處理器中能處理的七種事件類型。屬性 in表示狀態(tài)遷移的起始狀態(tài)。屬性 next表示狀態(tài)遷移的結(jié)束狀態(tài)，如果省略該屬性的話，則默認(rèn)為表示當(dāng)前狀態(tài) 的 _self_。屬性 weight用來指明狀態(tài)遷移的權(quán)重。一個(gè)狀態(tài)的所有遷移是按照其權(quán)重升序排列的。對于當(dāng)前狀態(tài)，如果有多個(gè)可能的遷移，排序靠前的遷移將會發(fā)生。代碼中的第一個(gè)標(biāo)注聲明了如果當(dāng)前狀態(tài)是“未連接”，并且接收到了 MESSAGE_RECEIVED事件，而且消息的內(nèi)容是一個(gè) LoginCommand對象的話，login方法會被調(diào)用，調(diào)用完成之后，當(dāng)前狀態(tài)遷移到“閑置”。第二個(gè)標(biāo)注聲明了對于任何的狀態(tài)，如果接收到了 EXCEPTION_CAUGHT事件，exceptionCaught方法會被調(diào)用。最后一個(gè)標(biāo)注聲明了一個(gè)狀態(tài)遷移，其起始狀態(tài)是 ROOT，表示該遷移對所有的事件都起作用。不過它的 weight是 100，優(yōu)先級比較低。該狀態(tài)遷移的作用是處理其它沒有對應(yīng)狀態(tài)遷移的事件。

使用了 Apache MINA 提供的狀態(tài)機(jī)之后，創(chuàng)建 I/O 處理器的方式也發(fā)生了變化。I/O 處理器的實(shí)例由狀態(tài)機(jī)來創(chuàng)建，如 清單 10 所示。

private static IoHandler createIoHandler() {
            StateMachine sm = StateMachineFactory.getInstance(
            IoHandlerTransition.class).create(ServerHandler.NOT_CONNECTED,
            new ServerHandler());
            return new StateMachineProxyBuilder().setStateContextLookup(
            new IoSessionStateContextLookup(new StateContextFactory() {
            public StateContext create() {
            return new ServerHandler.TetrisServerContext();
            }
            })).create(IoHandler.class, sm);
            }
            

在 清單 10 中，TetrisServerContext是提供給狀態(tài)機(jī)的上下文對象，用來在狀態(tài)之間共享數(shù)據(jù)。當(dāng)然用 IoSession也是可以實(shí)現(xiàn)的，不過上下文對象的好處是類型安全，不需要做額外的類型轉(zhuǎn)換。

在介紹完?duì)顟B(tài)機(jī)后，下面介紹一些高級話題，包括異步操作以及 Apache MINA 與 JMX 和 Spring 的集成。

回頁首

高級話題

在前面章節(jié)中介紹了 Apache MINA 的基本概念和具體用法，下面討論一些高級話題。

異步操作

Apache MINA 中的很多操作都是異步的，比如連接的建立、連接的關(guān)閉、還有數(shù)據(jù)的發(fā)送等。在編寫網(wǎng)絡(luò)應(yīng)用的時(shí)候，需要考慮這一點(diǎn)。比如 IoConnector的 connect方法，其返回值是 org.apache.mina.core.future.ConnectFuture類的對象。通過此對象，可以查詢連接操作的狀態(tài)。清單 3 中已經(jīng)使用了 ConnectFuture。另外一個(gè)常用的是發(fā)送數(shù)據(jù)時(shí)使用的 org.apache.mina.core.future.WriteFuture，如 清單 11 所示。

IoSession session = ...; // 獲取 I/O 會話對象
            WriteFuture future = session.write("Hello World"); // 發(fā)送數(shù)據(jù)
            future.awaitUninterruptibly(); // 等待發(fā)送數(shù)據(jù)操作完成
            if(future.isWritten())
            {
            // 數(shù)據(jù)已經(jīng)被成功發(fā)送
            }
            else
            {
            // 數(shù)據(jù)發(fā)送失敗
            }
            

由于這樣的需求很常見，I/O 處理器中提供了 messageSent方法，當(dāng)數(shù)據(jù)發(fā)送成功的時(shí)候，該方法會被調(diào)用。

JMX 集成

Apache MINA 可以集成 JMX 來對網(wǎng)絡(luò)應(yīng)用進(jìn)行管理和監(jiān)測。下面通過對前面給出的計(jì)算器服務(wù)進(jìn)行簡單修改，來說明如何集成 JMX。所需的修改如 清單 12 所示。

MBeanServer mBeanServer = ManagementFactory.getPlatformMBeanServer();
            IoAcceptor acceptor = new NioSocketAcceptor();
            IoServiceMBean acceptorMBean = new IoServiceMBean(acceptor);
            ObjectName acceptorName = new ObjectName(acceptor.getClass()
            .getPackage().getName()
            + ":type=acceptor,name=" + acceptor.getClass().getSimpleName());
            mBeanServer.registerMBean(acceptorMBean, acceptorName);
            

如 清單 12 所示，首先獲取平臺提供的受控 bean 的服務(wù)器，接著創(chuàng)建受控 bean（MBean）來包裝想要管理和監(jiān)測的對象，這里使用的是 I/O 連接器對象。最后把創(chuàng)建出來的受控 bean 注冊到服務(wù)器即可。

在啟動計(jì)算器服務(wù)應(yīng)用的時(shí)候，添加下面的啟動參數(shù)：-Dcom.sun.management.jmxremote -Dcom.sun.management.jmxremote.port=8084 -Dcom.sun.management.jmxremote.authenticate=false -Dcom.sun.management.jmxremote.ssl=false，就啟用了 JMX。接著通過 JVM 提供的“Java 監(jiān)視和管理控制臺”（運(yùn)行 jconsole）就可以連接到此應(yīng)用進(jìn)行管理和監(jiān)測了。監(jiān)測的結(jié)果如 圖 6 所示。

Spring 集成

Apache MINA 可以和流行的開源框架 Spring 進(jìn)行集成，由 Spring 來管理 Apache MINA 中的對象。與 Spring 集成的方式也比較簡單，只需要編寫相應(yīng)的 Spring 配置文件即可。清單 13 中給出了與 Spring 集成之后的計(jì)算器服務(wù)的配置文件。

 <?xml version="1.0" encoding="UTF-8"?>
            <beans>
            <bean id="calculatorHandler" class="calculator.CalculatorHandler" />
            <bean id="loggingFilter" class="org.apache.mina.filter.logging.LoggingFilter" />
            <bean id="calculatorCodecFilter" class="org.apache.mina.filter.
            codec.ProtocolCodecFilter">
            <constructor-arg>
            <bean class="org.apache.mina.filter.codec.textline.TextLineCodecFactory" />
            </constructor-arg>
            </bean>
            <bean id="filterChainBuilder"
            class="org.apache.mina.core.filterchain.DefaultIoFilterChainBuilder">
            <property name="filters">
            <map>
            <entry key="loggingFilter" value-ref="loggingFilter" />
            <entry key="codecFilter" value-ref="calculatorCodecFilter" />
            </map>
            </property>
            </bean>
            <bean class="org.springframework.beans.factory.config.CustomEditorConfigurer">
            <property name="customEditors">
            <map>
            <entry key="java.net.SocketAddress">
            <bean class="org.apache.mina.integration.beans.InetSocketAddressEditor"/>
            </entry>
            </map>
            </property>
            </bean>
            <bean id="ioAcceptor" class="org.apache.mina.transport.socket.nio.NioSocketAcceptor"
            init-method="bind" destroy-method="unbind">
            <property name="defaultLocalAddress" value=":10010" />
            <property name="handler" ref="calculatorHandler" />
            <property name="filterChainBuilder" ref="filterChainBuilder" />
            </bean>
            </beans>
            

清單 13 中創(chuàng)建 I/O 處理器和 I/O 過濾器的方式很直接。由于不能直接從 I/O 接受器獲取過濾器鏈，這里創(chuàng)建了一個(gè) org.apache.mina.core.filterchain.DefaultIoFilterChainBuilder類的 bean，用來構(gòu)建過濾器鏈。由 Apache MINA 提供的網(wǎng)絡(luò)地址編輯器 org.apache.mina.integration.beans.InetSocketAddressEditor允許以“主機(jī)名 : 端口”的形式指定網(wǎng)絡(luò)地址。在聲明 I/O 接受器的時(shí)候，通過 init-method指明了當(dāng) I/O 接受器創(chuàng)建成功之后，調(diào)用其 bind方法來接受連接；通過 destroy-method聲明了當(dāng)其被銷毀的時(shí)候，調(diào)用其 unbind來停止監(jiān)聽。

回頁首

總結(jié)

Apache MINA 是一個(gè)很好的網(wǎng)絡(luò)應(yīng)用框架，它通過事件驅(qū)動的 API 為開發(fā)網(wǎng)絡(luò)應(yīng)用提供了良好的架構(gòu)基礎(chǔ)，同時(shí)也提供了豐富的 I/O 服務(wù)和 I/O 過濾器的實(shí)現(xiàn)，使得開發(fā)網(wǎng)絡(luò)應(yīng)用變得簡單。本文詳細(xì)介紹了 Apache MINA 中的基本概念，包括 I/O 服務(wù)、I/O 會話、I/O 過濾器和 I/O 處理器等，同時(shí)介紹了如何利用狀態(tài)機(jī)來實(shí)現(xiàn)邏輯復(fù)雜的 I/O 處理器。除此之外，還討論了 Apache MINA 如何與 JMX 和 Spring 進(jìn)行集成。本文提供了一個(gè)簡單的計(jì)算器服務(wù)和復(fù)雜的俄羅斯方塊聯(lián)機(jī)游戲作為示例，可以幫助讀者更好的掌握基于 Apache MINA 的網(wǎng)絡(luò)應(yīng)用開發(fā)。