基于Mina开发网络通信程序，在传感器数据接入领域应用的很广泛，今天我无意中发现一个问题，那就是我在前端session.write(msg)数据出去之后，却没有经过Filter的Encoder方法，同样能够写入远程服务器。因为我所发送的数据不需要很复杂的编码，所以encoder方法也一直没有去看，今天发现无法被自己写的过滤器所编码，针对这个问题，我打开以前的代码以及以前的项目中的相关代码，有些同事也是session.write(IoBuffer)之后，在encoder方法里面还加上了一句out.write(message);通过跟踪Mina源码发现，session写出去的数据类型是IoBuffer格式的，就不经过自定义的过滤器了。所以下面的代码压根是多余的

  1. @Override
  2. public void encode(IoSession session, Object message, ProtocolEncoderOutput out) throws Exception {
  3. out.write(message);//IoBuffer格式写出去之后，跳过了encoder.
  4. }

下面我把自己跟踪调试Mina的过程记录下来. 
一、场景 
客户端需要每隔Time时间向服务端发送心跳包，代码如下： 
session.write(IoBuffer.wrap("心跳包XXX".getBytes())); 
二、现象 
MyFilter中的Encoder方法encoder不执行

  1. public class MyFilter implements ProtocolCodecFactory {
  2. private ProtocolEncoder encoder = new MyEncoder();
  3. private ProtocolDecoder decoder = new MyDecoder();
  4. @Override
  5. public ProtocolEncoder getEncoder(IoSession session) throws Exception {
  6. return encoder;
  7. }
  8. @Override
  9. public ProtocolDecoder getDecoder(IoSession session) throws Exception {
  10. return decoder;
  11. }
  12. }

三、分析 
进入session.write方法，实现IoSession.write方法的是AbstractIoSession。直接调用的是

  1. public WriteFuture write(Object message) {
  2. return write(message, null);
  3. }

而AbstractIoSession.write(Object message, SocketAddress address) 
该方法的工作流程是：

  • 创建WriteFeature对象，用于返回值（session.write本身就是返回writeFeature）
  • 将session.write(message)中的Object类型的message封装成writeRequest.
  • 启动write动作，这个主要是IoFilterChain来完成的。

具体的核心代码如下：

  1. // Now, we can write the message. First, create a future
  2. WriteFuture writeFuture = new DefaultWriteFuture(this);
  3. WriteRequest writeRequest = new DefaultWriteRequest(message, writeFuture, remoteAddress);
  4. // Then, get the chain and inject the WriteRequest into it
  5. IoFilterChain filterChain = getFilterChain();
  6. filterChain.fireFilterWrite(writeRequest);

继续跟踪到fireFilterWrite里面去，可知IoFilterChain的默认实现类DefaultIoFilterChain中的关键方法：

  1. public void fireFilterWrite(WriteRequest writeRequest) {
  2. Entry tail = this.tail;
  3. callPreviousFilterWrite(tail, session, writeRequest);
  4. }

在这里先要介绍一下DefaultIoFiterChain的数据格式，主要的属性如下：

  1. private final Map<String, Entry> name2entry = new ConcurrentHashMap<String, Entry>();
  2. /** The chain head */
  3. private final EntryImpl head;
  4. /** The chain tail */
  5. private final EntryImpl tail;

其中 head与tail都是DefaultIoFilterChain固有的属性，name2entity是我们为FilterChain添加的过滤器。因而IoFilterChain是用一个链表来保存过滤器的（('tail', prev: 'myFilter:ProtocolCodecFilter', next: 'null')），其中表头和表位都是固定的head和tail，他们对应的Filter也是专有的，HeadFilter和TailFilter. 
关键方法是callPreviousFilterWrite(tail, session, writeRequest);

  1. try {
  2. IoFilter filter = entry.getFilter();
  3. NextFilter nextFilter = entry.getNextFilter();
  4. filter.filterWrite(nextFilter, session, writeRequest);
  5. } catch (Throwable e) {
  6. writeRequest.getFuture().setException(e);
  7. fireExceptionCaught(e);
  8. }

