Mina2.0框架源码剖析(六)

简介:
上文的内容还有一些没有结尾,这篇补上。在ExpiringMap类中,使用了一个私有内部类ExpiringObject来表示待检查超时的对象,它包括三个域,键,值,上次访问时间,以及用于上次访问时间这个域的读写锁:

        private K key;
        private V value;
        private long lastAccessTime;
        private final ReadWriteLock lastAccessTimeLock = new ReentrantReadWriteLock();
而ExpiringMap中包括了下述几个变量:
  private final ConcurrentHashMap<K, ExpiringObject> delegate;//超时代理集合,保存待检查对象
    private final CopyOnWriteArrayList<ExpirationListener<V>> expirationListeners;//超时监听者
    private final Expirer expirer;//超时检查线程
现在再来看看IoSession的一个抽象实现类AbstractIoSession。这是它的几个重要的成员变量:

    private IoSessionAttributeMap attributes;//会话属性映射图
    private WriteRequestQueue writeRequestQueue;//写请求队列
    private WriteRequest currentWriteRequest;//当前写请求
     当要结束当前会话时,会发送一个一个写请求CLOSE_REQUEST。而closeFuture这个CloseFuture会在连接关闭时状态被设置为”closed”,它的监听器是SCHEDULED_COUNTER_RESETTER。

close和closeOnFlush都是异步的关闭操作,区别是前者立即关闭连接,而后者是在写请求队列中放入一个CLOSE_REQUEST,并将其即时刷新出去,若要真正等待关闭完成,需要调用方在返回的CloseFuture等待

复制代码
public final CloseFuture close() {
        synchronized (lock) {
            if (isClosing()) {
                return closeFuture;
            } else {
                closing = true;
            }
        }
        getFilterChain().fireFilterClose();//fire出关闭事件
        return closeFuture;
    }

    public final CloseFuture closeOnFlush() {
        getWriteRequestQueue().offer(this, CLOSE_REQUEST);
        getProcessor().flush(this);
        return closeFuture;
    }
复制代码
     下面来看看读数据的过程:

复制代码
public final CloseFuture close() {
        synchronized (lock) {
            if (isClosing()) {
                return closeFuture;
            } else {
                closing = true;
            }
        }
        getFilterChain().fireFilterClose();//fire出关闭事件
        return closeFuture;
    }

    public final CloseFuture closeOnFlush() {
        getWriteRequestQueue().offer(this, CLOSE_REQUEST);
        getProcessor().flush(this);
        return closeFuture;
    }


    private Queue<ReadFuture> getReadyReadFutures() {//返回可被读数据队列
        Queue<ReadFuture> readyReadFutures =
            (Queue<ReadFuture>) getAttribute(READY_READ_FUTURES_KEY);//从会话映射表中取出可被读数据队列
        if (readyReadFutures == null) {//第一次读数据
            readyReadFutures = new CircularQueue<ReadFuture>();//构造一个新读数据队列
            Queue<ReadFuture> oldReadyReadFutures =
                (Queue<ReadFuture>) setAttributeIfAbsent(
                        READY_READ_FUTURES_KEY, readyReadFutures);
            if (oldReadyReadFutures != null) {
                readyReadFutures = oldReadyReadFutures;
            }
        }
        return readyReadFutures;
    }

    public final ReadFuture read() {//读数据
        if (!getConfig().isUseReadOperation()) {//会话配置不允许读数据(这是默认情况)
            throw new IllegalStateException("useReadOperation is not enabled.");
        }
        Queue<ReadFuture> readyReadFutures = getReadyReadFutures();//获取已经可被读数据队列
        ReadFuture future;
        synchronized (readyReadFutures) {//锁住读数据队列
            future = readyReadFutures.poll();//取队头数据
            if (future != null) {
                if (future.isClosed()) {//关联的会话已经关闭了,让读者知道此情况
                    readyReadFutures.offer(future);
                }
            } else {
                future = new DefaultReadFuture(this);
                getWaitingReadFutures().offer(future); //将此数据插入等待被读取数据的队列,这个代码和上面的getReadyReadFutures类似,只是键值不同而已

            }
        }
        return future;
    }
复制代码
     再来看写数据到指定远端地址的过程,可以写三种类型数据:IoBuffer,整个文件或文件的部分区域,这会通过传递写请求给过滤器链条来完成数据向目的远端的传输。

