mina框架的使用非常简单，并且功能强大，将复杂的NIO操作封装成一套成熟的框架，在mina的使用中，我们只需要实现编解码器、信息处理器等就可以建立一个支持NIO通信的B/S程序。但是了解mina的源码对我们认识这个框架以及NIO都有很大帮助，下面我们就从mina以发送信息session.write()作为切入点探析mina是如何工作的。
1.发送信息
session.write(message);
从这句代码开始像远程IP地址发送信息，我们首先看一下write()方法。IoSession是一个接口，抽象类AbstractIoSession实现了IoSession中的write()方法：
public abstract class AbstractIoSession implements IoSession {
...
public WriteFuture write(Object message) {
return write(message, null);
}
public WriteFuture write(Object message, SocketAddress remoteAddress) {
//确定message是否为空、是否建立了链接等
...
FileChannel openedFileChannel = null;
//创建WriteFuture、WriteRequest，WriteRequest封装了我们要发送的message
WriteFuture writeFuture = new DefaultWriteFuture(this);
WriteRequest writeRequest = new DefaultWriteRequest(message, writeFuture, remoteAddress);
//得到过滤器链
IoFilterChain filterChain = getFilterChain();
filterChain.fireFilterWrite(writeRequest);
...
//关闭FileChannel
return writeFuture;
}
}
在write()方法的中部，首先创建了writeFuture用于获取发送的结果，writeRequest进一步封装了要发送的信息message。接着去获取session的过滤器链：
IoFilterChain filterChain = getFilterChain();
getFilterChain()是IoSession接口中的一个方法，具体实现由DummySession实现：
public class DummySession extends AbstractIoSession {
...
private final IoFilterChain filterChain = new DefaultIoFilterChain(this);
public IoFilterChain getFilterChain() {
return filterChain;
}
}
而write()中接下来：
filterChain.fireFilterWrite(writeRequest);
fireFilterWrite()方法在DefaultIoFilterChain类中实现：
public class DefaultIoFilterChain implements IoFilterChain {
...
public void fireFilterWrite(WriteRequest writeRequest) {
//这里首先载入的过滤器是filterChain的尾处理器tail，说明向外写数据的时候是从尾至头调用过滤器处理数据，而接收（读取）数据则相反。
callPreviousFilterWrite(tail, session, writeRequest);
}
...
}
再看DefaultIoFilterChain.callPreviousFilterWrite()方法：
private void callPreviousFilterWrite(Entry entry, IoSession session, WriteRequest writeRequest) {
try {
IoFilter filter = entry.getFilter(); //获取当前过滤器，也就是tail过滤器
NextFilter nextFilter = entry.getNextFilter(); //获取tail前面的一个过滤器
filter.filterWrite(nextFilter, session, writeRequest);
} catch (Exception e) {
writeRequest.getFuture().setException(e);
fireExceptionCaught(e);
} catch (Error e) {
writeRequest.getFuture().setException(e);
fireExceptionCaught(e);
throw e;
}
}
这里的形参：Entry是接口IoFilterChain中的内部接口，表示了过滤器链filterChain中的一个过滤器对象。而DefaultIoFilterChain中包含内部类EntryImpl是Entry接口的实现，DefaultIoFilterChain的构造函数中还创建了包含headFiler和tailFilter的filterChain：
public class DefaultIoFilterChain implements IoFilterChain {
private final EntryImpl head; //头过滤器
private final EntryImpl tail; //尾过滤器
//构造函数
public DefaultIoFilterChain(AbstractIoSession session) {
if (session == null) {
throw new IllegalArgumentException("session");
}
this.session = session;
head = new EntryImpl(null, null, "head", new HeadFilter()); //创建HeadFilter
tail = new EntryImpl(head, null, "tail", new TailFilter()); ////创建TailFilter
head.nextEntry = tail; //初始时只包含这两个过滤器
}
...
/**
* 过滤器内部类
**/
private final class EntryImpl implements Entry {
private EntryImpl prevEntry;
private EntryImpl nextEntry;
private final String name;
private IoFilter filter;
private final NextFilter nextFilter;
private EntryImpl(EntryImpl prevEntry, EntryImpl nextEntry, String name, IoFilter filter) {
...
this.prevEntry = prevEntry;
this.nextEntry = nextEntry;
this.name = name;
this.filter = filter;
this.nextFilter = new NextFilter() {
...
public void filterWrite(IoSession session, WriteRequest writeRequest) {
Entry nextEntry = EntryImpl.this.prevEntry; //nextEntry其实是前一个过滤器
callPreviousFilterWrite(nextEntry, session, writeRequest);
}
...
};
}
}
/**
*尾过滤器
**/
private static class TailFilter extends IoFilterAdapter {
@Override
public void sessionOpened(NextFilter nextFilter, IoSession session) throws Exception {
session.getHandler().sessionOpened(session);
}
//重写的其他方法于sessionOpened()类似，都是交给session的handler对应的函数去处理，有点像代理方式。
...
//尾过滤器不处理数据，直接将数据交给前面的过滤器
@Override
public void filterWrite(NextFilter nextFilter, IoSession session, WriteRequest writeRequest) throws Exception {
nextFilter.filterWrite(session, writeRequest);
}
}
}
我们知道将数据交给tail过滤器后，tail不会处理数据，而是直接交给它前面的过滤器，下面的一个方法filter.filterWrite(): 这个方法在继承于IoFilterAdapter的ProtocolCodecFilter中：
public class ProtocolCodecFilter extends IoFilterAdapter {
...
public void filterWrite(NextFilter nextFilter, IoSession session, WriteRequest writeRequest) throws Exception {
//从WriteRequest中获取message
Object message = writeRequest.getMessage();
//如果是IoBuffer、FileRegion类型的message，则交给下一个过滤器处理
if ((message instanceof IoBuffer) || (message instanceof FileRegion)) {
nextFilter.filterWrite(session, writeRequest);
return;
}
//获取sesson的编码器（filterChain.addLast()进去的）
ProtocolEncoder encoder = factory.getEncoder(session);
ProtocolEncoderOutput encoderOut = getEncoderOut(session, nextFilter, writeRequest);
...
try {
// 调用encoder的encode方法进行编码，并将编码结果传入encoderOut中，用于下一个编码器的处理
encoder.encode(session, message, encoderOut);
// Send it directly
Queue<Object> bufferQueue = ((AbstractProtocolEncoderOutput) encoderOut).getMessageQueue();
// Write all the encoded messages now
while (!bufferQueue.isEmpty()) {
Object encodedMessage = bufferQueue.poll();
if (encodedMessage == null) {
break;
}
// Flush only when the buffer has remaining.
if (!(encodedMessage instanceof IoBuffer) || ((IoBuffer) encodedMessage).hasRemaining()) {
SocketAddress destination = writeRequest.getDestination();
WriteRequest encodedWriteRequest = new EncodedWriteRequest(encodedMessage, null, destination);
nextFilter.filterWrite(session, encodedWriteRequest);
}
}
// Call the next filter
nextFilter.filterWrite(session, new MessageWriteRequest(writeRequest));
} catch (Exception e) {
ProtocolEncoderException pee;
// Generate the correct exception
if (e instanceof ProtocolEncoderException) {
pee = (ProtocolEncoderException) e;
} else {
pee = new ProtocolEncoderException(e);
}
throw pee;
}
}
}
我们看到过滤器链将数据一个个的处理并且再将数据传入上一个过滤器中，那么最后一个处理器是怎么处理数据的呢？下面来看一下headFilter：
private class HeadFilter extends IoFilterAdapter {
@SuppressWarnings("unchecked")
@Override
public void filterWrite(NextFilter nextFilter, IoSession session, WriteRequest writeRequest) throws Exception {
AbstractIoSession s = (AbstractIoSession) session;
// Maintain counters.
if (writeRequest.getMessage() instanceof IoBuffer) {
IoBuffer buffer = (IoBuffer) writeRequest.getMessage();
// I/O processor implementation will call buffer.reset()
// it after the write operation is finished, because
// the buffer will be specified with messageSent event.
buffer.mark();
int remaining = buffer.remaining();
if (remaining > 0) {
s.increaseScheduledWriteBytes(remaining);
}
} else {
s.increaseScheduledWriteMessages();
}
WriteRequestQueue writeRequestQueue = s.getWriteRequestQueue();
if (!s.isWriteSuspended()) {
if (writeRequestQueue.isEmpty(session)) {
// session通过IoProcessor写信息
s.getProcessor().write(s, writeRequest);
} else {
s.getWriteRequestQueue().offer(s, writeRequest);
s.getProcessor().flush(s);
}
} else {
s.getWriteRequestQueue().offer(s, writeRequest);
}
}
}
