通過上面的兩個示例,你應該對Mina 如何編寫TCP/IP 協議棧的網絡通信有了一些感性的認識。
(1.)IoService:
這個接口是服務端IoAcceptor、客戶端IoConnector 的抽象,提供IO 服務和管理IoSession的功能,它有如下幾個常用的方法:
用的方法:
A. TransportMetadata getTransportMetadata():
這個方法獲取傳輸方式的元數據描述信息,也就是底層到底基於什么的實現,譬如:nio、apr 等
B. void addListener(IoServiceListener listener):
這個方法可以為IoService 增加一個監聽器,用於監聽IoService 的創建、活動、失效、空閑、銷毀,具體可以參考IoServiceListener 接口中的方法,這為你參與IoService 的生命周期提供了機會。
C. void removeListener(IoServiceListener listener):
這個方法用於移除上面的方法添加的監聽器。
D. void setHandler(IoHandler handler):
這個方法用於向IoService 注冊IoHandler,同時有getHandler()方法獲取Handler。
E. Map<Long,IoSession> getManagedSessions():
這個方法獲取IoService 上管理的所有IoSession,Map 的key 是IoSession 的id。
F. IoSessionConfig getSessionConfig():
這個方法用於獲取IoSession 的配置對象,通過IoSessionConfig 對象可以設置Socket 連接的一些選項。
(2.)IoAcceptor:
這個接口是TCPServer 的接口,主要增加了void bind()監聽端口、void unbind()解除對套接字的監聽等方法。這里與傳統的JAVA 中的ServerSocket 不同的是IoAcceptor 可以多次調用bind()方法(或者在一個方法中傳入多個SocketAddress 參數)同時監聽多個端口。
3.)IoConnector:
這個接口是TCPClient 的接口, 主要增加了ConnectFuture connect(SocketAddressremoteAddress,SocketAddress localAddress)方法,用於與Server 端建立連接,第二個參數如果不傳遞則使用本地的一個隨機端口訪問Server 端。這個方法是異步執行的,同樣的,也可以同時連接多個服務端。
(4.)IoSession:
這個接口用於表示Server 端與Client 端的連接,IoAcceptor.accept()的時候返回實例。
這個接口有如下常用的方法:
A. WriteFuture write(Object message):
這個方法用於寫數據,該操作是異步的。
B. CloseFuture close(boolean immediately):
這個方法用於關閉IoSession,該操作也是異步的,參數指定true 表示立即關閉,否則就在所有的寫操作都flush 之后再關閉。
C. Object setAttribute(Object key,Object value):
這個方法用於給我們向會話中添加一些屬性,這樣可以在會話過程中都可以使用,類似於HttpSession 的setAttrbute()方法。IoSession 內部使用同步的HashMap 存儲你添加的自定義屬性。
D. SocketAddress getRemoteAddress():
這個方法獲取遠端連接的套接字地址。
E. void suspendWrite():
這個方法用於掛起寫操作,那么有void resumeWrite()方法與之配對。對於read()方法同樣適用。
F. ReadFuture read():
這個方法用於讀取數據, 但默認是不能使用的, 你需要調用IoSessionConfig 的setUseReadOperation(true)才可以使用這個異步讀取的方法。一般我們不會用到這個方法,因為這個方法的內部實現是將數據保存到一個BlockingQueue,假如是Server 端,因為大量的Client 端發送的數據在Server 端都這么讀取,那么可能會導致內存泄漏,但對於Client,可能有的時候會比較便利。
G. IoService getService():
這個方法返回與當前會話對象關聯的IoService 實例。
(5.)IoSessionConfig:
這個方法用於指定此次會話的配置,它有如下常用的方法:
A. void setReadBufferSize(int size):
這個方法設置讀取緩沖的字節數,但一般不需要調用這個方法,因為IoProcessor 會自動調整緩沖的大小。你可以調用setMinReadBufferSize()、setMaxReadBufferSize()方法,這樣無論IoProcessor 無論如何自動調整,都會在你指定的區間。
B. void setIdleTime(IdleStatus status,int idleTime):
這個方法設置關聯在通道上的讀、寫或者是讀寫事件在指定時間內未發生,該通道就進入空閑狀態。一旦調用這個方法,則每隔idleTime 都會回調過濾器、IoHandler 中的sessionIdle()方法。
C. void setWriteTimeout(int time):
這個方法設置寫操作的超時時間。
D. void setUseReadOperation(boolean useReadOperation):
這個方法設置IoSession 的read()方法是否可用,默認是false。
(6.)IoHandler:
這個接口是你編寫業務邏輯的地方,從上面的示例代碼可以看出,讀取數據、發送數據基本都在這個接口總完成,這個實例是綁定到IoService 上的,有且只有一個實例(沒有給一個IoService 注入一個IoHandler 實例會拋出異常)。它有如下幾個方法:
A. void sessionCreated(IoSession session):
這個方法當一個Session 對象被創建的時候被調用。對於TCP 連接來說,連接被接受的時候調用,但要注意此時TCP 連接並未建立,此方法僅代表字面含義,也就是連接的對象IoSession 被創建完畢的時候,回調這個方法。對於UDP 來說,當有數據包收到的時候回調這個方法,因為UDP 是無連接的。
B. void sessionOpened(IoSession session):
這個方法在連接被打開時調用,它總是在sessionCreated()方法之后被調用。對於TCP 來說,它是在連接被建立之后調用,你可以在這里執行一些認證操作、發送數據等。對於UDP 來說,這個方法與sessionCreated()沒什么區別,但是緊跟其后執行。如果你每隔一段時間,發送一些數據,那么sessionCreated()方法只會在第一次調用,但是sessionOpened()方法每次都會調用。
C. void sessionClosed(IoSession session) :
對於TCP 來說,連接被關閉時,調用這個方法。對於UDP 來說,IoSession 的close()方法被調用時才會毀掉這個方法。
(7.)IoBuffer:
這個接口是對JAVA NIO 的ByteBuffer 的封裝,這主要是因為ByteBuffer 只提供了對基本數據類型的讀寫操作,沒有提供對字符串等對象類型的讀寫方法,使用起來更為方便,另外,ByteBuffer 是定長的,如果想要可變,將很麻煩。IoBuffer 的可變長度的實現類似於StringBuffer。IoBuffer 與ByteBuffer 一樣,都是非線程安全的。本節的一些內容如果不清楚,可以參考java.nio.ByteBuffer 接口。這個接口有如下常用的方法:
D. void sessionIdle(IoSession session, IdleStatus status) :
這個方法在IoSession 的通道進入空閑狀態時調用,對於UDP 協議來說,這個方法始終不會被調用。
E. void exceptionCaught(IoSession session, Throwable cause) :
這個方法在你的程序、Mina 自身出現異常時回調,一般這里是關閉IoSession。
F. void messageReceived(IoSession session, Object message) :
接收到消息時調用的方法,也就是用於接收消息的方法,一般情況下,message 是一個IoBuffer 類,如果你使用了協議編解碼器,那么可以強制轉換為你需要的類型。通常我們都是會使用協議編解碼器的, 就像上面的例子, 因為協議編解碼器是
TextLineCodecFactory,所以我們可以強制轉message 為String 類型。
G. void messageSent(IoSession session, Object message) :
當發送消息成功時調用這個方法,注意這里的措辭,發送成功之后,也就是說發送消息是不能用這個方法的。
發送消息的時機:
發送消息應該在sessionOpened()、messageReceived()方法中調用IoSession.write()方法完成。因為在sessionOpened()方法中,TCP 連接已經真正打開,同樣的在messageReceived()方法TCP 連接也是打開狀態,只不過兩者的時機不同。sessionOpened()方法是在TCP 連接建立之后,接收到數據之前發送;messageReceived()方法是在接收到數據之后發送,你可以完成依據收到的內容是什么樣子,決定發送什么樣的數據。因為這個接口中的方法太多,因此通常使用適配器模式IoHandlerAdapter,覆蓋你所感興趣的方法即可。
A. static IoBuffer allocate(int capacity,boolean useDirectBuffer):
這個方法內部通過SimpleBufferAllocator 創建一個實例,第一個參數指定初始化容量,第二個參數指定使用直接緩沖區還是JAVA 內存堆的緩存區,默認為false。
B. void free():
釋放緩沖區,以便被一些IoBufferAllocator 的實現重用,一般沒有必要調用這個方法,除非你想提升性能(但可能未必效果明顯)。
C. IoBuffer setAutoExpand(boolean autoExpand):
這個方法設置IoBuffer 為自動擴展容量,也就是前面所說的長度可變,那么可以看出長度可變這個特性默認是不開啟的。
D. IoBuffer setAutoShrink(boolean autoShrink):
這個方法設置IoBuffer 為自動收縮,這樣在compact()方法調用之后,可以裁減掉一些沒有使用的空間。如果這個方法沒有被調用或者設置為false,你也可以通過調用shrink()方法手動收縮空間。
E. IoBuffer order(ByteOrder bo):
這個方法設置是Big Endian 還是Little Endian,JAVA 中默認是Big Endian,C++和其他語言一般是Little Endian。
F. IoBuffer asReadOnlyBuffer():
這個方法設置IoBuffer 為只讀的。
G. Boolean prefixedDataAvailable(int prefixLength,int maxDataLength):
這個方法用於數據的最開始的1、2、4 個字節表示的是數據的長度的情況,
prefixLentgh表示這段數據的前幾個字節(只能是1、2、4 的其中一個),代表的是這段數據的長度,
maxDataLength 表示最多要讀取的字節數。返回結果依賴於等式
remaining()-prefixLength>=maxDataLength,也就是總的數據-表示長度的字節,剩下的字節數要比打算讀取的字節數大或者相等。
H. String getPrefixedString(int prefixLength,CharsetDecoder decoder):
如果上面的方法返回true,那么這個方法將開始讀取表示長度的字節之后的數據,注意要保持這兩個方法的prefixLength 的值是一樣的。
G、H 兩個方法在后面講到的PrefixedStringDecoder 中的內部實現使用。
IoBuffer 剩余的方法與ByteBuffer 都是差不多的,額外增加了一些便利的操作方法,例如:
IoBuffer putString(String value,CharsetEncoder encoder)可以方便的以指定的編碼方式存儲字符串、InputStream asInputStream()方法從IoBuffer 剩余的未讀的數據中轉為輸入流等。
(8.)IoFuture:
在Mina 的很多操作中,你會看到返回值是XXXFuture,實際上他們都是IoFuture 的子類,看到這樣的返回值,這個方法就說明是異步執行的,主要的子類有ConnectFuture、CloseFuture 、ReadFuture 、WriteFuture 。這個接口的大部分操作都和
java.util.concurrent.Future 接口是類似的,譬如:await()、awaitUninterruptibly()等,一般我們常用awaitUninterruptibly()方法可以等待異步執行的結果返回。這個接口有如下常用的方法:
A. IoFuture addListener(IoFutureListener<?> listener):
這個方法用於添加一個監聽器, 在異步執行的結果返回時監聽器中的回調方法operationComplete(IoFuture future),也就是說,這是替代awaitUninterruptibly()方法另一種等待異步執行結果的方法,它的好處是不會產生阻塞。
B. IoFuture removeListener(IoFutureListener<?> listener):
這個方法用於移除指定的監聽器。
C. IoSession getSession():
這個方法返回當前的IoSession。舉個例子,我們在客戶端調用connect()方法訪問Server 端的時候,實際上這就是一個異步執行的方法,也就是調用connect()方法之后立即返回,執行下面的代碼,而不管是否連接成功。
那么如果我想在連接成功之后執行一些事情(譬如:獲取連接成功后的IoSession對象),該怎么辦呢?按照上面的說明,你有如下兩種辦法:
第一種:

