一、mina總體框架與案例：

1.總體結(jié)構(gòu)圖：

簡(jiǎn)述：以上是一張來(lái)自網(wǎng)上比較經(jīng)典的圖，總體上揭示了mina的結(jié)構(gòu)，當(dāng)中IoService包括clientIoConnector和服務(wù)端IoAcceptor兩部分。即不管是client還是服務(wù)端都是這個(gè)結(jié)構(gòu)。IoService封裝了網(wǎng)絡(luò)傳輸層（TCP和UDP），而IoFilterChain中mina自帶的filter做了一些主要的操作之外，支持?jǐn)U展。經(jīng)過(guò)FilterChain之后終于調(diào)用IoHandler，IoHandler是詳細(xì)實(shí)現(xiàn)業(yè)務(wù)邏輯的處理接口，詳細(xì)的業(yè)務(wù)實(shí)現(xiàn)可擴(kuò)展。

2.一個(gè)可執(zhí)行的案例（案例來(lái)自網(wǎng)上，轉(zhuǎn)載后試驗(yàn)）：
Client.java:

import java.net.InetSocketAddress; import java.nio.charset.Charset; import java.util.Random;import org.apache.mina.core.future.ConnectFuture; import org.apache.mina.core.future.IoFutureListener; import org.apache.mina.core.service.IoConnector; import org.apache.mina.core.service.IoHandlerAdapter; import org.apache.mina.core.session.IoSession; import org.apache.mina.filter.codec.ProtocolCodecFilter; import org.apache.mina.filter.codec.textline.TextLineCodecFactory; import org.apache.mina.transport.socket.nio.NioSocketConnector;public class Client extends IoHandlerAdapter {private Random random = new Random(System.currentTimeMillis());public Client() {IoConnector connector = new NioSocketConnector();connector.getFilterChain().addLast("text",new ProtocolCodecFilter(new TextLineCodecFactory(Charset.forName(Server.ENCODE))));connector.setHandler(this);ConnectFuture future = connector.connect(new InetSocketAddress("127.0.0.1", Server.PORT));future.awaitUninterruptibly();future.addListener(new IoFutureListener<ConnectFuture>() {@Overridepublic void operationComplete(ConnectFuture future) {IoSession session = future.getSession();while (!session.isClosing()) {try {Thread.sleep(100);} catch (InterruptedException e) {e.printStackTrace();}String message = "你好。我roll了" + random.nextInt(100) + "點(diǎn).";session.write(message);}}});connector.dispose();}@Overridepublic void messageReceived(IoSession session, Object message)throws Exception {System.out.println("批復(fù):" + message.toString());}@Overridepublic void messageSent(IoSession session, Object message) throws Exception {System.out.println("報(bào)告:" + message.toString());}@Overridepublic void exceptionCaught(IoSession session, Throwable cause)throws Exception {cause.printStackTrace();session.close(true);}public static void main(String[] args) {for (int i = 0; i < 10; i++) {new Client();}} }

ServerHandler.java:

