欢迎您访问程序员文章站本站旨在为大家提供分享程序员计算机编程知识!
您现在的位置是: 首页

Mina Nio处理器

程序员文章站 2022-03-11 09:29:40
...
Mina Io处理器抽象实现:http://donald-draper.iteye.com/blog/2377663
引言:
上一篇文章我们看了Io处理器的抽象实现,先来回顾一下:
     抽象Io处理器AbstractPollingIoProcessor,主要几个关键内部变量为选择操作超时时间SELECT_TIMEOUT,用于腾出时间,处理空闲的会话; executor处理器内部执行器,用于运行内部处理器Processor;存储Io处理器等线程最大线程id的threadIds(Map);创建会话队列newSessions用于存储新创建的会话;移除会话队列removingSessions用于存放从处理器移除的会话;刷新会话队列flushingSessions,用于存放要发送写请求的会话;次序控制会话队列trafficControllingSessions用于存放会话待读写的会话;Io处理器线程引用processorRef。
     添加会话首先添加会话到Io处理器的创建会话队列中,启动处理器线程Processor。处理器的实际工作,尝试10次nbTries选择操作,在每次选择操作过程中,首先进行超时选择操作,然后检查Io处理器是否断开连接,尝试次数nbTries是否为零如果为0,则注册新的选择器;然后遍历创建会话队列,从队列拉取会话,如果会话为不null,则初始化会话,构建会话过滤链(从IoService继承)触发会话过滤链的会话创建和会话打开事件,并记录新创建的会话数量nSessions;更会会话状态,此过程为从会话次序控制队列获取会话,检查会话状态,如果状态为OPENED更新会话的读写状态,如果为OPENING放回次序控制会话队列;如果选择操作返回的SELECTKey的值大于0,即有相关的兴趣操作事件(读写事件),遍历选择后读写等操作就绪的会话,如果会话可读,则读取会话缓存区数据到buffer,触发过滤链消息接收事件MessageReceive,接收完消息后,如果会话输入流关闭则触发过滤链fireInputClosed事件,如果在这过程有异常发生,则触发过滤链异常事件ExceptionCaught,如果会话可写,则添加会话到刷新会话队列;遍历刷新会话队列,根据会话写请求消息类型为IoBuffer还是FileRegion,发送会话数据,发送会话数据后,如果会话还有些请求,则添加会话到队列,如果在这个过程中有异常,则添加会话到会话移除队列;遍历会话移除队列,如果会话为关闭,则尝试关闭会话,并清除会话写请求队列,如果会话数据已发送完,则触发会话过滤链消息发送事件fireMessageSent;更新处理器会话计数器nSessions;遍历处理器所有会话,触发会话过滤器会话空闲时间fireSessionIdle;如果在这个过程中,处理器会话计数器nSessions为0,则清除处理器引用;如果Io处理器正在关闭,则添加所有会话到移除会话队列,释放Io处理器先关的资源。
     抽象Io处理器AbstractPollingIoProcessor主要是处理IoProcessor关联会话message*事件,而所有的工作,都是通过处理器线程Processor完成。每当有会话添加到IoProcessor,则启动一个处理器线程Processor,处理会话的读写操作及相关事件。就连IoProcessor资源的释放,也是由处理器线程Processor处理。关闭IoProcessor时,现将处理器关联会话,添加移除会话队列,实际工作由IoProcessor的子类的doDispose方法完成。
今天来看Io处理器的一个具体实现NioProcessor:
/**
 * A processor for incoming and outgoing data get and written on a TCP socket.
 *
 * @author [url=http://mina.apache.org]Apache MINA Project[/url]
 */
public final class NioProcessor extends AbstractPollingIoProcessor<NioSession> {
    /** The selector associated with this processor */
    private Selector selector;//选择器
    /** A lock used to protect concurent access to the selector */
    private ReadWriteLock selectorLock = new ReentrantReadWriteLock();
    private SelectorProvider selectorProvider = null;//选择器提供者
 }

