Mina框架IoSession详解

2022-02-17 21:44:48

通过Mina官 网文档，我们可以看到，有如下几个状态：

  • 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)Closing : the session is being closed (the remaining messages are being flushed, cleaning up is not terminated)
    • 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
  • Closed : The session is now closed, nothing else can be done to revive it.

对应的状态迁移图，如图所示：

Mina框架IoSession详解

通过上面的状态图，我们可以看出，是哪个事件的发生使得IoSession进入哪个状态，比较直观明了。下面，我们看一下IoSession对应的设计，类继承关系如下所示：

Mina框架IoSession详解

对于IoSession接口类，我在上图把具有不同类型功能的操作进行了分类，说明如下：

  • 一个IoSession实例可以访问/持有哪些数据：前半部分以get开头的方法，能够返回对应的数据对象。
  • 一个IoSession实例可以检测哪些状态数据：中间部分以is开头的一些方法。
  • 一个IoSession实例可以执行哪些方法调用：后半部分以动词开头命名的方法。
  • 一个IoSession实例还可以获取通信过程相关的统计信息，如读取字节数/消息数、写入字节数/消息数，等等，在上面类图中省略 了这些方法。

可见，IoSession在Mina框架中的位置是相当重要的。

根据上面的类图，我们分析一下NioSocketSession类的源代码。
AbstractIoSession实现了IoSession接口中定义的大多数方法，我们关注读和写两个重要的方法，因为他们最终也被NioSocketSession类所继承。
先看读数据请求方法read，如下所示：

01 public final ReadFuture read() {
02 if (!getConfig().isUseReadOperation()) {
03 throw new IllegalStateException("useReadOperation is not enabled.");
04 }
05
06 Queue<ReadFuture> readyReadFutures = getReadyReadFutures(); // 获取到read请求就绪队列
07 ReadFuture future;
08 synchronized (readyReadFutures) { // 这个对就绪队列是共享的，对于读请求之间需要进行同步
09 future = readyReadFutures.poll(); // 出队
10 if (future != null) { // 如果队列中有就绪的read请求
11 if (future.isClosed()) { // 如果与该IoSession相关的ReadFuture已经关闭（读取完成）
12 readyReadFutures.offer(future); // 还要将这个ReadFuture放入到队列，等待该IoSession下次可能的读请求
13 }
14 } else {
15 future = new DefaultReadFuture(this); // 如果是与该IoSession相关的第一次读请求，目前读就绪队列肯定没有一个ReadFuture实例，则需要创建一个
16 getWaitingReadFutures().offer(future); // 将新创建的ReadFuture实例放入到等待读队列
17 }
18 }
19
20 return future; // 返回一个ReadFuture实例，无论是第一次发出读请求，还是上一次读请求已经完成，对于本次读请求都将返回这个ReadFuture实例
21 }

再看一下，写数据请求方法write，如下所示：

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 if (!getTransportMetadata().isConnectionless() && (remoteAddress != null)) {
07 throw new UnsupportedOperationException();
08 }
09
10 if (isClosing() || !isConnected()) { // 如果该次会话正在关闭，或者就没有连接过，则封装一个异常返回一个WriteFuture对象
11 WriteFuture future = new DefaultWriteFuture(this);
12 WriteRequest request = new DefaultWriteRequest(message, future, remoteAddress);
13 WriteException writeException = new WriteToClosedSessionException(request);
14 future.setException(writeException);
15 return future;
16 }
17
18 FileChannel openedFileChannel = null;
19
20 try {
21 if ((message instanceof IoBuffer) && !((IoBuffer) message).hasRemaining()) {// 没有写任何数据
22 throw new IllegalArgumentException("message is empty. Forgot to call flip()?");
23 } else if (message instanceof FileChannel) {
24 FileChannel fileChannel = (FileChannel) message;
25 message = new DefaultFileRegion(fileChannel, 0, fileChannel.size()); // 如果是FileChannel，则创建一个DefaultFileRegion对象，用来被Mina操纵
26 } else if (message instanceof File) {
27 File file = (File) message;
28 openedFileChannel = new FileInputStream(file).getChannel();
29 message = new FilenameFileRegion(file, openedFileChannel, 0, openedFileChannel.size()); // 如果是File，则创建FilenameFileRegion
30 }
31 } catch (IOException e) {
32 ExceptionMonitor.getInstance().exceptionCaught(e);
33 return DefaultWriteFuture.newNotWrittenFuture(this, e);
34 }
35
36 // 可以写message了，做写前准备
37 WriteFuture writeFuture = new DefaultWriteFuture(this);
38 WriteRequest writeRequest = new DefaultWriteRequest(message, writeFuture, remoteAddress);
39
40 // Then, get the chain and inject the WriteRequest into it
41 IoFilterChain filterChain = getFilterChain(); // 获取到与该IoSession相关的IoFilterChain，方法getFilterChain实现可以看NioSocketSession类中的实现：filterChain = new DefaultIoFilterChain(this);
42 filterChain.fireFilterWrite(writeRequest); // 触发写事件，将WriteRequest注入到IoFilter实例链，执行注册的IoFilter的逻辑
43
44 // 不关心FileChannel的操作，不进行处理，直接关闭掉
45 if (openedFileChannel != null) {
46 final FileChannel finalChannel = openedFileChannel;
47 writeFuture.addListener(new IoFutureListener<WriteFuture>() {
48 public void operationComplete(WriteFuture future) {
49 try {
50 finalChannel.close();
51 } catch (IOException e) {
52 ExceptionMonitor.getInstance().exceptionCaught(e);
53 }
54 }
55 });
56 }
57
58 return writeFuture; // 返回WriteFuture，等待写操作异步完成
59 }