import java.net.InetSocketAddress; import java.util.regex.Matcher; import java.util.regex.Pattern; import org.apache.mina.core.service.IoHandlerAdapter; import org.apache.mina.core.session.IdleStatus; import org.apache.mina.core.session.IoSession;public class ServerHandler extends IoHandlerAdapter {@Overridepublic void exceptionCaught(IoSession session, Throwable cause)throws Exception {cause.printStackTrace();session.close(false);}public void messageReceived(IoSession session, Object message)throws Exception {String s = message.toString();System.out.println("收到請(qǐng)求：" + s);if (s != null) {int i = getPoint(s);if (session.isConnected()) {if (i >= 95) {session.write("運(yùn)氣不錯(cuò)，你能夠出去了.");session.close(false);return;}Integer count = (Integer) session.getAttribute(Server.KEY);count++;session.setAttribute(Server.KEY, count);session.write("抱歉。你運(yùn)氣太差了，第" + count + "次請(qǐng)求未被通過(guò)。繼續(xù)在小黑屋呆著吧.");} else {session.close(true);}}}@Overridepublic void messageSent(IoSession session, Object message) throws Exception {System.out.println("發(fā)給client：" + message.toString());}@Overridepublic void sessionClosed(IoSession session) throws Exception {long l = session.getCreationTime();System.out.println("來(lái)自" + getInfo(session) + "的會(huì)話已經(jīng)關(guān)閉，它已經(jīng)存活了"+ (System.currentTimeMillis() - 1) + "毫秒");}@Overridepublic void sessionCreated(IoSession session) throws Exception {System.out.println("給" + getInfo(session) + "創(chuàng)建了一個(gè)會(huì)話");}@Overridepublic void sessionIdle(IoSession session, IdleStatus status)throws Exception {System.out.println("來(lái)自" + getInfo(session) + "的會(huì)話閑置，狀態(tài)為"+ status.toString());}public void sessionOpened(IoSession session) throws Exception {session.setAttribute(Server.KEY, 0);System.out.println("和" + getInfo(session) + "的會(huì)話已經(jīng)打開(kāi).");}public String getInfo(IoSession session) {if (session == null) {return null;}InetSocketAddress address = (InetSocketAddress) session.getRemoteAddress();int port = address.getPort();String ip = address.getAddress().getHostAddress();return ip + ":" + port;}public int getPoint(String s) {if (s == null) {return -1;}Pattern p = Pattern.compile("^[\u0041-\uFFFF,]*(\\d+).*$");Matcher m = p.matcher(s);if (m.matches()) {return Integer.valueOf(m.group(1));}return 0;} }

Server.java:

import java.io.IOException; import java.net.InetSocketAddress; import java.nio.charset.Charset;import org.apache.mina.filter.codec.ProtocolCodecFilter; import org.apache.mina.filter.codec.textline.TextLineCodecFactory; import org.apache.mina.transport.socket.SocketAcceptor; import org.apache.mina.transport.socket.nio.NioSocketAcceptor;public class Server {public static final int PORT = 2534;public static String ENCODE = "UTF-8";public static final String KEY = "roll";public static void main(String[] args){ SocketAcceptor acceptor = new NioSocketAcceptor();acceptor.getFilterChain().addLast("text",new ProtocolCodecFilter(new TextLineCodecFactory(Charset.forName(ENCODE))));acceptor.setHandler(new ServerHandler());try {acceptor.bind(new InetSocketAddress(PORT));System.out.println("游戲開(kāi)始，你想出去嗎，來(lái)，碰碰運(yùn)氣吧！");} catch (IOException e) {e.printStackTrace();acceptor.dispose();}} }

本案例依賴的jar例如以下圖：

簡(jiǎn)述：以上是依賴mina實(shí)現(xiàn)的一個(gè)可執(zhí)行的案例，就不多說(shuō)了，結(jié)合總體的結(jié)構(gòu)圖和案例實(shí)現(xiàn)能夠看出mina框架還是非常輕量級(jí)的。以下分析一下mina的源代碼結(jié)構(gòu)和一些時(shí)序流程。

二、mina 核心源代碼分析：

1.mina的啟動(dòng)時(shí)序(結(jié)合上面的案例)：

簡(jiǎn)述：SocketAcceptor作為服務(wù)端對(duì)外啟動(dòng)接口類，在bind網(wǎng)絡(luò)地址的時(shí)候，會(huì)觸發(fā)服務(wù)端一系列服務(wù)的啟動(dòng)，從調(diào)用鏈能夠清晰找到相應(yīng)的源代碼閱讀。

當(dāng)中AbstractPollingIoAcceptor是一個(gè)核心類，它會(huì)調(diào)用自身的startupAcceptor方法，來(lái)啟動(dòng)一個(gè)存放Acceptor的線程池用來(lái)處理client傳輸過(guò)來(lái)的請(qǐng)求。
AbstractPollingIoAcceptor 類的 startupAcceptor 方法例如以下：

