深入阅读Mina源码(3) —— Mina之IOAdapter(一)

1. 预览

 

     依然遵循之前的原则，只看core(其实还包括了transport,虽然它不在core中，但socket相关的类在此包中)，了解mina的内部机制，忽略细节，这里就直接从服务器的入口第一步开始，即连接管理的IOService。直接来看下对应的包：org.apache.mina.core.service。 by the way 我的mina源码版本是：2.0.7

 

 

 

     通过包结构就可以说一下这部分的职能和相关的代码，先有个整个的印象。

     可以总结成五部分：service责任（IoService）、Processor线程处理（IoProcessor）、handler处理器（IoHandler）、服务器端接收器（IoAcceptor）和客户端连接器（IoConnector），括号中是对应的处理接口，我们首先只讨论服务器架构图入口部分，即1、4、5。另外handler部分属于流程的末端，IoProcessor和IoSession密切相连，负责完成IoSession中一些实际的IO操作，不太适合在这部分介绍。

 

 

2. IOService

 

      第一部分先说一下顶层的接口IOService，官方的介绍上给出了IOService的职能和abstract封装了一层的抽象类

 

 

    对应着这几个功能看一下IoService的所有方法就比较清楚各个方法的作用了。因为是接口，看了这个接口大概就明白是做什么的了。

    AbstractIoService的注释提到An instance of IoService contains an Executor which will handle the incoming events。而上面提到的不止这个功能，照着上面的功能去看相应的方法是个比较好的办法，就可以了解每部分怎么实现的，一眼看过去有些add，set，contain之类的方法都可以忽略了，这里最后的manages不要被字面误解，搜一下源码里的管理，其实只是maintain。那么就剩下了两个方法值得我们关注，一个就是注释提到的executor，另一个就是涉及到默认实现的dispose。

 

    Creates Executor, if not provided(在构造方法中实现，觉着有些地方同样需要提一下跟整体的设计相关的，就增大的信息量，把构造方法都贴出来了):

 

 

protected AbstractIoService(IoSessionConfig sessionConfig, Executor executor) {
        if (sessionConfig == null) {
            throw new IllegalArgumentException("sessionConfig");
        }

        if (getTransportMetadata() == null) {
            throw new IllegalArgumentException("TransportMetadata");
        }

        if (!getTransportMetadata().getSessionConfigType().isAssignableFrom(sessionConfig.getClass())) {
            throw new IllegalArgumentException("sessionConfig type: " + sessionConfig.getClass() + " (expected: "
                    + getTransportMetadata().getSessionConfigType() + ")");
        }

        // Create the listeners, and add a first listener : a activation listener
        // for this service, which will give information on the service state.
        listeners = new IoServiceListenerSupport(this);
        listeners.add(serviceActivationListener);

        // Stores the given session configuration
        this.sessionConfig = sessionConfig;

        // Make JVM load the exception monitor before some transports
        // change the thread context class loader.
        ExceptionMonitor.getInstance();

        if (executor == null) {
            this.executor = Executors.newCachedThreadPool();
            createdExecutor = true;
        } else {
            this.executor = executor;
            createdExecutor = false;
        }

        threadName = getClass().getSimpleName() + '-' + id.incrementAndGet();
}

 

 

        只是看一下上面提到的功能，这里多提一下就是 ExceptionMonitor.getInstance(); 有兴趣去看一下org.apache.mina.uti.DefaultExceptionMonitor 很简短，只有一个方法，这里只调用一下就是实例化DefaultExceptionMonitor ：

 

 

public void exceptionCaught(Throwable cause) {
        if (cause instanceof Error) {
            throw (Error) cause;
        }

        LOGGER.warn("Unexpected exception.", cause);
}

 

 

      我想大概明白是怎么一回事了，这里对Exception的处理做了初始化，这个位置也比较合适的选择，Mina将异常处理的日志打印放在了单独的类中进行处理uncaught exceptions。

 

      回到之前的话题，看到mina是通过线程池来装载这些服务的，如果你像我一样发现自己弱爆了concurrent包没用过就到网上看看吧，在学习过程中大致熟悉了之后再发相关的博客，现在发肯定要误导大众了。

      这里只需要知道下面这句话就ok了：并发编程的一种编程方式是把任务拆分为一些列的小任务，即Runnable，然后在提交给一个Executor执行，Executor.execute(Runnalbe) 。Executor在执行时使用内部的线程池完成操作。

 

      下面来看dispose方法，这段代码算是学习并发包一个很好的例子了：

 

 

