NIO框架之MINA源代码解析(二):mina核心引擎

NIO框架之MINA源代码解析(一):背景

MINA的底层还是利用了jdk提供了nio功能,mina仅仅是对nio进行封装。包含MINA用的线程池都是jdk直接提供的。

MINA的server端主要有accept、processor、session三部分组成的。当中accept主要负责在指定的port监听。若有新连接则建立一个新的session。processor则负责处理session相应的发送数据和接收数据并调用上层处理;session则缓存当前连接数据

MINA採用了线程懒启动的技术,即最少启动线程。在MINA server启动的时候,仅仅有一个线程-accept,而且accept线程仅仅有一个,在指定的port进行监听(能够同一时候监听多个port,mina能够绑定多port)。

1、acceptor

先看下acceptor的主要类图吧。

watermark/2/text/aHR0cDovL2Jsb2cuY3Nkbi5uZXQvY2hhb2ZhbndlaQ==/font/5a6L5L2T/fontsize/400/fill/I0JBQkFCMA==/dissolve/70/gravity/Center" />

mina server的启动入口是在NioSocketAcceptor.bind(InetSocketAddress)或者NioSocketAcceptor.bind(SocketAddress...)方法, acceptor.bind(new InetSocketAddress(1234));  
然后会调用AbstractPollingIoAcceptor.bindInternal(List<? extends SocketAddress>)方法,在bindInternal方法里面会调用startupAcceptor()方法提交一个accept线程到线程池里面(仅仅提交一次),并初始化acceptor端的Selector。就这样一个acceptor线程启动了。

acceptor端业务相对简单,相当于在当前Selector里面监听acceptor事件,处理新连接并新建一个session放到相应的processor里面。

acceptor 代码。非常easy。

 private class Acceptor implements Runnable {
        public void run() {
            assert (acceptorRef.get() == this);

            int nHandles = 0;

            // Release the lock
            lock.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
                    int selected = select();

                    // this actually sets the selector to OP_ACCEPT,
                    // and binds to the port on which this class will
                    // listen on  在通道里面注冊连接事件
                    nHandles += 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 loop
                    break;
                } 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();
                    }
                }
            }
        }

2、processor

processor顾名思义,就是进行IO处理,处理当前session的数据读写,并进行业务处理。

在mina server初始化的时候,会初始化一个processor池,通过NioSocketAcceptor的构造器传入池的大小,默认是当前处理器的个数+1。

processor池里面有一个jdk提供的 线程池 - Executors.newCachedThreadPool()。各个processor线程会引用此线程池,即每一个processor线程都在这个线程池里面运行。

在mina server实际处理时,每一个processor相当于一个线程,轮流处理当前的session队列里面的数据(每一个processor里面的session相当于顺序处理,共享一个线程)。

每一个processor有一个Selector对象

processor类图

processor端的处理逻辑相对有点复杂。看以下的流程图。

1、把新加入进来的session注冊到当前processor的Selector里面的read事件,并初始化session;
2、推断当前Selector是否有读写事件。
3、若第2步有读事件时。进入步骤4,若没有的话,直接到第6步。
4、处理当前读事件,并把处理后的数据放入到flush队列里面。
5、把第4步运行的结果flush到client;
6、处理session。比方session idle时间等。

7、又一次运行第1步,循环运行。

processor端代码。

 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();
                    int 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 false
                            wakeupCalled.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 false
                        wakeupCalled.getAndSet(false);

                        // and continue the loop
                        continue;
                    }

                    // Manage newly created session first  处理新加入进来的session,并注冊到当前processor的Selector读事件
                    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...处理读事件。并把结果放入flush队列里面
                        process();
                    }

                    // Write the pending requests  把flush队列里面的session的处理完的数据发送给client
                    long currentTime = System.currentTimeMillis();
                    flush(currentTime);

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

                    // Last, not least, send Idle events to the idle sessions
                    notifyIdleSessions(currentTime);

                    // Get a chance to exit the infinite loop if there are no
                    // more sessions on this Processor
                    if (nSessions == 0) {
                        processorRef.set(null);

                        if (newSessions.isEmpty() && isSelectorEmpty()) {
                            // newSessions.add() precedes startupProcessor
                            assert (processorRef.get() != this);
                            break;
                        }

                        assert (processorRef.get() != this);

                        if (!processorRef.compareAndSet(null, this)) {
                            // startupProcessor won race, so must exit processor
                            assert (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 loop
                    break;
                } 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);
            }
        }
    }

3、session

session做为一个连接的详细对象,缓存当前连接用户的一些信息。

session类图

watermark/2/text/aHR0cDovL2Jsb2cuY3Nkbi5uZXQvY2hhb2ZhbndlaQ==/font/5a6L5L2T/fontsize/400/fill/I0JBQkFCMA==/dissolve/70/gravity/Center" />

session对象是绑定在SelectableChannel的一个attach。

class  NioProcessor
 @Override
    protected void init(NioSession session) throws Exception {
        SelectableChannel ch = (SelectableChannel) session.getChannel();
        ch.configureBlocking(false);
        session.setSelectionKey(ch.register(selector, SelectionKey.OP_READ, session));//chanel 注冊读事件,并把session当做一个attach辅导SelectableChannel里面。
    }
时间: 2024-08-06 02:09:06

NIO框架之MINA源代码解析(二):mina核心引擎的相关文章