/*** This method is called by the doBind() and doUnbind()* methods. If the acceptor is null, the acceptor object will* be created and kicked off by the executor. If the acceptor* object is null, probably already created and this class* is now working, then nothing will happen and the method* will just return.*/ private void startupAcceptor() throws InterruptedException {// If the acceptor is not ready, clear the queues// TODO : they should already be clean : do we have to do that ?

if (!selectable) { registerQueue.clear(); cancelQueue.clear(); } // start the acceptor if not already started Acceptor acceptor = acceptorRef.get(); //這里僅僅會(huì)啟動(dòng)一個(gè)worker if (acceptor == null) { lock.acquire(); acceptor = new Acceptor(); if (acceptorRef.compareAndSet(null, acceptor)) { executeWorker(acceptor); } else { lock.release(); } } }

上面調(diào)用到 AbstractIoService 的 executeWorker方法例如以下：

protected final void executeWorker(Runnable worker) {executeWorker(worker, null); }protected final void executeWorker(Runnable worker, String suffix) {String actualThreadName = threadName;if (suffix != null) {actualThreadName = actualThreadName + '-' + suffix;}executor.execute(new NamePreservingRunnable(worker, actualThreadName)); }

簡(jiǎn)述：有類AbstractPollingIoAcceptor 的 startupAcceptor方法（上文）能夠看到，一個(gè)SocketAcceptor僅僅啟動(dòng)了一個(gè)Worker線程（即代碼中的Acceptor對(duì)象）而且把他加到線程池中。反過(guò)來(lái)講，也能夠看出AbstractIoService維護(hù)了Worker的線程池。（ps：這個(gè)Worker就是服務(wù)端處理請(qǐng)求的線程）。

2.Mina處理client鏈接的過(guò)程（啟動(dòng)后）：

概述：從1中的啟動(dòng)時(shí)序能夠看到，啟動(dòng)過(guò)程通過(guò)創(chuàng)建SocketAcceptor將有類AbstractPollingIoAcceptor的內(nèi)部類Acceptor放到了 AbstractIoService的線程池里面，而這個(gè)Acceptor就是處理client網(wǎng)絡(luò)請(qǐng)求的worker。而以下這個(gè)時(shí)序就是線程池中每一個(gè)worker處理client網(wǎng)絡(luò)請(qǐng)求的時(shí)序流程。

處理請(qǐng)求時(shí)序：?

簡(jiǎn)述：worker線程Acceptor的run方法中會(huì)調(diào)用NioSocketAcceptor或者AprSocketAccetpor的select方法。
ps:APR（Apache Protable Runtime Library，Apache可移植執(zhí)行庫(kù)）是能夠提供非常好的可拓展性、性能以及對(duì)底層操作系統(tǒng)一致性操作的技術(shù)，說(shuō)白了就是apache實(shí)現(xiàn)的一套標(biāo)準(zhǔn)的通訊接口。

AprSocketAcceptor先不做深入了解，主要了解下NioSocketAcceptor，NioSocketAcceptor顧名思義，它調(diào)用了java NIO的API實(shí)現(xiàn)了NIO的網(wǎng)絡(luò)連接處理過(guò)程。

AbstractPolling$Acceptor 的run方法的核心代碼例如以下：

private class Acceptor implements Runnable {public void run() {assert (acceptorRef.get() == this);int nHandles = 0;// Release the locklock.release();while (selectable) {try {// Detect if we have some keys ready to be processed// The select() will be woke up if some new connection// have occurred, or if the selector has been explicitly// woke up//調(diào)用了NioSocketAcceptor的select方法，獲取了selectKeyint selected = select();// this actually sets the selector to OP_ACCEPT,// and binds to the port on which this class will// listen onnHandles += registerHandles();// Now, if the number of registred handles is 0, we can// quit the loop: we don't have any socket listening// for incoming connection.if (nHandles == 0) {acceptorRef.set(null);if (registerQueue.isEmpty() && cancelQueue.isEmpty()) {assert (acceptorRef.get() != this);break;}if (!acceptorRef.compareAndSet(null, this)) {assert (acceptorRef.get() != this);break;}assert (acceptorRef.get() == this);}if (selected > 0) {// We have some connection request, let's process// them here.processHandles(selectedHandles());}// check to see if any cancellation request has been made.nHandles -= unregisterHandles();} catch (ClosedSelectorException cse) {// If the selector has been closed, we can exit the loopbreak;} catch (Throwable e) {ExceptionMonitor.getInstance().exceptionCaught(e);try {Thread.sleep(1000);} catch (InterruptedException e1) {ExceptionMonitor.getInstance().exceptionCaught(e1);}}}// Cleanup all the processors, and shutdown the acceptor.if (selectable && isDisposing()) {selectable = false;try {if (createdProcessor) {processor.dispose();}} finally {try {synchronized (disposalLock) {if (isDisposing()) {destroy();}}} catch (Exception e) {ExceptionMonitor.getInstance().exceptionCaught(e);} finally {disposalFuture.setDone();}}}}

簡(jiǎn)述：從上面的代碼中能夠看出一個(gè)典型的網(wǎng)絡(luò)請(qǐng)求處理的程序，在循環(huán)中拿到處理的請(qǐng)求后就調(diào)用AbstractPollingIoAcceptor的processHandles（）對(duì)網(wǎng)絡(luò)請(qǐng)求做處理。

代碼例如以下：

/*** This method will process new sessions for the Worker class. All* keys that have had their status updates as per the Selector.selectedKeys()* method will be processed here. Only keys that are ready to accept* connections are handled here.* <p/>* Session objects are created by making new instances of SocketSessionImpl* and passing the session object to the SocketIoProcessor class.*/@SuppressWarnings("unchecked")private void processHandles(Iterator<H> handles) throws Exception {while (handles.hasNext()) {H handle = handles.next();handles.remove();// Associates a new created connection to a processor,// and get back a session//這里調(diào)用了NioSocketAcceptor的accept方法S session = accept(processor, handle);if (session == null) {continue;}initSession(session, null, null);// add the session to the SocketIoProcessor// 這步處理add操作，會(huì)觸發(fā)對(duì)client請(qǐng)求的異步處理。

session.getProcessor().add(session); } }

NioSocketAcceptor的accept方法new了一個(gè)包裝Process處理線程的session實(shí)例:而且在調(diào)用session.getProcessor().add(session)的操作的時(shí)候觸發(fā)了對(duì)client請(qǐng)求的異步處理。

/*** {@inheritDoc}*/ @Override protected NioSession accept(IoProcessor<NioSession> processor, ServerSocketChannel handle) throws Exception {SelectionKey key = handle.keyFor(selector);if ((key == null) || (!key.isValid()) || (!key.isAcceptable())) {return null;}// accept the connection from the clientSocketChannel ch = handle.accept();if (ch == null) {return null;}return new NioSocketSession(this, processor, ch); }

再看上面時(shí)序圖：有一步是AbstractPollingIoProcessor調(diào)用了startupProcessor方法。代碼例如以下：

/*** Starts the inner Processor, asking the executor to pick a thread in its* pool. The Runnable will be renamed*/ private void startupProcessor() {Processor processor = processorRef.get();if (processor == null) {processor = new Processor();if (processorRef.compareAndSet(null, processor)) {executor.execute(new NamePreservingRunnable(processor, threadName));}}// Just stop the select() and start it again, so that the processor// can be activated immediately.wakeup(); }

簡(jiǎn)述：這個(gè)startupProcessor方法在調(diào)用 session里包裝的processor的add方法是，觸發(fā)了將處理client請(qǐng)求的processor放入異步處理的線程池中。興許詳細(xì)Processor怎么處理client請(qǐng)求的流程，涉及到FilterChain的過(guò)濾。以及Adapter的調(diào)用。用來(lái)處理業(yè)務(wù)邏輯。詳細(xì)的異步處理時(shí)序看以下的時(shí)序圖：

簡(jiǎn)述：這個(gè)時(shí)序就是將待處理的client鏈接，通過(guò)NIO的形式接受請(qǐng)求，并將請(qǐng)求包裝成Processor的形式放到處理的線程池中異步的處理。

在異步的處理過(guò)程中則調(diào)用了Processor的run方法，詳細(xì)的filterchain的調(diào)用和業(yè)務(wù)Adapter的調(diào)用也是在這一步得到處理。

值得注意的是。Handler的調(diào)用是封裝在DefaultFilterchain的內(nèi)部類誒TairFilter中觸發(fā)調(diào)用的。Processor的run方法代碼例如以下：

private class Processor implements Runnable {public void run() {assert (processorRef.get() == this);int nSessions = 0;lastIdleCheckTime = System.currentTimeMillis();for (;;) {try {// This select has a timeout so that we can manage// idle session when we get out of the select every// second. (note : this is a hack to avoid creating// a dedicated thread).long t0 = System.currentTimeMillis();//調(diào)用了NioProcessorint selected = select(SELECT_TIMEOUT);long t1 = System.currentTimeMillis();long delta = (t1 - t0);if ((selected == 0) && !wakeupCalled.get() && (delta < 100)) {// Last chance : the select() may have been// interrupted because we have had an closed channel.if (isBrokenConnection()) {LOG.warn("Broken connection");// we can reselect immediately// set back the flag to falsewakeupCalled.getAndSet(false);continue;} else {LOG.warn("Create a new selector. Selected is 0, delta = " + (t1 - t0));// Ok, we are hit by the nasty(討厭的) epoll// spinning.// Basically, there is a race condition// which causes a closing file descriptor not to be// considered as available as a selected channel, but// it stopped the select. The next time we will// call select(), it will exit immediately for the same// reason, and do so forever, consuming 100%// CPU.// We have to destroy the selector, and// register all the socket on a new one.registerNewSelector();}// Set back the flag to falsewakeupCalled.getAndSet(false);// and continue the loopcontinue;}// Manage newly created session firstnSessions += handleNewSessions();updateTrafficMask();// Now, if we have had some incoming or outgoing events,// deal with themif (selected > 0) {//LOG.debug("Processing ..."); // This log hurts one of the MDCFilter test...//觸發(fā)了詳細(xì)的調(diào)用邏輯process();}// Write the pending requestslong currentTime = System.currentTimeMillis();flush(currentTime);// And manage removed sessionsnSessions -= removeSessions();// Last, not least, send Idle events to the idle sessionsnotifyIdleSessions(currentTime);// Get a chance to exit the infinite loop if there are no// more sessions on this Processorif (nSessions == 0) {processorRef.set(null);if (newSessions.isEmpty() && isSelectorEmpty()) {// newSessions.add() precedes startupProcessorassert (processorRef.get() != this);break;}assert (processorRef.get() != this);if (!processorRef.compareAndSet(null, this)) {// startupProcessor won race, so must exit processorassert (processorRef.get() != this);break;}assert (processorRef.get() == this);}// Disconnect all sessions immediately if disposal has been// requested so that we exit this loop eventually.if (isDisposing()) {for (Iterator<S> i = allSessions(); i.hasNext();) {scheduleRemove(i.next());}wakeup();}} catch (ClosedSelectorException cse) {// If the selector has been closed, we can exit the loopbreak;} catch (Throwable t) {ExceptionMonitor.getInstance().exceptionCaught(t);try {Thread.sleep(1000);} catch (InterruptedException e1) {ExceptionMonitor.getInstance().exceptionCaught(e1);}}}try {synchronized (disposalLock) {if (disposing) {doDispose();}}} catch (Throwable t) {ExceptionMonitor.getInstance().exceptionCaught(t);} finally {disposalFuture.setValue(true);}} }

簡(jiǎn)述：這么一坨代碼能夠看出，這個(gè)處理器也調(diào)用了java的Nio API是一個(gè)NIO模型。當(dāng)中select和process方法各自是從session拿到要處理的請(qǐng)求，并進(jìn)行處理。而詳細(xì)的Processor實(shí)例是NioProcessor。從加入凝視的代碼中有一步調(diào)用了自身的process方法，這步調(diào)用觸發(fā)了詳細(xì)業(yè)務(wù)邏輯的調(diào)用。能夠結(jié)合代碼和時(shí)序圖看下。在Process方法中會(huì)調(diào)用reader（session）或wirte（session）方法，然后調(diào)用fireMessageReceived方法，這種方法又調(diào)用了callNextMessageReceived方法致使觸發(fā)了整個(gè)FilterChain和Adapter的調(diào)用。read方法的核心代碼例如以下：

private void read(S session) {IoSessionConfig config = session.getConfig();int bufferSize = config.getReadBufferSize();IoBuffer buf = IoBuffer.allocate(bufferSize);final boolean hasFragmentation = session.getTransportMetadata().hasFragmentation();try {int readBytes = 0;int ret;try {if (hasFragmentation) {while ((ret = read(session, buf)) > 0) {readBytes += ret;if (!buf.hasRemaining()) {break;}}} else {ret = read(session, buf);if (ret > 0) {readBytes = ret;}}} finally {buf.flip();}if (readBytes > 0) {IoFilterChain filterChain = session.getFilterChain();filterChain.fireMessageReceived(buf);buf = null;if (hasFragmentation) {if (readBytes << 1 < config.getReadBufferSize()) {session.decreaseReadBufferSize();} else if (readBytes == config.getReadBufferSize()) {session.increaseReadBufferSize();}}}if (ret < 0) {scheduleRemove(session);}} catch (Throwable e) {if (e instanceof IOException) {if (!(e instanceof PortUnreachableException)|| !AbstractDatagramSessionConfig.class.isAssignableFrom(config.getClass())|| ((AbstractDatagramSessionConfig) config).isCloseOnPortUnreachable()) {scheduleRemove(session);}}IoFilterChain filterChain = session.getFilterChain();filterChain.fireExceptionCaught(e);} }

從這段代碼并結(jié)合上面的時(shí)序圖能夠看出來(lái)觸發(fā)整個(gè)FilterChain的調(diào)用以及IoHandler的調(diào)用。

三、類結(jié)構(gòu)分析

參考第一部分的總體結(jié)構(gòu)圖，畫一下每一個(gè)部分大致的類結(jié)構(gòu)圖：

簡(jiǎn)述： 從類繼承結(jié)構(gòu)圖來(lái)看，能夠看到在IOService體系下，存在IoConnector和IoAcceptor兩個(gè)大的分支體系。IoConnector是做為client的時(shí)候使用，IoAcceptor是作為服務(wù)端的時(shí)候使用。實(shí)際上在Mina中，有三種worker線程各自是：Acceptor、Connector 和 I/O processor。

(1) Acceptor Thread 作為server端的鏈接線程，實(shí)現(xiàn)了IoService接口。線程的數(shù)量就是創(chuàng)建SocketAcceptor的數(shù)量。

(2) Connector Thread 作為client請(qǐng)求建立的鏈接線程，實(shí)現(xiàn)了IoService接口，維持了一個(gè)和服務(wù)端Acceptor的一個(gè)鏈接，線程的數(shù)量就是創(chuàng)建SocketConnector的數(shù)量。

(3) I/O processorThread 作為I/O真正處理的線程，存在于server端和client。線程的數(shù)量是能夠配置的，默認(rèn)是CPU個(gè)數(shù)+1。

上面那個(gè)圖僅僅是表述了IoService類體系，而I/O Processor的類體系并不在當(dāng)中，見(jiàn)下圖：

簡(jiǎn)述：IOProcessor主要分為兩種。各自是AprIOProcessor和NioProcessor，Apr的解釋見(jiàn)上文：ps:APR（Apache Protable Runtime Library，Apache可移植執(zhí)行庫(kù)）。

NioProcessor也是Nio的一種實(shí)現(xiàn)，用來(lái)處理client連接過(guò)來(lái)的請(qǐng)求。在Processor中會(huì)調(diào)用到 FilterChain 和 Handler，見(jiàn)上文代碼。先看下FilterChain的類結(jié)構(gòu)圖例如以下：

Filter 和 Handler的類結(jié)構(gòu)例如以下：

Handler的類結(jié)構(gòu)例如以下：

Mina的session類結(jié)構(gòu)圖例如以下：

Mina的Buffer的類結(jié)構(gòu)圖例如以下：

版權(quán)聲明：本文博主原創(chuàng)文章，博客，未經(jīng)同意不得轉(zhuǎn)載。

總結(jié)

以上是生活随笔為你收集整理的MINA2 源代码学习--源代码结构梳理的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

如果覺(jué)得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。