mina源码剖析

最近阅读了mina的源代码，这里给大家做个分享:

一、mina server请求处理模型

下面的几张图对大家后面的理解会有一定的帮助，先贴出来:

1.acceptor模型:

 2.filterchain和processor处理模型

 

 

二、mina server核心组件剖析

1.acceptor:

首先附上一张acceptor的uml图,方面大家了阅读源码时了解到我们讲到的一些细节：

 
mina框架中负责监听连接的类，使用单个线程运行。使用nio的selector实现。将serversocketchannel的accept注册到selector中，使用单个线程轮训选择器的方式不断检测是否有新的连接进入。在Acceptor类中你会看到如下代码：

while (selectable) {
                try {
              
                    int selected = select();

                    nHandles += registerHandles();

                    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) {
                
                        processHandles(selectedHandles());
                    }                    
                } 

 我们从registerHandles方法开始读起，这方法中会在acceptor自己的选择器中注册serversocketchannel，并将accept标记为感兴趣的事件。选择器是java nio中异步的基础，这里我做下简单的介绍。首先有几个概念需要提一下，1.

SelectableChannel 2.selector 3.selectkey。java nio中所有继承至SelectableChannel的类都是可以做异步io的，那什么是异步io呢？这是相对于同步io来说的，如果你使用的是同步io，当你需要从本机socket中读数据，但是socket的buffer中又没有数据的情况下，当前线程是要被block住的。而使用异步io，即使socket的buffer中没有数据，当前线程可以不被阻塞掉，而可以去做其他事情。selector是java能做异步io的重要组件，其最重要的功能就是就绪选择，这是java socket异步通信的基础。每个SelectableChannel都可以将自己感兴趣的操作注册到selector中，当自己感兴趣的操作被选择器确定为已就绪时，SelectableChannel就可以进行相应的处理。那你肯定要问SelectableChannel和selector是通过什么联系起来的？答案就是selectkey，每一个注册进selector的SelectableChannel都会生成一个selectkey，这个selectkey标示了当前SelectableChannel的状态。selector中维护了三个selectkey集合：1.已选择的selectkey集合。2.已注册的selectkey集合。3.已取消的selectkey集合。每次调用selector的select方法时，都会更新一次选择键的集合，这里我们比较关心的是已选择的selectkey集合。mian中的acceptor和nioprocessor都是使用selector实现的，这就让单个线程可以处理很多请求，你可能不会想像到，mina维护socket连接的线程只有一个(acceptor)，你在稍后的代码中就可以看到。先看下serversocketchannle是如何在selector中注册自身的：

private int registerHandles() {
        for (;;) {
            // The register queue contains the list of services to manage
            // in this acceptor.
            AcceptorOperationFuture future = registerQueue.poll();

            if (future == null) {
                return 0;
            }

            // We create a temporary map to store the bound handles,
            // as we may have to remove them all if there is an exception
            // during the sockets opening.
            Map<SocketAddress, H> newHandles = new ConcurrentHashMap<SocketAddress, H>();
            List<SocketAddress> localAddresses = future.getLocalAddresses();

            try {
                // Process all the addresses
                for (SocketAddress a : localAddresses) {
                    H handle = open(a);
                    newHandles.put(localAddress(handle), handle);
                }
..............

 再次进入到open方法：

        // Creates the listening ServerSocket
        ServerSocketChannel channel = ServerSocketChannel.open();

        boolean success = false;

        try {
            // This is a non blocking socket channel
            channel.configureBlocking(false);

            // Configure the server socket,
            ServerSocket socket = channel.socket();

            // Set the reuseAddress flag accordingly with the setting
            socket.setReuseAddress(isReuseAddress());

            // and bind.
            socket.bind(localAddress, getBacklog());

            // Register the channel within the selector for ACCEPT event
            channel.register(selector, SelectionKey.OP_ACCEPT);
            success = true;
        } finally {
            if (!success) {
                close(channel);
            }
        }
        return channel;
    

 如上所示，serversocketchannle将自身注册到了acceptor的selector当中，并将accept事件标记为感兴趣的事件，每当有新的socket连接建立时，accept事件都会被触发，当调用selector的select方法时，selector中的已选择的选择键集合就会被更新，然后通过选择键可以获取到serversocketchannel，调用serversocketchannel的accept方法就可以拿到新的socket连接的socketchannel。这个在稍后负责建立连接的acceptor线程中可以看到。我们继续阅读第一段代码，你会看到：

if (selected > 0) {
                
                        processHandles(selectedHandles());
                    } 

selected是selector.select()的结果，当有通道被选择器选择时，selected的值就会>0。selectedHadnles方法会返回一个可遍历的已选择的选择键集合,我们直接进入processHandles方法:

 

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
                S session = accept(processor, handle);

                if (session == null) {
                    continue;
                }

                initSession(session, null, null);

                // add the session to the SocketIoProcessor
                session.getProcessor().add(session);
            }
        }
    }

 以上代码对以选择的选择键进行了遍历,accept选择键遍历出，然后从中获取新的连接对应的serversocketchannel并将其放入到NioSession中，mina会将每一个连接对应的socketchannel都放入到一个新的niosession中，你可以将niosession理解为对一个连接会话的抽象。稍后我会着重将下niosession的实现，里面有一个filterchain是mina可扩展业务模型的核心。现在我们进入accept方法中看下：