第二種:

為了更好的看清楚使用監聽器是異步的,而不是像awaitUninterruptibly()那樣會阻塞主線程的執行,我們在回調方法中暫停5 秒鍾,然后輸出+++,在最后輸出***。我們執行代碼之后,你會發現首先輸出***(這證明了監聽器是異步執行的),然后IoSession 對象Created,系統暫停5 秒,然后輸出+++,最后IoSession 對象Opened,也就是TCP 連接建立。
4.日志配置:
前面的示例代碼中提到了使用SLF4J 作為日志門面,這是因為Mina 內部使用的就是SLF4J,你也使用SLF4J 可以與之保持一致性。Mina 如果想啟用日志跟蹤Mina 的運行細節,你可以配置LoggingFilter 過濾器,這樣你可
以看到Session 建立、打開、空閑等一系列細節在日志中輸出,默認SJF4J 是按照DEBUG級別輸出跟蹤信息的,如果你想給某一類別的Mina 運行信息輸出指定日志輸出級別,可以調用LoggingFilter 的setXXXLogLevel(LogLevel.XXX)。

這里IoSession 被打開的跟蹤信息將以ERROR 級別輸出到日志。
5.過濾器:
前面我們看到了LoggingFilter、ProtocolCodecFilter 兩個過濾器,一個負責日志輸出,一個負責數據的編解碼,通過最前面的Mina 執行流程圖,在IoProcessor 與IoHandler 之間可以有很多的過濾器,這種設計方式為你提供可插拔似的擴展功能提供了非常便利的方式,目前的Apache CXF、Apache Struts2 中的攔截器也都是一樣的設計思路。Mina 中的IoFilter 是單例的,這與CXF、Apache Struts2 沒什么區別。IoService 實例上會綁定一個DefaultIoFilterChainBuilder 實例,DefaultIoFilterChainBuilder 會把使用內部的EntryImpl 類把所有的過濾器按照順序連在一起,組成一個過濾器鏈。
DefaultIoFilterChainBuilder 類如下常用的方法:
A. void addFirst(String name,IoFilter filter):
這個方法把過濾器添加到過濾器鏈的頭部,頭部就是IoProcessor 之后的第一個過濾器。同樣的addLast()方法把過濾器添加到過濾器鏈的尾部。
B. void addBefore(String baseName,String name,IoFilter filter):
這個方法將過濾器添加到baseName 指定的過濾器的前面,同樣的addAfter()方法把過濾器添加到baseName 指定的過濾器的后面。這里要注意無論是那種添加方法,每個過濾器的名字(參數name)必須是唯一的。
C. IoFilter remove(Stirng name):
這個方法移除指定名稱的過濾器,你也可以調用另一個重載的remove()方法,指定要移除的IoFilter 的類型。
這個方法返回當前IoService 上注冊的所有過濾器。默認情況下,過濾器鏈中是空的,也就是getAll()方法返回長度為0 的List,但實際Mina內部有兩個隱藏的過濾器:HeadFilter、TailFilter,分別在List 的最開始和最末端,很明顯,TailFilter 在最末端是為了調用過濾器鏈之后,調用IoHandler。但這兩個過濾器對你來說是透明的,可以忽略它們的存在。編寫一個過濾器很簡單,你需要實現IoFilter 接口,如果你只關注某幾個方法,可以繼承IoFilterAdapter 適配器類。IoFilter 接口中主要包含兩類方法,一類是與IoHandler 中的方法名一致的方法,相當於攔截IoHandler 中的方法,另一類是IoFilter 的生命周期回調方法,這些回調方法的執行順序和解釋如下所示:
(1.)init()在首次添加到鏈中的時候被調用,但你必須將這個IoFilter 用
ReferenceCountingFilter 包裝起來,否則init()方法永遠不會被調用。
(2.)onPreAdd()在調用添加到鏈中的方法時被調用,但此時還未真正的加入到鏈。
(3.)onPostAdd()在調用添加到鏈中的方法后被調,如果在這個方法中有異常拋出,則過濾器會立即被移除,同時destroy()方法也會被調用(前提是使用ReferenceCountingFilter包裝)。
(4.)onPreRemove()在從鏈中移除之前調用。
(5.)onPostRemove()在從鏈中移除之后調用。
(6.)destory()在從鏈中移除時被調用,使用方法與init()要求相同。
無論是哪個方法,要注意必須在實現時調用參數nextFilter 的同名方法,否則,過濾器鏈的執行將被中斷,IoHandler 中的同名方法一樣也不會被執行,這就相當於Servlet 中的Filter 必須調用filterChain.doFilter(request,response)才能繼續前進是一樣的道理。