复制代码
    public final WriteFuture write(Object message, SocketAddress remoteAddress) {
        FileChannel openedFileChannel = null;
        try 
        {
            if (message instanceof IoBuffer&& !((IoBuffer) message).hasRemaining()) 
            {// 空消息
                throw new IllegalArgumentException(
                "message is empty. Forgot to call flip()?");
            } 
            else if (message instanceof FileChannel) 
            {//要发送的是文件的某一区域
                FileChannel fileChannel = (FileChannel) message;
                message = new DefaultFileRegion(fileChannel, 0, fileChannel.size());
            }
            else if (message instanceof File) 
            {//要发送的是文件,打开文件通道
                File file = (File) message;
                openedFileChannel = new FileInputStream(file).getChannel();
                message = new DefaultFileRegion(openedFileChannel, 0, openedFileChannel.size());
            }
        } 
        catch (IOException e) 
        {
            ExceptionMonitor.getInstance().exceptionCaught(e);
            return DefaultWriteFuture.newNotWrittenFuture(this, e);
        }
        WriteFuture future = new DefaultWriteFuture(this); 
        getFilterChain().fireFilterWrite(
                new DefaultWriteRequest(message, future, remoteAddress)); //构造写请求,通过过滤器链发送出去,写请求中指明了要发送的消息,目的地址,以及返回的结果
 
//如果打开了一个文件通道(发送的文件的部分区域或全部),就必须在写请求完成时关闭文件通道
        if (openedFileChannel != null) {
            final FileChannel finalChannel = openedFileChannel;
            future.addListener(new IoFutureListener<WriteFuture>() {
                public void operationComplete(WriteFuture future) {
                    try {
                        finalChannel.close();//关闭文件通道
                    } catch (IOException e) {
                        ExceptionMonitor.getInstance().exceptionCaught(e);
                    }
                }
            });
        }
        return future;//写请求成功完成
    }
复制代码
     最后,来看看一个WriteRequestQueue的实现,唯一加入的一个功能就是若在队头发现是请求关闭,则会去关闭会话。

复制代码
 private class CloseRequestAwareWriteRequestQueue implements WriteRequestQueue {
        private final WriteRequestQueue q;//内部实际的写请求队列
        public CloseRequestAwareWriteRequestQueue(WriteRequestQueue q) {
            this.q = q;
        }
        public synchronized WriteRequest poll(IoSession session) {
            WriteRequest answer = q.poll(session);
            if (answer == CLOSE_REQUEST) {
                AbstractIoSession.this.close();
                dispose(session);
                answer = null;
            }
            return answer;
        }
        public void offer(IoSession session, WriteRequest e) {
            q.offer(session, e);
        }
        public boolean isEmpty(IoSession session) {
            return q.isEmpty(session);
        }
        public void clear(IoSession session) {
            q.clear(session);
        }
        public void dispose(IoSession session) {
            q.dispose(session);
        }
    }
复制代码



本文转自Phinecos(洞庭散人)博客园博客,原文链接:http://www.cnblogs.com/phinecos/archive/2008/12/07/1349723.html,如需转载请自行联系原作者
目录
相关文章
|
7月前
|
缓存 网络协议 算法
《跟闪电侠学Netty》阅读笔记 - Netty入门程序解析
《跟闪电侠学Netty》阅读笔记 - Netty入门程序解析
223 0
|
Java
由浅入深Netty组件实战3
由浅入深Netty组件实战3
68 0
|
前端开发 算法 Java
由浅入深Netty组件实战2
由浅入深Netty组件实战2
229 0
|
前端开发 安全 Java
由浅入深Netty组件实战1
由浅入深Netty组件实战1
91 0
|
7月前
|
设计模式 网络协议 算法
《跟闪电侠学Netty》阅读笔记 - Netty入门程序解析(一)
《跟闪电侠学Netty》阅读笔记 - Netty入门程序解析(一)
246 1
《跟闪电侠学Netty》阅读笔记 - Netty入门程序解析(一)
|
7月前
|
缓存 网络协议 算法
《跟闪电侠学Netty》阅读笔记 - Netty入门程序解析(二)
《跟闪电侠学Netty》阅读笔记 - Netty入门程序解析
139 1
|
设计模式 监控 前端开发
第 10 章 Netty 核心源码剖析
第 10 章 Netty 核心源码剖析
137 0
|
Java 安全 API
Netty 源码阅读入门实战(一)-介绍
1 简介 以 Netty 为底层的框架 Netty 是什么 Netty 技术和方法的特点 设计 针对多种传输类型的统一接口 - 阻塞和非阻塞 简单但更强大的线程模型 真正的无连接的数据报套接字支持 链接逻辑支持复用 易用性 大量的 Javadoc 和 代码实例 除了在 JDK 1.6 + 额外的限制。
1752 0
|
Java
Netty 源码阅读入门实战(四)-NioEventLoop
1 NioEventLoop概述 总述 2 NioEventLoop创建概述 ...
1184 0
|
Web App开发 Java
探秘Netty6:基于Netty自己动手实现Web框架
大厨小鲜——基于Netty自己动手实现Web框架 钱文品 Good news everyone! ​关注他 17 人赞了该文章 上节课我们自己动手制作了一个RPC框架,本节课我们挑战一个稍有难度的一点的任务,手动制作一个Web框架。