@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 client
        SocketChannel ch = handle.accept();

        if (ch == null) {
            return null;
        }

        return new NioSocketSession(this, processor, ch);
    }

 正如我刚才说的这里会拿出新连接的socketchannel并将他放入到niosesison中。NioSocketSession对象的构造函数中的processor是一个processor对象池，这是在acceptor初始化的时候建立的，接下来我们就进入niosession的源码阅读。

 

2.niosession

和上面一样，首先附上一张niosession的uml图：

 niosocketSession中一个比较核心的东西就是filterchain,看到filter就觉得很熟悉吧？会把他和用户自定义的filter进行联系？没错这个filterchain就包含了用户自定义的filter以及用户自定义的handler。看到filter我的第一感觉就是mina应该会用职责链的设计模式来将filterchain做成可扩展的设计。我们进入NioSession看看filterchain是如何被初始化的：

protected NioSession(IoProcessor<NioSession> processor, IoService service, Channel channel) {
        super(service);
        this.channel = channel;
        this.processor = processor;
        filterChain = new DefaultIoFilterChain(this);
    }

接着进入DefaultIoFilterChain的构造函数：

public DefaultIoFilterChain(AbstractIoSession session) {
        if (session == null) {
            throw new IllegalArgumentException("session");
        }

        this.session = session;
        head = new EntryImpl(null, null, "head", new HeadFilter());
        tail = new EntryImpl(head, null, "tail", new TailFilter());
        head.nextEntry = tail;
    }

这是比较关键的地方，filterchain在初始化的时候会构建两个默认的filter：headfilter和tailfiter，看他们的名字应该能想到他们是处理链的头和尾。这两个filter分别被封装进两个entry中，并组成了entry链，进入EntryImpl的构造方法:

private EntryImpl(EntryImpl prevEntry, EntryImpl nextEntry, String name, IoFilter filter) {
            if (filter == null) {
                throw new IllegalArgumentException("filter");
            }
            if (name == null) {
                throw new IllegalArgumentException("name");
            }

            this.prevEntry = prevEntry;
            this.nextEntry = nextEntry;
            this.name = name;
            this.filter = filter;
            this.nextFilter = new NextFilter() {
                public void sessionCreated(IoSession session) {
                    Entry nextEntry = EntryImpl.this.nextEntry;
                    callNextSessionCreated(nextEntry, session);
                }

                public void sessionOpened(IoSession session) {
                    Entry nextEntry = EntryImpl.this.nextEntry;
                    callNextSessionOpened(nextEntry, session);
                }

                public void sessionClosed(IoSession session) {
                    Entry nextEntry = EntryImpl.this.nextEntry;
                    callNextSessionClosed(nextEntry, session);
                }

                public void sessionIdle(IoSession session, IdleStatus status) {
                    Entry nextEntry = EntryImpl.this.nextEntry;
                    callNextSessionIdle(nextEntry, session, status);
                }

                public void exceptionCaught(IoSession session, Throwable cause) {
                    Entry nextEntry = EntryImpl.this.nextEntry;
                    callNextExceptionCaught(nextEntry, session, cause);
                }

                public void messageReceived(IoSession session, Object message) {
                    Entry nextEntry = EntryImpl.this.nextEntry;
                    callNextMessageReceived(nextEntry, session, message);
                }

                public void messageSent(IoSession session, WriteRequest writeRequest) {
                    Entry nextEntry = EntryImpl.this.nextEntry;
                    callNextMessageSent(nextEntry, session, writeRequest);
                }

                public void filterWrite(IoSession session, WriteRequest writeRequest) {
                    Entry nextEntry = EntryImpl.this.prevEntry;
                    callPreviousFilterWrite(nextEntry, session, writeRequest);
                }

                public void filterClose(IoSession session) {
                    Entry nextEntry = EntryImpl.this.prevEntry;
                    callPreviousFilterClose(nextEntry, session);
                }

                public String toString() {
                    return EntryImpl.this.nextEntry.name;
                }
            };
        }

 从上面的代码可以看到，每个entryimpl都包含preentry、nextentry、filter、nextfilter这几个关键的属性。你现在能想像到mina的职责链是如何设计的吗？我们随机挑选一个call**方法就很清楚了：