NIO框架之MINA源代码解析(一):背景

?? "你们的agent占了好多系统的port.把我们的非常多业务系统都给整死了,给我们造成了非常大的损失.要求你们的相关领导下周过来道歉"?? --?? 来自我们的一个客户. ?怎么可能呢,我们都不相信,我们的agent仅仅占一个port啊! 事实胜过雄辩.经过查证.确实是因为我们的agent占了好多系统的port.我看了一下日志.基本把系统可用的port占完了! watermark/2/text/aHR0cDovL2Jsb2cuY3Nkbi5uZXQvY2hhb2ZhbndlaQ=

【原创】NIO框架入门(一):服务端基于Netty4的UDP双向通信Demo演示

申明:本文由作者基于日常实践整理,希望对初次接触MINA.Netty的人有所启发.如需与作者交流,见文签名,互相学习. 学习交流 更多学习资料:点此进入 推荐 移动端即时通讯交流: 215891622 推荐 前言 NIO框架的流行,使得开发大并发.高性能的互联网服务端成为可能.这其中最流行的无非就是MINA和Netty了,MINA目前的主要版本是MINA2.而Netty的主要版本是Netty3和Netty4(Netty5已经被取消开发了:详见此文). 本文将演示的是一个基于Netty4的UDP服

【原创】NIO框架入门(三):iOS与MINA2、Netty4的跨平台UDP双向通信实战

前言 本文将演示一个iOS客户端程序,通过UDP协议与两个典型的NIO框架服务端,实现跨平台双向通信的完整Demo.服务端将分别用MINA2和Netty4进行实现,而通信时服务端你只需选其一就行了.同时用MINA2和Netty4分别实现服务端的目的,是因为很多人都在纠结到底是用MINA还是Netty来实现高并发的Java网络通信服务端,在此干脆两个都实现了,就看你怎么选择了,够吊吧. NIO框架的流行,使得开发大并发.高性能的互联网服务端成为可能.这其中最流行的无非就是MINA和Netty了,M

Spring源代码解析

Spring源代码解析(一):IOC容器:http://www.iteye.com/topic/86339 Spring源代码解析(二):IoC容器在Web容器中的启动:http://www.iteye.com/topic/86594 Spring源代码解析(三):Spring JDBC:http://www.iteye.com/topic/87034 Spring源代码解析(四):Spring MVC:http://www.iteye.com/topic/87692 Spring源代码解析(五

Spring源代码解析(收藏)

Spring源代码解析(收藏)Spring源代码解析(一):IOC容器:http://www.iteye.com/topic/86339 Spring源代码解析(二):IoC容器在Web容器中的启动:http://www.iteye.com/topic/86594 Spring源代码解析(三):Spring JDBC:http://www.iteye.com/topic/87034 Spring源代码解析(四):Spring MVC:http://www.iteye.com/topic/8769

NIO框架之MINA源码解析(五):NIO超级陷阱和使用同步IO与MINA通信

1.NIO超级陷阱 之所以说NIO超级陷阱,就是因为我在本系列开头的那句话,因为使用缺陷导致客户业务系统瘫痪.当然,我对这个问题进行了很深的追踪,包括对MINA源码的深入了解,但其实之所以会出现这个问题,它的根不是MINA的原因,而是JDK底层的问题. JDK底层在实现nio时,为了能够唤醒等待在io上的线程,在windows平台使用了两个端口建立连接发消息实现.看如下代码: [java] view plain copy print? public class NIOTest { @Test p

C#使用zxing,zbar,thoughtworkQRcode解析二维码,附源代码

最近做项目需要解析二维码图片,找了一大圈,发现没有人去整理下开源的几个库案例,花了点时间 做了zxing,zbar和thoughtworkqrcode解析二维码案例,希望大家有帮助. zxing是谷歌开源的二维码库,zbar,thoughtworkQRcode也是开源的,三者之间比较各有优劣 下面通过一个案例demo源码,来认识学习下这三者的实际解码效果, 第一次上传demo源码,献丑了 zbar解析关键代码: Image primaryImage = Image.FromFile(fileNa

SDWebImage源代码解析(二)

上一篇:SDWebImage源代码解析(一) 2.缓存 为了降低网络流量的消耗.我们都希望下载下来的图片缓存到本地.下次再去获取同一张图片时.能够直接从本地获取,而不再从远程server获取.这样做的还有一个优点是提升了用户体验,用户第二次查看同一幅图片时,能高速从本地获取图片直接呈现给用户. SDWebImage提供了对图片缓存的支持,而该功能是由SDImageCache类来完毕的.该类负责处理内存缓存及一个可选的磁盘缓存.当中磁盘缓存的写操作是异步的,这样就不会对UI操作造成影响. 2.1

【原创】NIO框架入门(四):Android与MINA2、Netty4的跨平台UDP双向通信实战

概述 本文演示的是一个Android客户端程序,通过UDP协议与两个典型的NIO框架服务端,实现跨平台双向通信的完整Demo. 当前由于NIO框架的流行,使得开发大并发.高性能的互联网服务端成为可能.这其中最流行的无非就是MINA和Netty了,MINA目前的主要版本是MINA2.而Netty的主要版本是Netty3和Netty4(Netty5已经被取消开发了:详见此文). 本文中,服务端将分别用MINA2和Netty4进行实现,但在你实际的项目中服务端实现只需选其一就行了.本文中的Demo同时