zoukankan      html  css  js  c++  java

  • Mina源码阅读笔记（五）—Mina对连接的操作IoSession

    IoSession是Mina管理两端的一个重要部分，也是Mina的核心，Session具有了生命周期的概念，它的生命周期和连接时紧密相关的，这点在后面的介绍中会涉及。另外，好像hibernate中也有session也有生命周期（真的是好久没有用了，连hibernate有里session是干嘛的都想不起来了）。

    在读源码之前，我们还是先了解下IoSession的作用和一些基本概念。IoSession的主要作用有如下一些：

    l  管理连接。注意，这里的管理连接并不是直接去控制我们上次讲的最底层的连接acceptor和connector。如果acceptor和connector建立的一条管道，那session就是在管道内的管理者，他是没有办法将管道对半拆分开的，他只能从内部阻断两边的通信。管理连接还有部分就是可以配置缓冲区的大小，闲置时间等等。

    l  存储信息。和web里的session一样，这里的session也有存储attribute的功能，不过一般来说，这里存储的都是和连接有关的东西，并不会像web开发一样存一些业务上的东西。

    l  驱动读写操作。我不知道用驱动这个词是否合适，那个例子来说，session.write。

    l  统计功能。Session还记录了连接中的byte、message等数量。

    Session在使用中是通过ConnectionFuture获得的：

    1 ConnectFuture future = connector.connect(new InetSocketAddress(
    2                     HOST, PORT));// 创建连接
    3             future.awaitUninterruptibly();// 等待连接创建完成
    4             session = future.getSession();// 获得session

    说完作用，就是session的状态了，我们读源码的时候，也会跟着session的状态来读：

    • Connected : the session has been created and is available
    • Idle : the session hasn't processed any request for at least a period of time (this period is configurable)
      • Idle for read : no read has actually been made for a period of time
      • Idle for write : no write has actually been made for a period of time
      • Idle for both : no read nor write for a period of time
    • Closing : the session is being closed (the remaining messages are being flushed, cleaning up is not terminated)
    • Closed : The session is now closed, nothing else can be done to revive it.

    注意，这里的最开始的connect，最后的closed这些都说的是acceptor和connector之间的操作，是他们的状态来影响session的状态。和hibernate不一样，那个的操作也是session自己的（session.close等），这里的session是没办法控制通道的。

    了解完了基础的，我们来看看代码层面的实现，在org,apache.mina.core.session包中主要实现了IoSession的相关功能。我们从IoSession这个接口开始看起：

    内容比较多，比之前的那几篇都难多了。我们还是按照上图画出来的方框对这些方法进行分分类，看看是不是跟我们之前分析的一样：

    红色：得到一系列配置等。我们说了session是mina的核心。

    蓝色：驱动读写操作。

    绿色：管理连接。

    黑色：存储功能，是不是和JSP里的session很像。

    橘色：统计数据。

    有些没有框上的并不是说不属于这里面，而是有些我确实不了解，有些是比较难划分，具体的含义可以看源码中，都有说明。

    了解完IoSession的最基本功能之后，我们要看看它的具体实现类AbstractIoSession。看具体实现之前，我们先看看这个类关联了哪些比较重要的引用：

    01 private final IoHandler handler;
    02  
    03     protected IoSessionConfig config;
    04  
    05       private final IoService service;
    06  
    07     private static final AttributeKey READY_READ_FUTURES_KEY = new AttributeKey(AbstractIoSession.class,
    08             "readyReadFutures");
    09 private static final IoFutureListener<CloseFuture> SCHEDULED_COUNTER_RESETTER;
    10  
    11   private static final WriteRequest CLOSE_REQUEST = new DefaultWriteRequest(new Object());
    12  
    13     private IoSessionAttributeMap attributes;
    14  
    15     private WriteRequestQueue writeRequestQueue;
    16  
    17 private WriteRequest currentWriteRequest;
    18  
    19 private final CloseFuture closeFuture = new DefaultCloseFuture(this);

    在上面的代码中，我们有熟悉的handler，这里我们也可以稍微明确一下handler和session之间的关系了，每次我们在创建服务端和客户端的时候，都必须设置一个handler，如果不设置则报异常，handler里主要处理seesion各种状态时的业务。Service用来管理session。CloseFutrue用来设置通道的关闭，在上面我们已经说过，session的关闭是通过closefuture来操作的。

    这些成员变量中，除了我们见过的handler和future，凡是带有write的都来自org.apache.mina.core.write包，这个包作为一个内部的工具类，在session的写操作中起到辅助作用。其他类均来自org.apache.mina.core.session中，这些类组成都比较简单，但都是要了解AbstractIoSession之前，我们要对这里提到的这些对象有所了解。

    首先是ioSessionConfig，和它的具体实现AbstractIoSessionCnfig:

    从上面的图我们很容易就能看到mina的一些默认传输配置，当然这些数字都不是随便写的，为什么最小要64，最大是65535，我相信计算机网络相关课程里应该都会有涉及。

    接下来是IoSessionAttributeMap接口，这个接口主要作用就是规定了get、set、remove Attribute的方法。注意存储在session中的变量是一种map关系的变量（key-value），所以我们也很容易明白这个接口命名时为什么后面要多个map出来。至于这个类的实现，它隐藏的很好，放在了一个内部类中。具体可以看IoSessionDataStructureFactory这个接口，这个接口主要是为了这个map形势提供数据结构和规定基本操作。至于这个session中map的底层实现则用了ConcurrentHashMap来做容器，这部分具体可以看内部类DefaultIoSessionAttributeMap。

    还有一个就是AttributeKey，这个类主要重写equals方法和hashCode方法，为了将session中的key和对应的session联系起来。因为一个项目中可能有多个session，而不同session中的key可能会相同，所以在构造key和hash的时候会将session也考虑进去。

    1 public AttributeKey(Class<?> source, String name) {
    2         this.name = source.getName() + '.' + name + '@' + Integer.toHexString(this.hashCode());
    3     }

    现在我们可以看AbstractIoSession了。主要看读写操作，其他操作都是统计和配置稍稍看过即可：

    01 public final ReadFuture read() {
    02         if (!getConfig().isUseReadOperation()) {
    03             throw new IllegalStateException("useReadOperation is not enabled.");
    04         }
    05  
    06         Queue<ReadFuture> readyReadFutures = getReadyReadFutures();
    07         ReadFuture future;
    08         synchronized (readyReadFutures) {
    09             future = readyReadFutures.poll();
    10             if (future != null) {
    11                 if (future.isClosed()) {
    12                     // Let other readers get notified.
    13                     readyReadFutures.offer(future);
    14                 }
    15             } else {
    16                 future = new DefaultReadFuture(this);
    17                 getWaitingReadFutures().offer(future);
    18             }
    19         }
    20  
    21         return future;
    22     }

    采用队列进行读取，这里只用了Queue，没有用concurrent中的那些同步队列，而是用了synchronized关键字来处理同步，主要是我们要明白这里要同步的不是队列里的内容，而是读的这个过程，session都是独立的，所以一个session内一个队列无论怎么抢还是能排除顺序的。所以对于读操作来说，主要是要保证在读一条的时候，不能有其他线程再读。

    下面是写操作：

    01 public WriteFuture write(Object message, SocketAddress remoteAddress) {
    02         if (message == null) {
    03             throw new IllegalArgumentException("Trying to write a null message : not allowed");
    04         }
    05  
    06         // We can't send a message to a connected session if we don't have
    07         // the remote address
    08         if (!getTransportMetadata().isConnectionless() && (remoteAddress != null)) {
    09             throw new UnsupportedOperationException();
    10         }
    11  
    12         // If the session has been closed or is closing, we can't either
    13         // send a message to the remote side. We generate a future
    14         // containing an exception.
    15         if (isClosing() || !isConnected()) {
    16             WriteFuture future = new DefaultWriteFuture(this);
    17             WriteRequest request = new DefaultWriteRequest(message, future, remoteAddress);
    18             WriteException writeException = new WriteToClosedSessionException(request);
    19             future.setException(writeException);
    20             return future;
    21         }
    22  
    23         FileChannel openedFileChannel = null;
    24  
    25         // TODO: remove this code as soon as we use InputStream
    26         // instead of Object for the message.
    27         try {
    28             if ((message instanceof IoBuffer) && !((IoBuffer) message).hasRemaining()) {
    29                 // Nothing to write : probably an error in the user code
    30                 throw new IllegalArgumentException("message is empty. Forgot to call flip()?");
    31             } else if (message instanceof FileChannel) {
    32                 FileChannel fileChannel = (FileChannel) message;
    33                 message = new DefaultFileRegion(fileChannel, 0, fileChannel.size());
    34             } else if (message instanceof File) {
    35                 File file = (File) message;
    36                 openedFileChannel = new FileInputStream(file).getChannel();
    37                 message = new FilenameFileRegion(file, openedFileChannel, 0, openedFileChannel.size());
    38             }
    39         } catch (IOException e) {
    40             ExceptionMonitor.getInstance().exceptionCaught(e);
    41             return DefaultWriteFuture.newNotWrittenFuture(this, e);
    42         }
    43  
    44         // Now, we can write the message. First, create a future
    45         WriteFuture writeFuture = new DefaultWriteFuture(this);
    46         WriteRequest writeRequest = new DefaultWriteRequest(message, writeFuture, remoteAddress);
    47  
    48         // Then, get the chain and inject the WriteRequest into it
    49         IoFilterChain filterChain = getFilterChain();
    50         filterChain.fireFilterWrite(writeRequest);
    51  
    52         // TODO : This is not our business ! The caller has created a
    53         // FileChannel,
    54         // he has to close it !
    55         if (openedFileChannel != null) {
    56             // If we opened a FileChannel, it needs to be closed when the write
    57             // has completed
    58             final FileChannel finalChannel = openedFileChannel;
    59             writeFuture.addListener(new IoFutureListener<WriteFuture>() {
    60                 public void operationComplete(WriteFuture future) {
    61                     try {
    62                         finalChannel.close();
    63                     } catch (IOException e) {
    64                         ExceptionMonitor.getInstance().exceptionCaught(e);
    65                     }
    66                 }
    67             });
    68         }
    69  
    70         // Return the WriteFuture.
    71         return writeFuture;
    72     }

    这里面有个instanceof FileChannel和File是不是感到有点儿奇怪，mina不是写出去的是IoBuffer么，怎么现在又可以写文件了。Mina作为一个封装好的框架，自然可以直接做文件的传输，这里面会有相应的handler来处理这些业务。

    Session部分最主要的就是了解他的生命周期以及相关联的那些引用，这里我们可以看到与读写最密切的就是Future了，所以这部分，就是我下篇会写的主题。
  • 相关阅读:
    面对一个全新的环境,作为一个Mysql DBA,首先应该了解什么?
    Puppet学习之puppet的安装和配置
    Puppet学习之文件管理
    AJAx在windows 2003 系统下没有响应
    如何学习jquery
    系统底层图书推荐
    Jquery+Aspx实现Ajax得不到响应
    如何深入学习asp.net
    asp.net精华文章推荐
    爱恨交加的asp.net服务器控件
  • 原文地址：https://www.cnblogs.com/wuyida/p/6300942.html
Copyright © 2011-2022 走看看