private void callNextSessionCreated(Entry entry, IoSession session) {
        try {
            IoFilter filter = entry.getFilter();
            NextFilter nextFilter = entry.getNextFilter();
            filter.sessionCreated(nextFilter, session);
        } catch (Throwable e) {
            fireExceptionCaught(e);
        }
    }

 哈哈，真相就在这里，注意方法中entry入參，这里的entry对象传入的是当前entry的nextentry。这里执行了filterchain中当前链节点下一个链节点的xxx方法，而nexfilterchain中的方法又对下下个链节点中的filter进行了xxx方法的调用。可以想到，如果要在处理链中加入自己的filter，你只需要在filter中实现自己的业务逻辑，在最后调用下xxx方法，并把nextfilter做成如參传入。关于用户自定义的链是如何被加入到filterchain中的以及handler是如何被扩展、如何被处理的，我在稍后会提到。大概过完niosession的filterchain，现在我们继续进入下一个环节，nioprocessor.

 

3.processor

我们回到刚才的主线代码继续阅读：

进入acceptor的processHandles方法：

 

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
                S session = accept(processor, handle);

                if (session == null) {
                    continue;
                }

                initSession(session, null, null);

                // add the session to the SocketIoProcessor
                session.getProcessor().add(session);
            }
        }

 可以看到，创建完session对象后，就将session对象放入到nioprocessor中。我们进入SimpleIoProcessorPool的add方法中，这个方法会从processorpool中选择出一个processor，并将session放入到processor的队列中，代码如下:

 

 

private IoProcessor<S> getProcessor(S session) {
        IoProcessor<S> processor = (IoProcessor<S>) session.getAttribute(PROCESSOR);

        if (processor == null) {
            if (disposed || disposing) {
                throw new IllegalStateException("A disposed processor cannot be accessed.");
            }

            processor = pool[Math.abs((int) session.getId()) % pool.length];

            if (processor == null) {
                throw new IllegalStateException("A disposed processor cannot be accessed.");
            }

            session.setAttributeIfAbsent(PROCESSOR, processor);
        }

        return processor;
    }

 通过一个取模算法，随机从processorpool中取出一个processor，我们继续进入nioprocessor继承的抽象类AbstractPollingIoProcessor的add方法中：

 

 

 public final void add(S session) {
        if (disposed || disposing) {
            throw new IllegalStateException("Already disposed.");
        }

        // Adds the session to the newSession queue and starts the worker
        newSessions.add(session);
        startupProcessor();
    }

 可以看到这个方法将session加入到了自己的队列中。读到这里可以猜想到，后面肯定会有地方不断的pool newSessions队列中的数据去处理，我们接着往下走，进入startupProcessor方法：

 

 

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();
    }

 正如上所示，如果Processor对象不为空，则立即唤醒选择器，你一定会问选择器在什么时候会被阻塞？这个问题问得好，当注册到选择器上所有通道的服务都不可用时，这个时候选择器会被阻塞掉。如果你调用的选择器时使用的是不带任何入參的select方法时，这个时候选择器会永远阻塞下去，mina processor的选择器是使用的带参数的select方法。所以要使用wakeup方法去唤醒阻塞的选择器，让它重新选择，当选择器没有阻塞时调用wakeup不会产生什么影响。我们接着进入Processor线程，看它的内部实现：

 

 

 // Manage newly created session first
                    nSessions += handleNewSessions();

                    updateTrafficMask();

                    // Now, if we have had some incoming or outgoing events,
                    // deal with them
                    if (selected > 0) {
                        //LOG.debug("Processing ..."); // This log hurts one of the MDCFilter test...
                        process();
                    }

                    // Write the pending requests
                    long currentTime = System.currentTimeMillis();
                    flush(currentTime);

                    // And manage removed sessions
                    nSessions -= removeSessions();

 进入handleNewSessions：

 

 

private int handleNewSessions() {
        int addedSessions = 0;

        for (S session = newSessions.poll(); session != null; session = newSessions.poll()) {
            if (addNow(session)) {
                // A new session has been created
                addedSessions++;
            }
        }

        return addedSessions;
    }

 可以看到，这里是在不停的将newSession中的session对象poll出，然后调用addNow对session中的socketchannel进行read操作的注册，同时将用户自定义的filter插入到session的filterchain当中。进入addNow方法：

 

 