来看构造方法
  
 /**
     *
     * Creates a new instance of NioProcessor.
     *
     * @param executor The executor to use
     */
    public NioProcessor(Executor executor) {
        super(executor);

        try {
            // Open a new selector
            selector = Selector.open();
        } catch (IOException e) {
            throw new RuntimeIoException("Failed to open a selector.", e);
        }
    }
    /**
     *
     * Creates a new instance of NioProcessor.
     *
     * @param executor The executor to use
     * @param selectorProvider The Selector provider to use
     */
    public NioProcessor(Executor executor, SelectorProvider selectorProvider) {
        super(executor);

        try {
            // Open a new selector
            if (selectorProvider == null) {
                selector = Selector.open();
            } else {
                this.selectorProvider = selectorProvider;
                selector = selectorProvider.openSelector();
            }
        } catch (IOException e) {
            throw new RuntimeIoException("Failed to open a selector.", e);
        }
}

从构造函数可以看出,NioProcessor主要是初始化线程执行器和选择器。
再来选择操作:
@Override
    protected int select(long timeout) throws Exception {
        selectorLock.readLock().lock();
        
        try {
            return selector.select(timeout);
        } finally {
            selectorLock.readLock().unlock();
        }
    }

    @Override
    protected int select() throws Exception {
        selectorLock.readLock().lock();
        
        try {
            return selector.select();
        } finally {
            selectorLock.readLock().unlock();
        }
    }

从上来看Nio处理器的选择操作,实际通过内部的选择器完成。
 
  @Override
    protected boolean isSelectorEmpty() {
        selectorLock.readLock().lock();
        
        try {
            return selector.keys().isEmpty();
        } finally {
            selectorLock.readLock().unlock();
        }
    }
    @Override
    protected void wakeup() {
        wakeupCalled.getAndSet(true);
        selectorLock.readLock().lock();
        
        try {
            selector.wakeup();
        } finally {
            selectorLock.readLock().unlock();
        }
    }
    @Override
    protected Iterator<NioSession> allSessions() {
        selectorLock.readLock().lock();
        
        try {
            return new IoSessionIterator(selector.keys());
        } finally {
            selectorLock.readLock().unlock();
        }
    }
    @SuppressWarnings("synthetic-access")
    @Override
    protected Iterator<NioSession> selectedSessions() {
        return new IoSessionIterator(selector.selectedKeys());
    }
 
   //初始化会话,主要是配置会话通道为非阻塞模式,注册会话通道读事件到选择器
    @Override
    protected void init(NioSession session) throws Exception {
        SelectableChannel ch = (SelectableChannel) session.getChannel();
        ch.configureBlocking(false);
        selectorLock.readLock().lock();
        
        try {
            session.setSelectionKey(ch.register(selector, SelectionKey.OP_READ, session));
        } finally {
            selectorLock.readLock().unlock();
        }
    }
    
    @Override
    //关闭会话关联的字节通道及选择key
    protected void destroy(NioSession session) throws Exception {
        ByteChannel ch = session.getChannel();
        SelectionKey key = session.getSelectionKey();
        if (key != null) {
            key.cancel();
        }
        if ( ch.isOpen() ) {
            ch.close();
        }
    }

再来看注册新选择器
  
 /**
     * In the case we are using the java select() method, this method is used to
     * trash the buggy selector and create a new one, registering all the
     * sockets on it.
     */
    @Override
    protected void registerNewSelector() throws IOException {
        selectorLock.writeLock().lock();
        
        try {
	    //获取选择器选择key集合
            Set<SelectionKey> keys = selector.keys();
            Selector newSelector;
	    //创建一个新的选择器
            // Open a new selector
            if (selectorProvider == null) {
                newSelector = Selector.open();
            } else {
                newSelector = selectorProvider.openSelector();
            }
            //注册旧选择器的选择key关联的会话,通道,及通道兴趣事件集到新的选择器。
            // Loop on all the registered keys, and register them on the new selector
            for (SelectionKey key : keys) {
                SelectableChannel ch = key.channel();

                // Don't forget to attache the session, and back !
                NioSession session = (NioSession) key.attachment();
                SelectionKey newKey = ch.register(newSelector, key.interestOps(), session);
                session.setSelectionKey(newKey);
            }

            // Now we can close the old selector and switch it
            selector.close();
            selector = newSelector;
        } finally {
            selectorLock.writeLock().unlock();
        }

    }