protected final Object disposalLock = new Object();
private volatile boolean disposing;
private volatile boolean disposed;

public final void dispose(boolean awaitTermination) {
        if (disposed) {
            return;
        }

        synchronized (disposalLock) {
            if (!disposing) {
                disposing = true;

                try {
                    dispose0();
                } catch (Exception e) {
                    ExceptionMonitor.getInstance().exceptionCaught(e);
                }
            }
        }

        if (createdExecutor) {
            ExecutorService e = (ExecutorService) executor;
            e.shutdownNow();
            if (awaitTermination) {

                //Thread.currentThread().setName();

                try {
                    LOGGER.debug("awaitTermination on {} called by thread=[{}]", this, Thread.currentThread().getName());
                    e.awaitTermination(Integer.MAX_VALUE, TimeUnit.SECONDS);
                    LOGGER.debug("awaitTermination on {} finished", this);
                } catch (InterruptedException e1) {
                    LOGGER.warn("awaitTermination on [{}] was interrupted", this);
                    // Restore the interrupted status
                    Thread.currentThread().interrupt();
                }
            }
        }
        disposed = true;
}

 

 

    dispose的作用在官网的IOAdapter中明确提到了，还是尽量使用官方的解释：

 

 

Disposal

The service can be stopped by calling the dispose() method. The service will be stopped only when all the pending sessions have been processed :

// Stop the service, waiting for the pending sessions to be inactive
acceptor.dispose();

You can also wait for every thread being executed to be properly completed by passing a boolean parameter to this method :

// Stop the service, waiting for the processing session to be properly completed
acceptor.dispose( true );

 

 

      那顺着它的作用看下源码，仅仅是内容大都比较好理解，主要是学习下实现的机制，这里用了volatile关键字修饰布尔型变量的方法，同时使用内置锁机制控制线程的访问。

     dispose0用于子类实现，来释放相应的资源，createdExecutor在刚才看到的构造函数中初始化，对应是否做了Executors.newCachedThreadPool()开启线程池。

     推荐一篇volatile的文章：http://www.ibm.com/developerworks/cn/java/j-jtp06197.html，使用需谨慎，多学习优秀做法。

 

 

3. IOAdapter

 

 

    We have many of those implementing classes

• NioSocketAcceptor : the non-blocking Socket transport IoAcceptor
• NioDatagramAcceptor : the non-blocking UDP transport IoAcceptor
• AprSocketAcceptor : the blocking Socket transport IoAcceptor, based on APR
• VmPipeSocketAcceptor : the in-VM IoAcceptor

 

    又来张图片，木有办法，mina官网的资料就是那么给力了，因此User Guide还是必备资料。这里还是要复制大多数入门篇都在写的那段官方Quick start guide的服务器代码了：

 

 

public class MinaTimeServer
{
    private static final int PORT = 9123;

    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 TimeServerHandler() );
        acceptor.getSessionConfig().setReadBufferSize( 2048 );
        acceptor.getSessionConfig().setIdleTime( IdleStatus.BOTH_IDLE, 10 );
        acceptor.bind( new InetSocketAddress(PORT) );
    }
}

 

 

 

     先走上面的图左边走下去，NioSocketAcceptor应该是最常用到的，Mina构造socket服务器也是近期项目需要用到的。

     首先IoAcceptor在IoService基础之上又实现了一些自己定义的方法，主要是bind系列和SocketAddress相关系列，看到要到NioSocketAcceptor左右都需要继续向下一层。继续粘图就太多了，这里可以对应一下类的方法列表看，SocketAdaptor出现了分叉，也就是和DatagramAcceptor有区分的定义了一些新的方法，但是方法很少，先不详细说明。同时比如getDefaultLocalAddress方法的返回值缩小了范围，变成了SocketAddress的子类InetSocketAddress，同时这部分的代码都已经在了org.apache.mina.transport.socket包下了。

     左边已经都到的NioSocketAcceptor的直接实现接口，那么继续开始从右边向下，首先是继承自AbstractIoService的AbstractIoAcceptor，主要为了实现IoAcceptor通用的方法，就是刚才说到的bind系列和SocketAddress相关系列。先看一下定义的几个属性

 

 

    private final List<SocketAddress> defaultLocalAddresses = new ArrayList<SocketAddress>();

    private final List<SocketAddress> unmodifiableDefaultLocalAddresses = Collections
            .unmodifiableList(defaultLocalAddresses);

    private final Set<SocketAddress> boundAddresses = new HashSet<SocketAddress>();

    private boolean disconnectOnUnbind = true;

    /**
     * The lock object which is acquired while bind or unbind operation is performed.
     * Acquire this lock in your property setters which shouldn't be changed while
     * the service is bound.
     */
    protected final Object bindLock = new Object();

 

 

     疑问马上来了：

 