private boolean addNow(S session) {
        boolean registered = false;

        try {
            init(session);
            registered = true;

            // Build the filter chain of this session.
            IoFilterChainBuilder chainBuilder = session.getService().getFilterChainBuilder();
            chainBuilder.buildFilterChain(session.getFilterChain());

            // DefaultIoFilterChain.CONNECT_FUTURE is cleared inside here
            // in AbstractIoFilterChain.fireSessionOpened().
            // Propagate the SESSION_CREATED event up to the chain
            IoServiceListenerSupport listeners = ((AbstractIoService) session.getService()).getListeners();
            listeners.fireSessionCreated(session);
        } catch (Throwable e) {
            ExceptionMonitor.getInstance().exceptionCaught(e);

            try {
                destroy(session);
            } catch (Exception e1) {
                ExceptionMonitor.getInstance().exceptionCaught(e1);
            } finally {
                registered = false;
            }
        }

        return registered;
    }

 init方法中将注册session中socketchannel的read操作:

protected void init(NioSession session) throws Exception {
        SelectableChannel ch = (SelectableChannel) session.getChannel();
        ch.configureBlocking(false);
        session.setSelectionKey(ch.register(selector, SelectionKey.OP_READ, session));
    }

 ch.configureBlocking(false);这行代码比较关键，在每次将通道注册到selector中时都需要将通道定义为非阻塞的。否则选择器无法生效。接着我们看看用户自定义的filterchain是如何被加入到session的filterchain当中的，进入DefaultIoFilterChainBuilder的buildFilterChain方法：

public void buildFilterChain(IoFilterChain chain) throws Exception {
        for (Entry e : entries) {
            chain.addLast(e.getName(), e.getFilter());
        }
    }

 可以看到方法中只是简单的调用了filterchain的addLast方法，进入DefaultIoFilterChain的addLast方法：

public synchronized void addLast(String name, IoFilter filter) {
        checkAddable(name);
        register(tail.prevEntry, name, filter);
    }

private void register(EntryImpl prevEntry, String name, IoFilter filter) {
        EntryImpl newEntry = new EntryImpl(prevEntry, prevEntry.nextEntry, name, filter);

        try {
            filter.onPreAdd(this, name, newEntry.getNextFilter());
        } catch (Exception e) {
            throw new IoFilterLifeCycleException("onPreAdd(): " + name + ':' + filter + " in " + getSession(), e);
        }

        prevEntry.nextEntry.prevEntry = newEntry;
        prevEntry.nextEntry = newEntry;
        name2entry.put(name, newEntry);

        try {
            filter.onPostAdd(this, name, newEntry.getNextFilter());
        } catch (Exception e) {
            deregister0(newEntry);
            throw new IoFilterLifeCycleException("onPostAdd(): " + name + ':' + filter + " in " + getSession(), e);
        }
    }

 

可以看到，这里将用户自定义的每一个filter都封装进entry中，并将entry放入到entry链的倒数第二个位置构造新的entry链。你一定会有疑问，这里只讲到了用户自定义的filter是如何被加入到filterchain中的，而没有讲到用户自定义的handler是怎么被加入到链中的？这是个很好的问题，handler是在什么时候被调用的？答案就在TailFilter中，进入TailFilter你就会明白:

private static class TailFilter extends IoFilterAdapter {
        public void sessionCreated(NextFilter nextFilter, IoSession session)
                throws Exception {
            session.getHandler().sessionCreated(session);
        }

        public void sessionOpened(NextFilter nextFilter, IoSession session)
                throws Exception {
            try {
                session.getHandler().sessionOpened(session);
            } finally {
                // Notify the related ConnectFuture
                // if the session is created from SocketConnector.
                ConnectFuture future = (ConnectFuture) session
                        .removeAttribute(CONNECT_FUTURE);
                if (future != null) {
                    future.setSession(session);
                }
            }
        }

        public void sessionClosed(NextFilter nextFilter, IoSession session)
                throws Exception {
            try {
                session.getHandler().sessionClosed(session);
            } finally {
                // Remove all filters.
                session.getFilterChain().clear();
            }
        }
....}

 看到了吧，这里处理了所有handler的方法，所以mina的结构是可以有多个filter用户自定义扩展，但是只能有一个handler自定义扩展，至于这样设计的好处和不好的地方，如果有兴趣可以私下讨论。到这里，mina框架的核心部分做了个简单的介绍，细节的地方我会继续补充。