从上可以看着注册新选择器,主要是注册旧选择器的选择key(集合)关联的会话,通道,及通道兴趣事件集到新的选择器;会话时附加在通道选择key的Attachment上。
再看其他操作
 /**
 * {@inheritDoc}
 */
@Override
//判断处理器是否关闭,主要是看注册到选择器的选择key关联的通道是否有断开连接,
//有一个断开连接,则处理器断开连接
protected boolean isBrokenConnection() throws IOException {
    // A flag set to true if we find a broken session
    boolean brokenSession = false;
    selectorLock.readLock().lock();
    try {
        // Get the selector keys
        Set<SelectionKey> keys = selector.keys();
        // Loop on all the keys to see if one of them
        // has a closed channel
        for (SelectionKey key : keys) {
            SelectableChannel channel = key.channel();
            if (((channel instanceof DatagramChannel) && !((DatagramChannel) channel).isConnected())
                    || ((channel instanceof SocketChannel) && !((SocketChannel) channel).isConnected())) {
                // The channel is not connected anymore. Cancel
                // the associated key then.
                key.cancel();

                // Set the flag to true to avoid a selector switch
                brokenSession = true;
            }
        }
    } finally {
        selectorLock.readLock().unlock();
    }
    return brokenSession;
}

/**
 * {@inheritDoc}
 如果会话关联的选择key有效,即会话状态为打开,为null则正在打开,否则会话关闭。
 */
@Override
protected SessionState getState(NioSession session) {
    SelectionKey key = session.getSelectionKey();
    if (key == null) {
        // The channel is not yet registred to a selector
        return SessionState.OPENING;
    }
    if (key.isValid()) {
        // The session is opened
        return SessionState.OPENED;
    } else {
        // The session still as to be closed
        return SessionState.CLOSING;
    }
}
//会话是否可读
@Override
protected boolean isReadable(NioSession session) {
    SelectionKey key = session.getSelectionKey();
    return (key != null) && key.isValid() && key.isReadable();
}
//会话是否可写
@Override
protected boolean isWritable(NioSession session) {
    SelectionKey key = session.getSelectionKey();
    return (key != null) && key.isValid() && key.isWritable();
}
//会话是否可读
@Override
protected boolean isInterestedInRead(NioSession session) {
    SelectionKey key = session.getSelectionKey();
    return (key != null) && key.isValid() && ((key.interestOps() & SelectionKey.OP_READ) != 0);
}
//是否关注写事件
@Override
protected boolean isInterestedInWrite(NioSession session) {
    SelectionKey key = session.getSelectionKey();

    return (key != null) && key.isValid() && ((key.interestOps() & SelectionKey.OP_WRITE) != 0);
}
/**
 * {@inheritDoc}
 设置读事件为会话兴趣事件
 */
@Override
protected void setInterestedInRead(NioSession session, boolean isInterested) throws Exception {
    SelectionKey key = session.getSelectionKey();

    if ((key == null) || !key.isValid()) {
        return;
    }
    int oldInterestOps = key.interestOps();
    int newInterestOps = oldInterestOps;
    if (isInterested) {
        newInterestOps |= SelectionKey.OP_READ;
    } else {
        newInterestOps &= ~SelectionKey.OP_READ;
    }
    if (oldInterestOps != newInterestOps) {
        key.interestOps(newInterestOps);
    }
}

/**
 * {@inheritDoc}
 设置写事件为会话兴趣事件
 */
@Override
protected void setInterestedInWrite(NioSession session, boolean isInterested) throws Exception {
    SelectionKey key = session.getSelectionKey();

    if ((key == null) || !key.isValid()) {
        return;
    }
    int newInterestOps = key.interestOps();
    if (isInterested) {
        newInterestOps |= SelectionKey.OP_WRITE;
    } else {
        newInterestOps &= ~SelectionKey.OP_WRITE;
    }
    key.interestOps(newInterestOps);
}

再来看读写操作
@Override
protected int read(NioSession session, IoBuffer buf) throws Exception {
    ByteChannel channel = session.getChannel();
    //委托给会话关联通道
    return channel.read(buf.buf());
}