• 为什么要用List形式的defaultLocalAddresses来维护本地的默认地址
• bind的是地址维护用的Set，而defaultLocalAddresses用的List，这是为什么

     那就一个一个来回答吧：

 

• 使用List来维持本地地址是考虑到了机器多网口的问题，可见Mina设计的细致。
• 为什么两个要用不同的方式来维护呢，原因很简单。先看一下getDefaultLocalAddress返回的是return defaultLocalAddresses.iterator().next(); 而另个bound的地址，想一下unbind后会怎么操作了呢，再看一眼HashSet，相信凭直觉就知道了原因，mina可以用一个IoAdapter绑定多个端口进行监听，而更好的管理是必要的，也就是为什么使用了HashSet，而非List。

 

     这里提到了getDefaultLocalAddress方法就对应了另一个getLocalAddress方法，getLocalAddress返回的是boundAddresses拷贝的一份Set中的next，需要注意这一点，因为他们都是public方法。已经长篇大论了，这个细节的地方提示注意一下，就不贴代码了。

     另一个提一下就是unmodifiableList这个方法，简直就是刚看过的Effective java中的‘使类和成员的可访问性最小化’中讲解的完美实例，学习就是在这么一次次的突然新鲜东西和学习的东西碰撞中产生乐趣和不断熟悉的，突然有种两者都恍然大悟的感觉，扯远了。说一下这个方法是对list做一个不可变的拷贝，因为后面有提供返回list的方法提供，因此需要维护这样一个变量（因为长度非0的数组总是可变的，再深入就不是这里要讨论的了）。

     接下来就是这个AbstractIoAcceptor类的核心两个方法了，其他一切方法都是围绕他们进行的：bind和setDefaultLocalAddresses，本想都拿代码出来说一下的，但是太多了，就说一下setDefaultLocalAddresses

 

 

public final void setDefaultLocalAddresses(Iterable<? extends SocketAddress> localAddresses) {
        if (localAddresses == null) {
            throw new IllegalArgumentException("localAddresses");
        }

        synchronized (bindLock) {
            synchronized (boundAddresses) {
                if (!boundAddresses.isEmpty()) {
                    throw new IllegalStateException("localAddress can't be set while the acceptor is bound.");
                }

                Collection<SocketAddress> newLocalAddresses = new ArrayList<SocketAddress>();

                for (SocketAddress a : localAddresses) {
                    checkAddressType(a);
                    newLocalAddresses.add(a);
                }

                if (newLocalAddresses.isEmpty()) {
                    throw new IllegalArgumentException("empty localAddresses");
                }

                this.defaultLocalAddresses.clear();
                this.defaultLocalAddresses.addAll(newLocalAddresses);
            }
        }
}

 

 

      这里想说的就是其中体现到的逻辑，一定是在没有绑定之前处理的defaultLocalAddresses。几次synchronized保证的线程安全是很重要的，如果需要继承这些类的时候是记得利用这两处的synchronized
 和知道基础方法中有使用synchronized都是很重要的。

      bind是这个方法：public final void bind(Iterable<? extends SocketAddress> localAddresses) throws IOException，需要提一下的就是最后这一句：

 

 

if (activate) {
      getListeners().fireServiceActivated();
}

     与前面的提到的Mina事件驱动机制联系起来了，最终调用的方法的注释为：Invoked when a new service is activated by an IoService.

 

（发现篇幅有点太长了，就分成两部分吧，最后一个抽象类和一个最终的实现类放在下一篇）