這里我們將MyIoFilter 用ReferenceCountingFilter 包裝起來,這樣你可以看到init()、destroy()方法調用。我們啟動客戶端訪問,然后關閉客戶端,你會看到執行順序如下所示:
init onPreAdd onPostAdd sessionCreated sessionOpened messageReceived filterClose sessionClosed onPreRemove onPostRemove destroy。
IoHandler 的對應方法會跟在上面的對應方法之后執行,這也就是說從橫向(單獨的看一個過濾器中的所有方法的執行順序)上看,每個過濾器的執行順序是上面所示的順序;從縱向(方法鏈的調用)上看,如果有filter1、filter2 兩個過濾器,sessionCreated()方法的執行順序如下所示:
filter1-sessionCreated filter2-sessionCreated IoHandler-sessionCreated。
這里你要注意init、onPreAdd、onPostAdd 三個方法並不是在Server 啟動時調用的,而是IoSession 對象創建之前調用的,也就是說IoFilterChain.addXXX()方法僅僅負責初始化過濾器並注冊過濾器,但並不調用任何方法,包括init()初始化方法也是在IoProcessor 開始工作的時候被調用。IoFilter 是單例的,那么init()方法是否只被執行一次呢?這個是不一定的,因為IoFilter是被IoProcessor 調用的,而每個IoService 通常是關聯多個IoProcessor,所以IoFilter的init()方法是在每個IoProcessor 線程上只執行一次。關於Mina 的線程問題,我們后面會詳細討論,這里你只需要清楚,init()與destroy()的調用次數與IoProceesor 的個數有關,假如一個IoService 關聯了3 個IoProcessor,有五個並發的客戶端請求,那么你會看到三次init()方法被調用,以后將不再會調用。Mina中自帶的過濾器:
過濾器 說明
BlacklistFilter 設置一些IP 地址為黑名單,不允許訪問。
BufferedWriteFilter 設置輸出時像BufferedOutputStream 一樣進行緩沖。
CompressionFilter 設置在輸入、輸出流時啟用JZlib 壓縮。
ConnectionThrottleFilter 這個過濾器指定同一個IP 地址(不含端口號)上的請求在多長的毫秒值內可以有一個請求,如果小於指定的時間間隔就有連續兩個請求,那么第二個請求將被忽略(IoSession.close())。正如Throttle 的名字一樣,調節訪問的頻率這個過濾器最好放在過濾器鏈的前面。
FileRegionWriteFilter 如果你想使用File 對象進行輸出,請使用這個過濾器。要注意,你需要使用WriteFuture 或者在
messageSent() 方法中關閉File 所關聯的FileChannel 通道。
StreamWriteFilter 如果你想使用InputStream 對象進行輸出,請使用這個過濾器。要注意,你需要使用WriteFuture或者在messageSent()方法中關閉File 所關聯的
FileChannel 通道。NoopFilter 這個過濾器什么也不做,如果你想測試過濾器鏈是否起作用,可以用它來測試。
ProfilerTimerFilter 這個過濾器用於檢測每個事件方法執行的時間,所以最好放在過濾器鏈的前面。
ProxyFilter 這個過濾器在客戶端使用ProxyConnector 作為實現時,會自動加入到過濾器鏈中,用於完成代理功能。
RequestResponseFilter 暫不知曉。
SessionAttributeInitializingFilter 這個過濾器在IoSession 中放入一些屬性(Map),通常放在過濾器的前面,用於放置一些初始化的信息。
MdcInjectionFilter 針對日志輸出做MDC 操作,可以參考LOG4J 的MDC、NDC 的文檔。
WriteRequestFilter CompressionFilter、RequestResponseFilter 的基類,用於包裝寫請求的過濾器。
還有一些過濾器,會在各節中詳細討論,這里沒有列出,譬如:前面的LoggingFilger 日志過濾器。