从上面两个代码片段中，可以看出，IoFilterChain首先从列表中找到tail，从tail开始查找filter，顺序调用每个filter的filterWrite()方法。这里的‘顺序调用’，指的是从tail->head调用，也就是逆向调用Filter。但是看到filter.filterWrite(nextFilter, session, writeRequest)；这行代码中的参数可以发现，nextFilter，表面的意思是下一个过滤器，有点误解，感觉tail下一个过滤器不就是null吗，其实不然，进入filterWriter可知。

  1. Entry nextEntry = EntryImpl.this.prevEntry;
  2. callPreviousFilterWrite(nextEntry, session, writeRequest);

对于除head和tail过滤器外，其他的过滤器是如何工作的呢？我们看看ProtocolCodecFilter中的fireFilter方法，做了这样的处理：

  1. if ((message instanceof IoBuffer) || (message instanceof FileRegion)) {
  2. nextFilter.filterWrite(session, writeRequest);
  3. return;
  4. }

到这里，就明白了为什么session.write(IoBuffer.wrap())这样写出去，无法经过自己定义的过滤器了，原来在fireFilter中，对message做了判断，如果已经是IoBuffer类型的，就直接return了。 
最后执行的是HeadFilter的fireFilter方法，直接看内容：

  1. if (writeRequest.getMessage() instanceof IoBuffer) {
  2. IoBuffer buffer = (IoBuffer) writeRequest.getMessage();
  3. // I/O processor implementation will call buffer.reset()
  4. // it after the write operation is finished, because
  5. // the buffer will be specified with messageSent event.
  6. buffer.mark();
  7. int remaining = buffer.remaining();
  8. if (remaining == 0) {
  9. // Zero-sized buffer means the internal message
  10. // delimiter.
  11. s.increaseScheduledWriteMessages();
  12. } else {
  13. s.increaseScheduledWriteBytes(remaining);
  14. }
  15. } else {
  16. s.increaseScheduledWriteMessages();
  17. }
  18. s.getWriteRequestQueue().offer(s, writeRequest);
  19. if (!s.isWriteSuspended()) {
  20. s.getProcessor().flush(s);
  21. }

WriteRequestQueue的默认实现就是java.util.concurrent.ConcurrentLinkedQueue，舍去传入的session对象。

最新文章

  1. struts2学习笔记--使用servletAPI实现ajax的一个小Demo
  2. 选择c3p0作为连接池
  3. IOS之笑脸app
  4. [问题2014A06] 复旦高等代数 I（14级）每周一题（第八教学周）
  5. HDU 1498：50 years, 50 colors（二分图匹配）
  6. CodeForces 567C Geometric Progression 类似dp的递推统计方案数
  7. matlab中 mcc/mbuild/mex 区别
  8. CentOS 安装 ceph 单机版
  9. CSS画三角形图标
  10. 第1章 Linux文件类基础命令
  11. Dubbo基本原理机制
  12. linux test
  13. postgresql主从配置
  14. ConstraintLayout使用手册
  15. 普里姆Prim算法介绍
  16. ArrayList原理(一)
  17. Go语言管道
  18. VC++ 设置桌面壁纸
  19. JDK源码分析之concurrent包（三） -- Future方式的实现
  20. [学习笔记]对未来做出承诺的DP小结

热门文章

  1. linux SPI驱动——spidev之deive（五）
  2. 如何运用spring将dao注入到servlet中?
  3. JavaMelody tomcat应用监控
  4. python scrapy爬虫框架
  5. HDU 5374 Tetris （2015年多校比赛第7场）
  6. GS给客户单发包以及m_queGcWait(所有GC共享)
  7. EasyNVR RTSP转RTMP-HLS流媒体服务器前端构建之：使用BootstrapPagination以分页形式展示数据信息
  8. struct对齐
  9. notHere 对框架解决方案的框架预期处理
  10. ridge regression 无惩罚，导致预测结果空间过大而无实用价值