NIO框架之MINA源码解析：底层通信与责任链模式应用

Posted 2020-09-13 学无止境

本文主要介绍下在mina中责任链模式的应用以及mina对于数据读写的处理。

 

在mina中，对数据的读操作是在processor类里面触发的，收到新消息后就触发读数据链去处理新消息直到自己的业务逻辑代码（IoHandler）。

 

在mina中，数据的写(write)和发(send)差别相对较大，mina中的写消息最终的结果只是把要写的消息经过写数据链处理后的最终结果放在了一个缓存中，并把当前session标记为可发。

 

数据的发送就是传统中我们所说的发消息，就是把写消息最终处理的结果发送到客户端，待发送完成后，就调用发数据链进行操作，最后再调用IoHandlerde 发方法。

 

1、触发点

1.1、读数据

读操作是在Processor中的触发的，Processor是AbstractPollingIoProcessor的内部私有类。

Processor中有一个死循环，循环调用Selector的select方法，若有新消息，则进行process()。

 

详见Processor代码。

 

1.2、写数据

 

写操作很简单，是调用session的write方法，进行写数据的，写数据的最终结果保存在一个缓存队列里面，等待发送，并把当前session放入flushSession队列里面。

 

1.3、 发数据

 

发数据其实和读数据是差不多的，都在Processor中的触发的，在process()完新消息后，会调用flush（）方法，把flushSession队列里面的session取出来，并把缓存的消息发送到客户端。

 

详见Processor代码。

 

2、责任链模式

在mina中，责任链模式相关的类都定义在了org.apache.mina.core.filterchain包中，看类图：

 

IoFilter接口实现代表一个具体的执行单元，是写自定义IoFilter的接口，IoFilter有一个适配器IoFilterAdapter，里面什么都没有干。

NextFilter接口看上去和IoFilter接口差不多，但NextFilter接口代表的是“下一个filter”，这里的下是抽象的，因为在mina的各种链中，处理顺序有的是从头到尾，有的是从尾到头，而这里的下就代表了熟悉中的下一个filter。

Entry接口是链接中的具体对象，其封装了IoFilter接口和NextFilter接口，代表一个实体放在链队列里面。

DefaultIoFilterChainBuilder是mina提供的一个便于用户构造IoFilterChain的方法。

2.1、读数据-责任链

 

 

2.2、写数据-责任链

 

 

 

2.3、发数据-责任链

 

 

2.4 责任链引擎实现

 

这里以读数据-责任链为例，说下在mina中具体是怎么实现这个责任链的执行。

 

 

engin代码

 

 

[java] view plain copy

 

 print?

1. //class DefaultIoFilterChain  
2.   
3. //触发点，传入消息对象  
4.  public void fireMessageReceived(Object message) {  
5.         if (message instanceof IoBuffer) {  
6.             session.increaseReadBytes(((IoBuffer) message).remaining(), System.currentTimeMillis());  
7.         }  
8.     //从头开始  
9.         Entry head = this.head;  
10.         callNextMessageReceived(head, session, message);  
11.     }  
12.   
13.     //这个地方是个递归调用入口，目的就是执行entry里面的iofilter，并把当前entry的下一个nextfilter传入  
14.     private void callNextMessageReceived(Entry entry, iosession session, Object message) {  
15.         try {  
16.             IoFilter filter = entry.getFilter();  
17.             NextFilter nextFilter = entry.getNextFilter();  
18.             filter.messageReceived(nextFilter, session, message);  
19.         } catch (Throwable e) {  
20.             fireExceptionCaught(e);  
21.         }  
22.     }  

 

iofilter代码，这里面拿一个简单的LoggingFilter来说明

 

 

[java] view plain copy

 

 print?

1. //class LoggingFilter  
2.    public void messageReceived(NextFilter nextFilter, IoSession session, Object message) throws Exception {  
3.         log(messageReceivedLevel, "RECEIVED: {}", message);  
4.     //调用nextfilter的receive方法，执行下一个filter  
5.     //也可以不调用，直接返回  
6.         nextFilter.messageReceived(session, message);  
7.   
8.     //当然也可以在这里加一些处理后的业务逻辑。  
9.     }  

 

 

nextfilter代码

 

 

[java] view plain copy

 

 print?

1. private class EntryImpl implements Entry {  
2.         private EntryImpl prevEntry;  
3.   
4.         private EntryImpl nextEntry;  
5.   
6.         private final String name;  
7.   
8.         private IoFilter filter;  
9.   
10.         private final NextFilter nextFilter;  
11.   
12.         private EntryImpl(EntryImpl prevEntry, EntryImpl nextEntry, String name, IoFilter filter) {  
13.             if (filter == null) {  
14.                 throw new IllegalArgumentException("filter");  
15.             }  
16.             if (name == null) {  
17.                 throw new IllegalArgumentException("name");  
18.             }  
19.   
20.             this.prevEntry = prevEntry;  
21.             this.nextEntry = nextEntry;  
22.             this.name = name;  
23.             this.filter = filter;  
24.             this.nextFilter = new NextFilter() {  
25.                 
26.                 public void messageReceived(IoSession session, Object message) {  
27.                     Entry nextEntry = EntryImpl.this.nextEntry;  
28.             //把nextEntry指向相应的entry，继续进入循环体，注意这里的EntryImpl.this.nextEntry，即从头到尾  
29.                     callNextMessageReceived(nextEntry, session, message);  
30.                 }  
31.   
32.                 public void messageSent(IoSession session, WriteRequest writeRequest) {  
33.                     Entry nextEntry = EntryImpl.this.nextEntry;  
34.                     callNextMessageSent(nextEntry, session, writeRequest);  
35.                 }  
36.   
37.                 public void filterWrite(IoSession session, WriteRequest writeRequest) {  
38.                     Entry nextEntry = EntryImpl.this.prevEntry;  
39.             //把nextEntry指向相应的entry，继续进入循环体，注意这里的EntryImpl.this.prevEntry，即从尾到头  
40.                     callPreviousFilterWrite(nextEntry, session, writeRequest);  
41.                 }  
42.             
43.             };  
44.         }  
45.   
46.         public String getName() {  
47.             return name;  
48.         }  
49.   
50.         public IoFilter getFilter() {  
51.             return filter;  
52.         }         
53.     }