再看，NioSession类中增加了一个返回IoProcessor实例的抽象方法，而这个IoProcessor是在创建一个IoSession实例（例如，可以实例化一个NioSocketSession）的时候，由外部传到IoSession内部。我们知道，IoProcessor是Mina框架底层真正用来处理实际I/O操作的处理器，通过一个IoSession实例获取一个IoProcessor，可以方便地响应作用于IoSession的I/O读写请求，从而由这个IoProcessor直接去处理。
根据Mina框架架构设计，IoService->IoFilter Chain->IoHandler，我们知道在IoFilter Chain的一端（头部）之前会调用处理实际的I/O操作请求，也就是IoProcessor需要处理的逻辑，那么可以想象到，IoProcessor被调用的位置，可以查看org.apache.mina.core.filterchain.DefaultIoFilterChain类的源代码，其中定义了一个内部类，源码如下所示：

01 private class HeadFilter extends IoFilterAdapter {
02 @SuppressWarnings("unchecked")
03 @Override
04 public void filterWrite(NextFilter nextFilter, IoSession session, WriteRequest writeRequest) throws Exception {
05
06 AbstractIoSession s = (AbstractIoSession) session;
07
08 // Maintain counters.
09 if (writeRequest.getMessage() instanceof IoBuffer) {
10 IoBuffer buffer = (IoBuffer) writeRequest.getMessage();
11 // I/O processor implementation will call buffer.reset()
12 // it after the write operation is finished, because
13 // the buffer will be specified with messageSent event.
14 buffer.mark();
15 int remaining = buffer.remaining();
16
17 if (remaining == 0) {
18 // Zero-sized buffer means the internal message
19 // delimiter.
20 s.increaseScheduledWriteMessages();
21 } else {
22 s.increaseScheduledWriteBytes(remaining);
23 }
24 } else {
25 s.increaseScheduledWriteMessages();
26 }
27
28 WriteRequestQueue writeRequestQueue = s.getWriteRequestQueue();
29
30 if (!s.isWriteSuspended()) {
31 if (writeRequestQueue.size() == 0) {
32 // We can write directly the message
33 s.getProcessor().write(s, writeRequest);
34 } else {
35 s.getWriteRequestQueue().offer(s, writeRequest);
36 s.getProcessor().flush(s);
37 }
38 } else {
39 s.getWriteRequestQueue().offer(s, writeRequest);
40 }
41 }
42
43 @SuppressWarnings("unchecked")
44 @Override
45 public void filterClose(NextFilter nextFilter, IoSession session) throws Exception {
46 ((AbstractIoSession) session).getProcessor().remove(session);
47 }
48 }

最后，我们看一下NioSocketSession实例被创建的时机。其实很容易想到，当一次网络通信开始的时候，也就是客户端连接到服务器端的时候，服务器端首先进行accept，这时候一次会话才被启动，也就是在这个是被创建，拿Mina中的NioSocketAcceptor类来看，创建NioSocketSession的代码，如下所示：

01 protected NioSession accept(IoProcessor<NioSession> processor, ServerSocketChannel handle) throws Exception {
02
03 SelectionKey key = handle.keyFor(selector);
04
05 if ((key == null) || (!key.isValid()) || (!key.isAcceptable())) {
06 return null;
07 }
08
09 // accept the connection from the client
10 SocketChannel ch = handle.accept();
11
12 if (ch == null) {
13 return null;
14 }
15
16 return new NioSocketSession(this, processor, ch); // 创建NioSocketSession实例
17 }

通过上面的分析，我们可知，IoSession在基于Mina进行网络通信的过程中，对于网络通信相关操作的请求都是基于一个IoSession实例来进行的，所以说，IoSession在Mina中是一个很重要的结构。

上一篇：VMware拓展vSphere Integrated Containers功能，助力企业拥抱数字商务


下一篇：ELK日志服务使用-kafka传输日志(bbotte.com)