@Override
//委托给会话关联通道
protected int write(NioSession session, IoBuffer buf, int length) throws IOException {
    if (buf.remaining() <= length) {
        return session.getChannel().write(buf.buf());
    }
    int oldLimit = buf.limit();
    buf.limit(buf.position() + length);
    try {
        return session.getChannel().write(buf.buf());
    } finally {
        buf.limit(oldLimit);
    }
}
@Override
protected int transferFile(NioSession session, FileRegion region, int length) throws Exception {
    try {
        return (int) region.getFileChannel().transferTo(region.getPosition(), length, session.getChannel());
    } catch (IOException e) {
        // Check to see if the IOException is being thrown due to
        // http://bugs.sun.com/bugdatabase/view_bug.do?bug_id=5103988
        String message = e.getMessage();
        if ((message != null) && message.contains("temporarily unavailable")) {
            return 0;
        }
        throw e;
    }
}

从上面来看,处理器处理会话读写操作,主要是通过会话关联的通道完成。
@Override
protected void doDispose() throws Exception {
    selectorLock.readLock().lock();
    try {
        selector.close();//关闭选择器
    } finally {
        selectorLock.readLock().unlock();
    }
}

下面我们贴出NioSession的代码,以便理解Nio处理器,
public abstract class NioSession extends AbstractIoSession
{
    protected final IoProcessor processor;//Io处理器
    protected final Channel channel;//选择通道
    private SelectionKey key;//选择key
    private final IoFilterChain filterChain = new DefaultIoFilterChain(this);//过滤链
    protected NioSession(IoProcessor processor, IoService service, Channel channel)
    {
        super(service);
        this.channel = channel;
        this.processor = processor;
    }
    abstract ByteChannel getChannel();
    public IoFilterChain getFilterChain()
    {
        return filterChain;
    }
    SelectionKey getSelectionKey()
    {
        return key;
    }
    void setSelectionKey(SelectionKey key)
    {
        this.key = key;
    }
    public IoProcessor getProcessor()
    {
        return processor;
    }
    public final boolean isActive()
    {
        return key.isValid();
    }
}

从NioSession的定义可以看出,Nio会话关联一个Io处理器IoProcessor,选择通道Channel,选择key(SelectionKey)和一个过滤链IoFilterChain。其实个人感觉NioProcessor和NioSession我们可以理解为Java Nio中选择器Selector与选择通道Channel。
总结:
     NioProcessor内部有一个选择器Selector,一个可重入读写锁用于控制选择器相关的操作,构造主要是初始化线程执行器和选择器。Nio处理器的选择操作,唤醒等操作,实际通过内部的选择器完成。初始化会话,主要是配置会话通道为非阻塞模式,注册会话通道读事件到选择器。注册新选择器,主要是注册旧选择器的选择key(集合)关联的会话,通道,及通道兴趣事件集到新的选择器;会话时附加在通道选择key的Attachment上。处理器处理会话读写操作,主要是通过会话关联的通道完成。关闭会话主要是关闭会话关联的字节通道和取消会话关联选择key。
附:
//IoSessionIterator
 /**
     * An encapsulating iterator around the {@link Selector#selectedKeys()} or
     * the {@link Selector#keys()} iterator;
     */
    protected static class IoSessionIterator<NioSession> implements Iterator<NioSession> {
        private final Iterator<SelectionKey> iterator;

        /**
         * Create this iterator as a wrapper on top of the selectionKey Set.
         *
         * @param keys
         *            The set of selected sessions
         */
        private IoSessionIterator(Set<SelectionKey> keys) {
            iterator = keys.iterator();
        }

        /**
         * {@inheritDoc}
         */
        @Override
        public boolean hasNext() {
            return iterator.hasNext();
        }
        /**
         * {@inheritDoc}
         */
        @Override
        public NioSession next() {
            SelectionKey key = iterator.next();
            
            return (NioSession) key.attachment();
        }
        /**
         * {@inheritDoc}
         */
        @Override
        public void remove() {
            iterator.remove();
        }
    }
}
相关标签: mina