标签：session Mina private final 解读 源码 mina new public
阅读笔记（一）-整体解读
Apache MINA is a network application framework which helps users develop high performance and high scalability network applications easily. It provides an abstract ·event-driven · asynchronous API over various transports such as TCP/IP and UDP/IP via Java NIO.
写过NIO的人都知道，NIO里主要有那么几部分，用于内容切换的各类缓冲区，当然我们印象最深的就是ByteBuffer；用于IO服务直接连通的通道Channel，这里有用于文件和网络的通道；有用于实现NIO socket Reactor模式的选择器selector；和处理字符集的Charset。而mina就是将这几部分有机的整合，实现了这个高性能的框架。
在看代码之前，我们要先了解mina到底为我们提供了什么，我按照源码的结构整理如下：
|
用于缓冲区的IoBuffer |
org.apache.mina.core.buffer |
|
用于提供连接的service |
org.apache.mina.core.service org.apache.mina.transport.* |
|
用于提供两端状态的session |
org.apache.mina.core.session |
|
用于拦截所有IO事件和请求的filter chain和各类拦截器（在IoService和IoHandler之间） |
org.apache.mina.core.filterchain org.apache.mina.filter.* |
|
用于处理IO事件的handler |
org.apache.mina.handler.* |
|
用于实现异步IO操作的 future |
org.apache.mina.core.future |
|
用于实现IO轮询的的polling |
org.apache.mina.core.polling |
|
用于实现代理的proxy |
org.apache.mina.proxy.* |
至此，mina-core-2.0.7中的代码大致就可以分成上述这几类了。而我们对源码的浏览也会按照这几部分逐一进行。
在开始阅读代码之前，我们再熟悉一下mina在通信时会做的事情，这样的事情不是徒劳的，对过程理解的越透彻对代码的分析也会越深刻，我们不但要学习代码里的设计模式和组织结构，更要去理解里面的实现逻辑。没装UML的工具，就拿PPT画了：
这里的变现没有像之前mina通信那篇文章里一样，描述具体的编码和解码过程，这个图主要描述了两端对接的过程，是一个宏观上的通信过程：
1、 图中的一对应的是service执行的过程，首先肯定是要建立连接，对于服务端是Accept，客户端则是connect。
2、 图中的二，指的是在每次连接之后就会产生一个记录通信信息的session，我们看mina官方的描述：every time a client connects to the server, a new session is created, and will be kept in memory until the client is disconnected.
3、 图中的三指的是过滤器链，这里主要实现我们对通信的要求，比如编码、解码、日志等。
4、 图中的四是Handler，是filter chain的最后一步，通过adapter接入了session的整个生命周期。
服务器端反之亦然。
上面的图示mina官方给出的，表述的同一个意思。在IoHandler这里mina是这么给出描述的： The interface is hub of all activities done at the end of the Filter Chain.
分析清楚了整个代码的体系，后面开始，我们就可以针对每一个部分详细分析了。我们会从最基本的缓冲区开始，这部分相对比较独立，之后会根据连接过程进行一步步分析。后面的一篇，我会从IoBuffer开始写起。
阅读笔记（二）- IoBuffer的封装
在阅读IoBuffer源码之前，我们先看Mina对IoBuffer的描述：A byte buffer used by MINA applications. This is a replacement for ByteBuffer. 这是一个对ByteBuffer的replacement，同样是用作缓冲区，做内容的切换和承载的容器，为什么要用重新封装ByteBuffer，MINA是这么给出解释的Two Reasons：
l It doesn't provide useful getters and putters
l It is difficult to write variable-length data due to its fixed capacity
用过ByteBuffer的人可能经常会遇到BufferOverflowException这样的异常，原因是buffer在初始化allocate之后就不能再自动的改变大小了，如果项目很规整，约定的很好，那可能不太会出意外，怕就怕项目一大，好多东西就乱套了。所以在阅读IoBuffer源码的时候，我们会着重看它和ByteBuffer之间的差异。另外一点，就是IoBuffer作为一个应用框架的工具，必然会提供比原生Buffer更便捷的方法，比如IoBuffer中可以直接put和get String，可以直接将内容转成十六进制等等。
用法很简单，我倒是想从如何将一个已有的类进行封装和扩展的角度来看IoBuffer的源码。在看MINA的源码之前，我们有必要稍稍回顾一下ByteBuffer的构成：
ByteBuffer继承了Buffer类，这个继承关系约定了Buffer系列中特定的操作形式（有点儿像指针），limit/position/mark/capacity，以及在遍历中使用的hasRemaining。然后通过两个静态方法来构建出ByteBuffer：
使用Heap空间，堆空间的构造采用申请byte数组：
public static ByteBuffer allocate(int capacity) {
if (capacity < 0)
throw new IllegalArgumentException();
return new HeapByteBuffer(capacity, capacity);
}使用direct memory，这块内存的开辟就比较麻烦了，好多都是采用了Bit和native的方法：
public static ByteBuffer allocateDirect(int capacity) {
return new DirectByteBuffer(capacity);
}除了构造之外，剩下的主要是对数据的操作方法，wrap、get和put，下面的图没有截全，还有好多方法：
IoBuffer及其相关的类均在org.apache.mina.core.buffer下，IoBuffer定义了buffer使用的规则，AbseractIoBuffer提供了具体的实现：
IoBuffer没有继承任何类，只是实现了comparable接口，我们注意到IoBuffer类修饰符用的是abstract，跟ByteBuffer也是用abstract修饰，至于为什么要用abstract，我觉得也容易理解，毕竟这是一个要对外直接使用的类，同时需要对实现进行规则和扩展：
public abstract class IoBuffer implements Comparable<IoBuffer>在IoBuffer的一系列代码阅读中，你可以看到抽象类之间的继承，内部类的使用情况等等，后面，我会通过一个删减版的例子来盘点这中间的关系，所以大片的源码就不贴了。
UML工具不会用，关键是怕用错了，还是用PPT画了。囧一个，大家有好那种可以一键生成的工具推荐一下，我之前用的是JUDE和Visio。上图画出了IoBuffer中几个重要类之间的关系，两个内部类均继承了AbstractIoBuffer，AbstractIoBuffer和IoBufferWrapper均实现了IoBuffer中的具体操作部分。IoBufferAllocator接口主要定义了为缓冲区开辟空间的方法，所以IoBuffer中需要引用来自IoBufferAllocator的对象。
在IoBuffer中，我们熟知的allocate和wrap方法被声明成了static，通过引用IoBufferAllocator接口中的对象来实现，而其他诸如get、put等操作的方法都定义为abstract了，让其子类得以实现。IoBuffer中我们还值得关注的主要见我之前写过的一篇文章《IoBuffer和ByteBuffer》。
下面是这些中产生buffer的接口IoBufferAllocator和其实现类：
接口很简单，就定义了几个在IoBuffer中已经被static修饰的方法。有两个类都实现了IoBufferAllocator，但是在IoBuffer中使用的是SimpleBufferAllocator：
/** The allocator used to create new buffers */
private static IoBufferAllocator allocator = new SimpleBufferAllocator();
/** A flag indicating which type of buffer we are using : heap or direct */
private static boolean useDirectBuffer = false;所以我们主要关注SimpleBufferAllocator:
public IoBuffer allocate(int capacity, boolean direct) {
return wrap(allocateNioBuffer(capacity, direct));
}
public ByteBuffer allocateNioBuffer(int capacity, boolean direct) {
ByteBuffer nioBuffer;
if (direct) {
nioBuffer = ByteBuffer.allocateDirect(capacity);
} else {
nioBuffer = ByteBuffer.allocate(capacity);
}
return nioBuffer;
}
public IoBuffer wrap(ByteBuffer nioBuffer) {
return new SimpleBuffer(nioBuffer);
}
public void dispose() {
// Do nothing
}这是接口中定义的几个方法，这里调用内部类SimpleBuffer来生成相应的buffer，又由于SimpleBuffer继承了AbstractIoBuffer，所以真正实现的代码在AbstractIoBuffer中（这里有点儿绕，大家结合上面的图和源码一起读）。而且注意构造方法的protected关键字的使用：
private ByteBuffer buf;
protected SimpleBuffer(ByteBuffer buf) {
super(SimpleBufferAllocator.this, buf.capacity());
this.buf = buf;
buf.order(ByteOrder.BIG_ENDIAN);
}
protected SimpleBuffer(SimpleBuffer parent, ByteBuffer buf) {
super(parent);
this.buf = buf;
}看到了吧，底层还是用的NIO中的ByteBuffer。至于怎么实现AutoExpand这样的方法，我觉得不是源码的重点，这些都是算法上的事情，如果你不关注算法，可以稍稍看看即可，而且好多都是native的实现，也看不到。而我这边主要关注的还是他们之间的结构。
上图左边的路走通了，我们来走右边的路，右边主要看AbstractIoBuffer，他是IoBuffer的具体实现，但是它也是一个抽象类，也要被其他类继承，用于扩展。AbstractIoBuffer中，大多类都是final的。而且这里面主要的实现都是在处理limit/position/mark/capacity这之间的关系。而CachedBufferAllocator主要用于实现IoBuffer中自动扩展AutoExpand和收缩: that caches the buffers which are likely to be reused during auto-expansion of the buffers.
----------------------------------------------------------
最后，我们将上面的叙述用一个删减版的代码来模拟一下，这样有助于理解代码的结构，以后遇到类似的情况就可以类似的处理，我更希望，能在分析完所有源码之后，就能呈现一个类似的框架出来，不过这个真的只是想想，毕竟没那么多时间，如果你有时间，可以试着去阉割一下mina。
首先是IoBuffer：
package org.apache.mina.core.rewrite.buffer;
/**
* IoBuffer
*
* @author ChenHui
*
*/
public abstract class IoBuffer {
private static IoBufferAllocator allocator=new SimpleBufferAllocator();
private static boolean direct;
protected IoBuffer() {
// do nothing
}
public static IoBuffer allocate(int capacity) {
return allocator.allocate(capacity, direct);
}
public static IoBuffer wrap(byte[] byteArray, int offset, int length){
//TODO
return null;
}
public abstract IoBuffer get();
public abstract IoBuffer put(byte b);
public abstract boolean other();
}然后是他的继承：
package org.apache.mina.core.rewrite.buffer;
import java.nio.ByteBuffer;
/**
*
* @author ChenHui
*
*/
public abstract class AbstractIoBuffer extends IoBuffer{
protected AbstractIoBuffer(ByteBuffer buffer){
//TODO
}
@Override
public IoBuffer get() {
// TODO Auto-generated method stub
return null;
}
@Override
public IoBuffer put(byte b) {
// TODO Auto-generated method stub
return null;
}
}allocator:
package org.apache.mina.core.rewrite.buffer;
import java.nio.ByteBuffer;
/**
*
* @author ChenHui
*
*/
public interface IoBufferAllocator {
IoBuffer allocate(int capacity, boolean direct);
IoBuffer wrap(ByteBuffer nioBuffer);
ByteBuffer allocateNioBuffer(int capacity, boolean direct);
void dispose();
}allocator的实现：
package org.apache.mina.core.rewrite.buffer;
import java.nio.ByteBuffer;
/**
*
* @author ChenHui
*
*/
public class SimpleBufferAllocator implements IoBufferAllocator{
@Override
public IoBuffer allocate(int capacity, boolean direct) {
return wrap(allocateNioBuffer(capacity, direct));
}
@Override
public IoBuffer wrap(ByteBuffer nioBuffer) {
return new SimpleBuffer(nioBuffer);
}
@Override
public ByteBuffer allocateNioBuffer(int capacity, boolean direct) {
ByteBuffer nioBuffer;
if (direct) {
nioBuffer = ByteBuffer.allocateDirect(capacity);
} else {
nioBuffer = ByteBuffer.allocate(capacity);
}
return nioBuffer;
}
@Override
public void dispose() {
// TODO Auto-generated method stub
}
private class SimpleBuffer extends AbstractIoBuffer{
@SuppressWarnings("unused")
ByteBuffer buffer;
protected SimpleBuffer(ByteBuffer buffer){
super(buffer);
this.buffer=buffer;
}
@Override
public boolean other() {
// TODO Auto-generated method stub
return false;
}
/**这里重写是为了打印方便*/
@Override
public String toString() {
System.out.println(buffer);
return super.toString();
}
}
}最后是测试类和测试结果：
package org.apache.mina.core.rewrite.buffer;
public class Test {
public static void main(String[] args) {
IoBuffer buffer=IoBuffer.allocate(1024);
System.out.println(buffer);
}
}控制台输出：
java.nio.HeapByteBuffer[pos=0 lim=1024 cap=1024]
org.apache.mina.core.rewrite.buffer.SimpleBufferAllocator$SimpleBuffer@1da12fc0阅读笔记（三）-Mina的连接IoAccpetor
其实在mina的源码中，IoService可以总结成五部分service责任、Processor线程处理、handler处理器、接收器和连接器，分别对应着IoService、IoProcessor、IoHandler、IoAcceptor和IoConnector。在代码的中有如下包跟IoService关系密切：
org.apache.mina.core.service
org.apache.mina.transport.*
org.apache.mina.core.polling 这个包主要是实现了轮询策略其中core.service包中主要定义了上面五个部分的接口，以及IoHandler和IoProcessor的实现（Handler是一种常见的设计模式，具体的业务其实并没有多少，只是在结构上将层次划分的更清楚。Processor是mina内部定义的接口，一般不对外使用，用mina官方的话，主要performs actual I/O operations for IoSession. 这一部分我想放在IoSession中来讲）。而在transport包中实现了具体的连接方式，当然，这部分也是我们今天阅读的重点。我想要关注的也是mina底层如何用NIO实现各种连接。
所以这一节的重点是IoAcceptor和IoConnector，Handler和IoService只是稍稍带过，写点儿用法和构成。先看mina对IoService的介绍：IoService provides basic I/O Service and manages I/O Sessions within MINA.
上面的图简单介绍了IoService的职责，以及其具体实现类AbstractIoService中的职责。在比较大的框架中，都是采用了大量的抽象类之间继承，采用层级实现细节这样的方式来组织代码。所以在mina中看到Abstract开头的类，并不仅仅只是一个抽象，其实里面也包含很多的实现了。
IoService用来管理各种IO服务，在mina中，这些服务可以包括session、filter、handler等。在AbstractIoService中，也是通过线程池来装载这些服务的：
private final Executor executor;
private final boolean createdExecutor;
protected AbstractIoService(IoSessionConfig sessionConfig, Executor executor) {
//省略。。。
if (executor == null) {
this.executor = Executors.newCachedThreadPool();
createdExecutor = true;
} else {
this.executor = executor;
createdExecutor = false;
}
threadName = getClass().getSimpleName() + '-' + id.incrementAndGet();
}然后我们关注一下service中的dispose方法，学习一下线程安全在这里的用法：
/**
* A lock object which must be acquired when related resources are
* destroyed.
*/
protected final Object disposalLock = new Object();
private volatile boolean disposing;
private volatile boolean disposed;
private final boolean createdExecutor;
public final void dispose(boolean awaitTermination) {
if (disposed) {
return;
}
synchronized (disposalLock) {
if (!disposing) {
disposing = true;
try {
dispose0();
} catch (Exception e) {
ExceptionMonitor.getInstance().exceptionCaught(e);
}
}
}
if (createdExecutor) {
ExecutorService e = (ExecutorService) executor;
e.shutdownNow();
if (awaitTermination) {
//Thread.currentThread().setName();
try {
LOGGER.debug("awaitTermination on {} called by thread=[{}]", this, Thread.currentThread().getName());
e.awaitTermination(Integer.MAX_VALUE, TimeUnit.SECONDS);
LOGGER.debug("awaitTermination on {} finished", this);
} catch (InterruptedException e1) {
LOGGER.warn("awaitTermination on [{}] was interrupted", this);
// Restore the interrupted status
Thread.currentThread().interrupt();
}
}
}
disposed = true;
}为了多线程之间变量内存的可见性，防止抢占资源时候出现意想不到的问题，这里用了volatile关键字修饰布尔型变量这样的经典用法，同时使用内置锁机制控制线程的访问。与IoService比较相关的还有个TransportMetadata，这个类主要记录了IoService的相关信息，在使用中，也不是很常见，所以略过这部分了。更多的线程运用在IoProcessor中会体现，这部分放到后面写，今天的主要目的还是连接IoAcceptor和IoConnector。
从现在开始就要复杂了，我们先从服务器的接收端开始写起，也就是IoAccpetor，先上一张宏观的图，来理清思路，图来自mina官网：
简单介绍一下，连接的实现有四种：
We have many of those implementing classes
- NioSocketAcceptor : the non-blocking Socket transport IoAcceptor
- NioDatagramAcceptor : the non-blocking UDP transport IoAcceptor
- AprSocketAcceptor : the blocking Socket transport IoAcceptor, based on APR
- VmPipeSocketAcceptor : the in-VM IoAcceptor
我们按照图上用箭头标出的两条路来分析我们最常用的NioSocketAcceptor，回顾一下调用这个类的过程：
// Create a TCP acceptor
IoAcceptor acceptor = new NioSocketAcceptor();
// Associate the acceptor to an IoHandler instance (your application)
acceptor.setHandler(this);
// Bind : this will start the server...
acceptor.bind(new InetSocketAddress(PORT));
System.out.println("Server started...");从左边的路走起：
接口IoAcceptor直接继承了IoService接口，并定义了自己特有的操作。其操作的具体实现由AbstractIoAcceptor完成（注意是上图左边的类来实现的）。我们继续从左边往下看，与IoAcceptor直接关联的是SocketAcceptor接口，它们（IoAcceptor和SocketAcceptor）之间也是接口的继承关系，所以根据前面的经验，我们可以猜测，SocketAcceptor一定又新定义了一些属于自己需要去实现的操作，这样做肯定是为了与另一种实现DatagaramAcceptor来区别，事实也确实如此，看下图：
这里需要注意的是上图被框出来的部分，他们的返回类型均被缩小了（我记不得是向下转型还是向上转型，应该是向下吧，现在返回的都是子类），这样实现的好处是绝对的专一，避免了转型上的错误。
在看NioSocketAcceptor之前，我们还是要先看右边这条路：
public abstract class AbstractIoAcceptor extends AbstractIoService implements IoAcceptor回顾一下，AbstractIoService实现了对session的管理，IoAcceptor定义了一些建立连接时用到的一系列方法，这样一来，AbstractIoAcceptor一来有了对session使用的功能，二来需要实现建立连接里需要用到的那些方法，理清楚了这些，我们可以看AbstractIoAcceptor的具体实现：
private final List<SocketAddress> defaultLocalAddresses = new ArrayList<SocketAddress>();
private final List<SocketAddress> unmodifiableDefaultLocalAddresses = Collections
.unmodifiableList(defaultLocalAddresses);
private final Set<SocketAddress> boundAddresses = new HashSet<SocketAddress>();
private boolean disconnectOnUnbind = true;
/**
* The lock object which is acquired while bind or unbind operation is performed.
* Acquire this lock in your property setters which shouldn't be changed while
* the service is bound.
*/
protected final Object bindLock = new Object();这里有三点要说：
l 这里的Address不是用了list就是用了set，注意这些都是用来保存LocalAddress的，mina在设计的时候考虑的很全面，服务器可能会有多个网卡的啊。
l 解释下unmodifiableList，JDK自带方法：Returns an unmodifiable view of the specified list. This method allows modules to provide users with "read-only" access to internal lists. Query operations on the returned list "read through" to the specified list, and attempts to modify the returned list, whether direct or via its iterator, result in an UnsupportedOperationException. The returned list will be serializable if the specified list is serializable. Similarly, the returned list will implement RandomAccess if the specified list does.
l 时刻注意线程安全的问题。
AbstractIoAcceptor主要是对localAddress的操作，里面涉及到集合类的使用和内置锁的使用，其他没有什么特别的，大家可以看看源码，容易理解。主要可以看看里面的bind方法，这个里面涉及了两层锁。
AbstractPollingIoAcceptor在实现连接中至关重要的一个类，他是socket轮询策略的主要实现，有那么几点要关注：
l polling strategy : The underlying sockets will be checked in an active loop and woke up when an socket needed to be processed.
l An Executor will be used for running client accepting and an AbstractPollingIoProcessor will be used for processing client I/O operations like reading, writing and closing.（将连接和业务分开，AbstractPollingIoProcessor部分会在session部分写）。
l All the low level methods for binding, accepting, closing need to be provided by the subclassing implementation.
这个类光看用到的工具就知道他不简单了，之前我们见过的处理线程安全问题都只是用了内置锁，这里终于用到了concurrent包里的东西了。这个类里没有出现NIO中selector这样的东西，而是做了一种策略，这种策略是在线程处理上的，采用队列和信号量协同处理socket到来时候连接的策略。推荐大家仔细看看这个类，里面的注释很详细，我列几个成员变量，你看看有兴趣没：
/** A lock used to protect the selector to be waked up before it's created */
private final Semaphore lock = new Semaphore(1);
private final IoProcessor<S> processor;
private final boolean createdProcessor;
private final Queue<AcceptorOperationFuture> registerQueue = new ConcurrentLinkedQueue<AcceptorOperationFuture>();
private final Queue<AcceptorOperationFuture> cancelQueue = new ConcurrentLinkedQueue<AcceptorOperationFuture>();
private final Map<SocketAddress, H> boundHandles = Collections.synchronizedMap(new HashMap<SocketAddress, H>());
private final ServiceOperationFuture disposalFuture = new ServiceOperationFuture();
/** A flag set when the acceptor has been created and initialized */
private volatile boolean selectable;
/** The thread responsible of accepting incoming requests */
private AtomicReference<Acceptor> acceptorRef = new AtomicReference<Acceptor>();
protected boolean reuseAddress = false;在这个类里还有一个Acceptor的内部类，实现runnable接口，主要用作接收客户端的请求。这个类也有可看性。这里面的东西不是很好写，需要大家自己去细细品味。
看最后一个类，两边的焦点，NioSocketAcceptor。看之前在AbstractPollingIoAcceptor里有句话：All the low level methods for binding, accepting, closing need to be provided by the subclassing implementation.这里总该有NIO的一些东西了：
public final class NioSocketAcceptor extends AbstractPollingIoAcceptor<NioSession, ServerSocketChannel> implements
SocketAcceptor {
private volatile Selector selector;我想看到的东西终于来了，selector，先不说，接着往下看：
@Override
protected void init() throws Exception {
selector = Selector.open();
}
@Override
protected NioSession accept(IoProcessor<NioSession> processor, ServerSocketChannel handle) throws Exception {
SelectionKey key = handle.keyFor(selector);
if ((key == null) || (!key.isValid()) || (!key.isAcceptable())) {
return null;
}
// accept the connection from the client
SocketChannel ch = handle.accept();
if (ch == null) {
return null;
}
return new NioSocketSession(this, processor, ch);
}
@Override
protected ServerSocketChannel open(SocketAddress localAddress) throws Exception {
// Creates the listening ServerSocket
ServerSocketChannel channel = ServerSocketChannel.open();
boolean success = false;
try {
// This is a non blocking socket channel
channel.configureBlocking(false);
// Configure the server socket,
ServerSocket socket = channel.socket();
// Set the reuseAddress flag accordingly with the setting
socket.setReuseAddress(isReuseAddress());
// and bind.
socket.bind(localAddress, getBacklog());
// Register the channel within the selector for ACCEPT event
channel.register(selector, SelectionKey.OP_ACCEPT);
success = true;
} finally {
if (!success) {
close(channel);
}
}
return channel;
}都是最基本的java NIO吧，是不是很熟悉，是不是有种相见恨晚的感觉，是不是很简单。当然简单是相对的，因为之间做了那么多铺垫，所以程序设计真的是一种艺术，看你怎么去设计。
只是你想过没有，这是一个final的类，不能继承，那我们在调用的时候从来没有写acceptor.open，这时候这个方法什么时候执行呢，在这个类里面也没有展示出来。这个疑问先留着，后面会说，这当然也是一种机制。
阅读笔记（四）—Mina的连接IoConnector
上一篇写的是IoAcceptor是服务器端的接收代码，今天要写的是IoConnector，是客户端的连接器。在昨天，我们还留下一些问题没有解决，这些问题今天同样会产生，但是都要等到讲到session的时候才能逐步揭开。先回顾一下问题：
l 我们已经在AbstractPollingIoAcceptor中看到了，mina是将连接（命令）和业务（读写）分不同线程处理的，但是我们还没有看到mina是如何实现对这些线程的管理的。
l 在昨天的最后，我们看到在NioSocketAcceptor中的具体NIO实现，但是我们还没有看到mina是在哪里调用了这些具体操作的。当然这也是mina对连接线程管理的一部分。
这些问题今天也会出现，因为从名字上就能看出，IoConnector和IoAcceptor的构造相差不大，所以在写connector的分析时，主要会从结构和差异上入手，最后再给出昨天没写完的删减版的Acceptor。先回顾一下客户端连接的代码：
// 创建一个非阻塞的客户端程序
IoConnector connector = new NioSocketConnector();
// 设置链接超时时间
connector.setConnectTimeout(30000);
// 添加过滤器
connector.getFilterChain().addLast(
"codec",
new ProtocolCodecFilter(new MessageCodecFactory(
new InfoMessageDecoder(Charset.forName("utf-8")),
new InfoMessageEncoder(Charset.forName("utf-8")))));
// 添加业务逻辑处理器类
connector.setHandler(new ClientHandler());
IoSession session = null;
try {
ConnectFuture future = connector.connect(new InetSocketAddress(
HOST, PORT));// 创建连接
future.awaitUninterruptibly();// 等待连接创建完成
session = future.getSession();// 获得session
还是先看IoConnector的结构图，图来自mina官网，用XMind绘制的。在写构成之前，我们还是先看一下mina官网对这些connector的介绍：
As we have to use an IoAcceptor for servers, you have to implement the IoConnector. Again, we have many implementation classes :
- NioSocketConnector : the non-blocking Socket transport Connector
- NioDatagramConnector : the non-blocking UDP transport * Connector*
- AprSocketConnector : the blocking Socket transport * Connector*, based on APR
- ProxyConnector : a Connector providing proxy support
- SerialConnector : a Connector for a serial transport
- VmPipeConnector : the in-VM * Connector*
其中，NioSocketConnector是我们最常用到的，proxy方式虽然在mina的源码中也花了大篇幅去撰写，但可惜的是很少有相关的文档，所以学习的成本还挺高的。今天我们主要还是按照上图画的两条路来看NioSocketConnector。
和昨天一样，我们还是从左边的路走起，看interface IoConnector，这个接口主要定义了连接的方法以及socket连接时用到的参数。在mina中通过IoFuture来描述、侦听在IoSession上实现的异步IO操作，所以这IoConnector中的connect方法都返回了一个ConnectFuture实例。
而在SocketConnector接口中的定义中就显得更简单了，它和IoConnector之间也是接口的继承关系，在SocketConnector中就定义了两类方法，一个对远程地址的get和set，一个拿到session的配置。这些都容易理解。
再来看右边，AbstractIoConnector，这个抽象类主要作用就是实现IoConnector里定义的操作，至于他又继承了AbstractIoService，一是为了用到父类（AbstractIoService）的方法，二是为了将那些父类没有实现的方法继续传递下去，让它（AbstractIoConnector）的子类去实现。所以看多了，好多结构也能看明白了，这里我觉得主要要学习的还是接口、抽象类之间的引用关系。
继续看AbstractIoConnector，这个类主要是实现了connect的逻辑操作（封装了连接前后的一些必要执行步骤和check一些状态），具体的连接操作还是让子类去实现，这个和上篇写的AbstractIoAcceptor一模一样，在AbstractIoAcceptor中，主要也是封装了bind的逻辑操作，真正的bind过程是让子类去实现的简单看下代码：
public final ConnectFuture connect(SocketAddress remoteAddress, SocketAddress localAddress,
IoSessionInitializer<? extends ConnectFuture> sessionInitializer) {
if (isDisposing()) {
throw new IllegalStateException("The connector has been disposed.");
}
if (remoteAddress == null) {
throw new IllegalArgumentException("remoteAddress");
}
if (!getTransportMetadata().getAddressType().isAssignableFrom(remoteAddress.getClass())) {
throw new IllegalArgumentException("remoteAddress type: " + remoteAddress.getClass() + " (expected: "
+ getTransportMetadata().getAddressType() + ")");
}
if (localAddress != null && !getTransportMetadata().getAddressType().isAssignableFrom(localAddress.getClass())) {
throw new IllegalArgumentException("localAddress type: " + localAddress.getClass() + " (expected: "
+ getTransportMetadata().getAddressType() + ")");
}
if (getHandler() == null) {
if (getSessionConfig().isUseReadOperation()) {
setHandler(new IoHandler() {
public void exceptionCaught(IoSession session, Throwable cause) throws Exception {
// Empty handler
}
public void messageReceived(IoSession session, Object message) throws Exception {
// Empty handler
}
public void messageSent(IoSession session, Object message) throws Exception {
// Empty handler
}
public void sessionClosed(IoSession session) throws Exception {
// Empty handler
}
public void sessionCreated(IoSession session) throws Exception {
// Empty handler
}
public void sessionIdle(IoSession session, IdleStatus status) throws Exception {
// Empty handler
}
public void sessionOpened(IoSession session) throws Exception {
// Empty handler
}
});
} else {
throw new IllegalStateException("handler is not set.");
}
}
return connect0(remoteAddress, localAddress, sessionInitializer);
}
protected abstract ConnectFuture connect0(SocketAddress remoteAddress, SocketAddress localAddress,
IoSessionInitializer<? extends ConnectFuture> sessionInitializer);Connect0才是最后具体的操作，而这一步操作在这个类中被没有给出实现。具体实现放在了AbstractPollingIoConnector上。和昨天一样，这些设计都是对称的，我们还是看三点：
l implementing client transport using a polling strategy
l A Executor will be used for running client connection, and an AbstractPollingIoProcessor will be used for processing connected client I/O operations like reading, writing and closing.
l All the low level methods for binding, connecting, closing need to be provided by the subclassing implementation
至于内部的具体实现，那跟acceptor中没什么区别，连使用的工具类都差别不大，这部分就很容易读懂了，只不过一个是bind一个是connect。
最后我们看NioSocketConnector，具体连接的实现类，只有一个成员变量和NioSocketAcceptor一样：
private volatile Selector selector;
@Override
protected SocketChannel newHandle(SocketAddress localAddress) throws Exception {
SocketChannel ch = SocketChannel.open();
int receiveBufferSize = (getSessionConfig()).getReceiveBufferSize();
if (receiveBufferSize > 65535) {
ch.socket().setReceiveBufferSize(receiveBufferSize);
}
if (localAddress != null) {
ch.socket().bind(localAddress);
}
ch.configureBlocking(false);
return ch;
}只是需要注意，这里面专门有个内部类来处理selectionkey，将遍历的过程都抽离出来了，这个和我们用NIO的一般写法稍有不同，这样做的好处也是为了复用：
private static class SocketChannelIterator implements Iterator<SocketChannel> {
private final Iterator<SelectionKey> i;
private SocketChannelIterator(Collection<SelectionKey> selectedKeys) {
this.i = selectedKeys.iterator();
}
/**
* {@inheritDoc}
*/
public boolean hasNext() {
return i.hasNext();
}
/**
* {@inheritDoc}
*/
public SocketChannel next() {
SelectionKey key = i.next();
return (SocketChannel) key.channel();
}
/**
* {@inheritDoc}
*/
public void remove() {
i.remove();
}
}---------------------------------------------------------
补一个上篇就应该发的acceptor的阉割版，写这样的东西主要还是为了理清楚结构。我主要是把内容简化了，但是结构都没有变，核心的成员变量也没有少：
起点IoService:
package org.apache.mina.core.rewrite.service;
/**
* IO Service --handler/processor/acceptor/connector
*
* @author ChenHui
*
*/
public interface IoService {
/** 添加listener */
void addListener(IoServiceListener listener);
/** 销毁 */
void dispose(boolean awaitTermination);
/** 设置handler */
IoHandler getHandler();
void setHandler(IoHandler handler);
/** 管理session */
int getManagedSessionCount();
boolean isActive();
}左边的路
package org.apache.mina.core.rewrite.service;
import java.io.IOException;
import java.net.SocketAddress;
import java.util.Set;
/**
* 注意接口的继承，这里的方法都是新定义的
*
* Acceptor 主要用于：Accepts incoming connection, communicates with clients, and
* fires events to IoHandler
*
* @author ChenHui
*/
public interface IoAcceptor extends IoService {
SocketAddress getLocalAddress();
Set<SocketAddress> getLocalAddresses();
void bind(SocketAddress localAddress) throws IOException;
void bind(Iterable<? extends SocketAddress> localAddresses) throws IOException;
void unbind(SocketAddress localAddress);
/**没有写到IoSession 所以暂时不用*/
//IoSession newSession(SocketAddress remoteAddress,SocketAddress localAddress);
}SocketAcceptor：
package org.apache.mina.rewrite.transport.socket;
import java.net.InetSocketAddress;
import org.apache.mina.core.rewrite.service.IoAcceptor;
public interface SocketAcceptor extends IoAcceptor {
InetSocketAddress getLocalAddress();
void setDefaultLocalAddress(InetSocketAddress localAddress);
public boolean isReuseAddress();
// ...
// SocketSessionConfig getSessionConfig();
}再看右边的
package org.apache.mina.core.rewrite.service;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
public abstract class AbstractIoService implements IoService {
private static final AtomicInteger id = new AtomicInteger();
private final String threadName;
private final Executor executor;
private final boolean createdExecutor;
private IoHandler handler;
// 用于安全的关闭
protected final Object disposalLock = new Object();
private volatile boolean disposing;
private volatile boolean disposed;
/**
*
* @param param
* sessionConfig IoSessionConfig
* @param executor
* used for handling execution of IO event. can be null
*/
protected AbstractIoService(Object param, Executor executor) {
// TODO listener & session config
if (executor == null) {
this.executor = Executors.newCachedThreadPool();
createdExecutor = true;
} else {
this.executor = executor;
createdExecutor = false;
}
threadName = getClass().getSimpleName() + "-" + id.incrementAndGet();
}
@Override
public void addListener(IoServiceListener listener) {
// TODO add listener
}
/**注意这个不是override来的*/
protected final void ececuteWorker(Runnable worker, String suffix){
String actualThreadName=threadName;
if(suffix!=null){
actualThreadName=actualThreadName+"-"+suffix;
}
executor.execute(worker);
}
@Override
public void dispose(boolean awaitTermination) {
if (disposed) {
return;
}
synchronized (disposalLock) {
if (!disposing) {
disposing = true;
try {
/** 真正的关闭方法TODO */
dispose0();
} catch (Exception e) {
e.printStackTrace();
}
}
}
if (createdExecutor) {
ExecutorService e = (ExecutorService) executor;
e.shutdown();
if (awaitTermination) {
try {
e.awaitTermination(Integer.MAX_VALUE, TimeUnit.SECONDS);
} catch (InterruptedException e1) {
// 注意异常时的中断处理
Thread.currentThread().interrupt();
}
}
}
disposed = true;
}
protected abstract void dispose0() throws Exception;
@Override
public IoHandler getHandler() {
return this.handler;
}
@Override
public void setHandler(IoHandler handler) {
if (handler == null) {
throw new IllegalArgumentException("handler cannot be null");
}
// TODO isActive: when service is active, cannot be set handler
if(isActive()){
throw new IllegalStateException("when service is active, cannot be set handler");
}
this.handler = handler;
}
}AbstractIoAcceptor:
package org.apache.mina.core.rewrite.service;
import java.io.IOException;
import java.net.SocketAddress;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.Executor;
public abstract class AbstractIoAcceptor extends AbstractIoService implements
IoAcceptor {
private final List<SocketAddress> defaultLocalAddresses = new ArrayList<SocketAddress>();
private final List<SocketAddress> unmodifiableDeffaultLocalAddresses = Collections
.unmodifiableList(defaultLocalAddresses);
private final Set<SocketAddress> boundAddresses = new HashSet<SocketAddress>();
private boolean disconnectOnUnbind = true;
/** 这里不是很明白，为什么要用protected 而 不是private */
protected final Object bindLock = new Object();
/**
* 注意这个构造方法是一定要写的，否则编译不通过：抽象类继承时候，构造方法都要写,而且必须包含super
*
* @param param
* sessionConfig
* @param executor
*/
protected AbstractIoAcceptor(Object param, Executor executor) {
super(param, executor);
defaultLocalAddresses.add(null);
}
@Override
public SocketAddress getLocalAddress() {
Set<SocketAddress> localAddresses = getLocalAddresses();
if (localAddresses.isEmpty()) {
return null;
}
return localAddresses.iterator().next();
}
@Override
public final Set<SocketAddress> getLocalAddresses() {
Set<SocketAddress> localAddresses = new HashSet<SocketAddress>();
synchronized (boundAddresses) {
localAddresses.addAll(boundAddresses);
}
return localAddresses;
}
@Override
public void bind(SocketAddress localAddress) throws IOException {
// TODO Auto-generated method stub
}
@Override
public void bind(Iterable<? extends SocketAddress> localAddresses)
throws IOException {
// TODO isDisposing()
if (localAddresses == null) {
throw new IllegalArgumentException("localAddresses");
}
List<SocketAddress> localAddressesCopy = new ArrayList<SocketAddress>();
for (SocketAddress a : localAddresses) {
// TODO check address type
localAddressesCopy.add(a);
}
if (localAddressesCopy.isEmpty()) {
throw new IllegalArgumentException("localAddresses is empty");
}
boolean active = false;
synchronized (bindLock) {
synchronized (boundAddresses) {
if (boundAddresses.isEmpty()) {
active = true;
}
}
}
/** implement in abstractIoService */
if (getHandler() == null) {
throw new IllegalArgumentException("handler is not set");
}
try {
Set<SocketAddress> addresses = bindInternal(localAddressesCopy);
synchronized (boundAddresses) {
boundAddresses.addAll(addresses);
}
} catch (IOException e) {
throw e;
} catch (RuntimeException e) {
throw e;
} catch (Throwable e) {
throw new RuntimeException("Filed ti bind");
}
if(active){
//do sth
}
}
protected abstract Set<SocketAddress> bindInternal(
List<? extends SocketAddress> localAddress) throws Exception;
@Override
public void unbind(SocketAddress localAddress) {
// TODO Auto-generated method stub
}
}polling:
package org.apache.mina.core.rewrite.polling;
import java.net.SocketAddress;
import java.nio.channels.ServerSocketChannel;
import java.util.List;
import java.util.Set;
import java.util.concurrent.Executor;
import java.util.concurrent.Semaphore;
import java.util.concurrent.atomic.AtomicReference;
import org.apache.mina.core.rewrite.service.AbstractIoAcceptor;
public abstract class AbstractPollingIoAcceptor extends AbstractIoAcceptor {
private final Semaphore lock = new Semaphore(1);
private volatile boolean selectable;
private AtomicReference<Acceptor> acceptorRef = new AtomicReference<Acceptor>();
/**
* define the num of sockets that can wait to be accepted.
*/
protected int backlog = 50;
/**
* 一样的，这个构造方法也要写
*
* @param param
* @param executor
*/
protected AbstractPollingIoAcceptor(Object param, Executor executor) {
super(param, executor);
// TODO Auto-generated constructor stub
}
/**
* init the polling system. will be called at construction time
*
* @throws Exception
*/
protected abstract void init() throws Exception;
protected abstract void destory() throws Exception;
protected abstract int select() throws Exception;
/**这里有点儿变动*/
protected abstract ServerSocketChannel open(SocketAddress localAddress) throws Exception;
@Override
protected Set<SocketAddress> bindInternal(
List<? extends SocketAddress> localAddress) throws Exception {
// ...
try {
lock.acquire();
Thread.sleep(10);
} finally {
lock.release();
}
// ...
return null;
}
/**
* this class is called by startupAcceptor() method it's a thread accepting
* incoming connections from client
*
* @author ChenHui
*
*/
private class Acceptor implements Runnable {
@Override
public void run() {
assert (acceptorRef.get() == this);
int nHandles = 0;
lock.release();
while (selectable) {
try {
int selected = select();
// nHandles+=registerHandles();
if (nHandles == 0) {
acceptorRef.set(null);
// ...
}
} catch (Exception e) {
}
}
}
}
}好了最后看NioSoeketAcceptor:
package org.apache.mina.rewrite.transport.socket.nio;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.net.SocketAddress;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.util.concurrent.Executor;
import org.apache.mina.core.rewrite.polling.AbstractPollingIoAcceptor;
import org.apache.mina.rewrite.transport.socket.SocketAcceptor;
public final class NioSocketAcceptor extends AbstractPollingIoAcceptor
implements SocketAcceptor {
private volatile Selector selector;
protected NioSocketAcceptor(Object param, Executor executor) {
super(param, executor);
// TODO Auto-generated constructor stub
}
@Override
public int getManagedSessionCount() {
// TODO Auto-generated method stub
return 0;
}
/**
* 这个方法继承自AbstractIoAcceptor
*
* The type NioSocketAcceptor must implement the inherited abstract method
* SocketAcceptor.getLocalAddress() to override
* AbstractIoAcceptor.getLocalAddress()
*/
@Override
public InetSocketAddress getLocalAddress() {
// TODO Auto-generated method stub
return null;
}
@Override
public void setDefaultLocalAddress(InetSocketAddress localAddress) {
// TODO Auto-generated method stub
}
@Override
public boolean isReuseAddress() {
// TODO Auto-generated method stub
return false;
}
@Override
protected void init() throws Exception {
selector = Selector.open();
}
@Override
protected void destory() throws Exception {
if (selector != null) {
selector.close();
}
}
@Override
protected int select() throws Exception {
return selector.select();
}
@Override
protected void dispose0() throws Exception {
// TODO Auto-generated method stub
}
protected ServerSocketChannel open(SocketAddress localAddress)
throws Exception {
ServerSocketChannel channel =ServerSocketChannel.open();
boolean success=false;
try{
channel.configureBlocking(false);
ServerSocket socket=channel.socket();
socket.setReuseAddress(isReuseAddress());
socket.bind(localAddress);
channel.register(selector, SelectionKey.OP_ACCEPT);
success=true;
}finally{
if(!success){
//close(channel);
}
}
return channel;
}
@Override
public boolean isActive() {
// TODO Auto-generated method stub
return false;
}
}------------------------------------------------------
到此为止将连接部分都写完了，在连接部分还有些零碎的东西，比如handler、polling，这些都只是稍稍提了一下，具体后面会在介绍其他部分是肯定还会碰上，我还是想把重心放在最主要的部分去写，下一篇应该要写到session了。
阅读笔记（五）—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获得的：
ConnectFuture future = connector.connect(new InetSocketAddress(
HOST, PORT));// 创建连接
future.awaitUninterruptibly();// 等待连接创建完成
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。看具体实现之前，我们先看看这个类关联了哪些比较重要的引用：
private final IoHandler handler;
protected IoSessionConfig config;
private final IoService service;
private static final AttributeKey READY_READ_FUTURES_KEY = new AttributeKey(AbstractIoSession.class,
"readyReadFutures");
private static final IoFutureListener<CloseFuture> SCHEDULED_COUNTER_RESETTER;
private static final WriteRequest CLOSE_REQUEST = new DefaultWriteRequest(new Object());
private IoSessionAttributeMap attributes;
private WriteRequestQueue writeRequestQueue;
private WriteRequest currentWriteRequest;
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也考虑进去。
public AttributeKey(Class<?> source, String name) {
this.name = source.getName() + '.' + name + '@' + Integer.toHexString(this.hashCode());
}现在我们可以看AbstractIoSession了。主要看读写操作，其他操作都是统计和配置稍稍看过即可：
public final ReadFuture read() {
if (!getConfig().isUseReadOperation()) {
throw new IllegalStateException("useReadOperation is not enabled.");
}
Queue<ReadFuture> readyReadFutures = getReadyReadFutures();
ReadFuture future;
synchronized (readyReadFutures) {
future = readyReadFutures.poll();
if (future != null) {
if (future.isClosed()) {
// Let other readers get notified.
readyReadFutures.offer(future);
}
} else {
future = new DefaultReadFuture(this);
getWaitingReadFutures().offer(future);
}
}
return future;
}采用队列进行读取，这里只用了Queue，没有用concurrent中的那些同步队列，而是用了synchronized关键字来处理同步，主要是我们要明白这里要同步的不是队列里的内容，而是读的这个过程，session都是独立的，所以一个session内一个队列无论怎么抢还是能排除顺序的。所以对于读操作来说，主要是要保证在读一条的时候，不能有其他线程再读。
下面是写操作：
public WriteFuture write(Object message, SocketAddress remoteAddress) {
if (message == null) {
throw new IllegalArgumentException("Trying to write a null message : not allowed");
}
// We can't send a message to a connected session if we don't have
// the remote address
if (!getTransportMetadata().isConnectionless() && (remoteAddress != null)) {
throw new UnsupportedOperationException();
}
// If the session has been closed or is closing, we can't either
// send a message to the remote side. We generate a future
// containing an exception.
if (isClosing() || !isConnected()) {
WriteFuture future = new DefaultWriteFuture(this);
WriteRequest request = new DefaultWriteRequest(message, future, remoteAddress);
WriteException writeException = new WriteToClosedSessionException(request);
future.setException(writeException);
return future;
}
FileChannel openedFileChannel = null;
// TODO: remove this code as soon as we use InputStream
// instead of Object for the message.
try {
if ((message instanceof IoBuffer) && !((IoBuffer) message).hasRemaining()) {
// Nothing to write : probably an error in the user code
throw new IllegalArgumentException("message is empty. Forgot to call flip()?");
} else if (message instanceof FileChannel) {
FileChannel fileChannel = (FileChannel) message;
message = new DefaultFileRegion(fileChannel, 0, fileChannel.size());
} else if (message instanceof File) {
File file = (File) message;
openedFileChannel = new FileInputStream(file).getChannel();
message = new FilenameFileRegion(file, openedFileChannel, 0, openedFileChannel.size());
}
} catch (IOException e) {
ExceptionMonitor.getInstance().exceptionCaught(e);
return DefaultWriteFuture.newNotWrittenFuture(this, e);
}
// Now, we can write the message. First, create a future
WriteFuture writeFuture = new DefaultWriteFuture(this);
WriteRequest writeRequest = new DefaultWriteRequest(message, writeFuture, remoteAddress);
// Then, get the chain and inject the WriteRequest into it
IoFilterChain filterChain = getFilterChain();
filterChain.fireFilterWrite(writeRequest);
// TODO : This is not our business ! The caller has created a
// FileChannel,
// he has to close it !
if (openedFileChannel != null) {
// If we opened a FileChannel, it needs to be closed when the write
// has completed
final FileChannel finalChannel = openedFileChannel;
writeFuture.addListener(new IoFutureListener<WriteFuture>() {
public void operationComplete(WriteFuture future) {
try {
finalChannel.close();
} catch (IOException e) {
ExceptionMonitor.getInstance().exceptionCaught(e);
}
}
});
}
// Return the WriteFuture.
return writeFuture;
}这里面有个instanceof FileChannel和File是不是感到有点儿奇怪，mina不是写出去的是IoBuffer么，怎么现在又可以写文件了。Mina作为一个封装好的框架，自然可以直接做文件的传输，这里面会有相应的handler来处理这些业务。
Session部分最主要的就是了解他的生命周期以及相关联的那些引用，这里我们可以看到与读写最密切的就是Future了，所以这部分，就是我下篇会写的主题。
阅读笔记（六）—Mina异步IO的实现IoFuture
IoFuture是和IoSession紧密相连的一个类，在官网上并没有对它的描述，因为它一般不会显示的拿出来用，权当是一个工具类被session所使用。当然在作用上，这个系列可并不简单，我们先看源码的注释对它的描述：
IoFuture represents the completion of an asynchronous I/O operation on an IoSession.
这个类是提供异步操作的一个工具，所以在读源码之前，必须对异步IO操作有所了解，然后我们才可以顺着这条路往下走。关于异步IO的介绍可以看：《同步、异步、阻塞、非阻塞》
IoFuture通过IoFutureListener被IoSession绑定，它的实现都放在org.apache.mina.core.future下。在IoFuture的实现中，分别提供了读、写、连接、关闭的future，通过这四个future来实现异步操作。异步操作很重要的一部分就是对线程的控制，所以在IoFuture这个接口中，我们能很清楚的理清这几个方法：await、join。当然还有notify，但是notify没有必要写在接口中，它可以在程序里直接使用。
这个系列的类设计的很规整，从上图的结构就能看出来，图中省略了write和connection的图，它们分别和read与close一致。由于这个类的操作没有那么复杂，继承关系也没有那么多层，所以这里面都没有用到Abstract的类来做具体实现。
下面我们来看这里面最核心的一个类DefaultIoFuture。这个类实现IoFuture接口，主要实现了对await和join的操作，以及处理死锁的操作。我们先看这个类关联到的成员变量，都比较简单：
/** A number of seconds to wait between two deadlock controls ( 5 seconds ) */
private static final long DEAD_LOCK_CHECK_INTERVAL = 5000L;
/** The associated session */
private final IoSession session;
/** A lock used by the wait() method */
private final Object lock;
private IoFutureListener<?> firstListener;
private List<IoFutureListener<?>> otherListeners;
private Object result;
private boolean ready;
private int waiters;在我看来，读源码的目的，一是为了理清框架的设计逻辑，理清结构，学习这些关联关系；二是为了学习处理细节，比如死锁的处理、线程安全的处理。在这个类中，我们将看到mina作者是如何处理死锁的问题的。
我们先看await操作，await主要是为了等待异步操作的完成，然后通知相关的listener。我们先看一个简单的await操作和验证死锁的操作：
public IoFuture await() throws InterruptedException {
synchronized (lock) {
while (!ready) {
waiters++;
try {
lock.wait(DEAD_LOCK_CHECK_INTERVAL);
} finally {
waiters--;
if (!ready) {
checkDeadLock();
}
}
}
}
return this;
}我们应该要注意下在await方法中的wait操作，这里讲些题外话，面试中常问的wait和sleep的区别。wait的操作其实很规范，必须写在synchronized块内，必须由其他线程来notify，同时wait释放锁，不占资源。而sleep占着cup的资源区睡眠，时间没到不能被唤醒，只能通过中断来打断。在这个await方法中，wait了check dead lock的时间，并且设置了计数器waiters。这个waiters在setValue方法中被运用到，在setValue中：
public void setValue(Object newValue) {
synchronized (lock) {
// Allow only once.
if (ready) {
return;
}
result = newValue;
ready = true;
if (waiters > 0) {
lock.notifyAll();
}
}
notifyListeners();
}异步操作是没有一个固定的顺序，谁先做好谁就返回，所以一旦有异步任务完成了操作，就会notify所有的等待，让接下来先抢到的线程再执行。在DefaultIoFuture这个类中，我觉得最重要的到不是连接或者读写，而是上面提到的setValue和getValue，因为在后续的继承关系中，会不断的用到这两个方法。不仅在后续的继承关系中，这两个方法真正在传递值得操作是发生在IoService中，不要忘了虽然session很重要，但真正起连接作用的还是service。
然后我们再看下上面提到的check dead lock的方法，在抢占中只有读、写和连接会产生死锁的情况：
private void checkDeadLock() {
if (!(this instanceof CloseFuture || this instanceof WriteFuture || this instanceof ReadFuture || this instanceof ConnectFuture)) {
return;
}
StackTraceElement[] stackTrace = Thread.currentThread().getStackTrace();
// Simple and quick check.
for (StackTraceElement s : stackTrace) {
if (AbstractPollingIoProcessor.class.getName().equals(s.getClassName())) {
IllegalStateException e = new IllegalStateException("t");
e.getStackTrace();
throw new IllegalStateException("DEAD LOCK: " + IoFuture.class.getSimpleName()
+ ".await() was invoked from an I/O processor thread. " + "Please use "
+ IoFutureListener.class.getSimpleName() + " or configure a proper thread model alternatively.");
}
}
// And then more precisely.
for (StackTraceElement s : stackTrace) {
try {
Class<?> cls = DefaultIoFuture.class.getClassLoader().loadClass(s.getClassName());
if (IoProcessor.class.isAssignableFrom(cls)) {
throw new IllegalStateException("DEAD LOCK: " + IoFuture.class.getSimpleName()
+ ".await() was invoked from an I/O processor thread. " + "Please use "
+ IoFutureListener.class.getSimpleName()
+ " or configure a proper thread model alternatively.");
}
} catch (Exception cnfe) {
// Ignore
}
}
}在追踪堆栈信息时，这里采用了两种check方式，简单和精确。在简单检测中，只是对比了类名，也就是对当前类有效，是一个一对一的比较。而在精确的检测中，采用isAssignableFrom方法来分别和其父类和本类进行比较。如果有死锁，就抛异常。另外join方法被废弃，由awaitUninterruptibly代替，虽然叫join，其实还是一种wait操作，等到一定时间将flag转变一下。
下面我们看ReadFuture接口，这个接口直接继承IoFuture接口，并添加了相关的写操作。接口由DefaultReadFuture实现。在使用中，ReadFuture在IoSession中read方法中被使用，也可以说，在session层，直接读写的是future，我们再看下AbstractIoSession中的read代码：
public final ReadFuture read() {
if (!getConfig().isUseReadOperation()) {
throw new IllegalStateException("useReadOperation is not enabled.");
}
Queue<ReadFuture> readyReadFutures = getReadyReadFutures();
ReadFuture future;
synchronized (readyReadFutures) {
future = readyReadFutures.poll();
if (future != null) {
if (future.isClosed()) {
// Let other readers get notified.
readyReadFutures.offer(future);
}
} else {
future = new DefaultReadFuture(this);
getWaitingReadFutures().offer(future);
}
}
return future;
}每次都是从队列中拿出一个future，同理，每次写也是往队列里写入一个future。在DefaultReadFuture中的方法都比较简单，这里就不贴出来了。另外WriteFuture和DefaultWriteFuture和read类似，也不再赘述。
最后我们看看ConnectFuture，我们常常写这么一段话来拿到session：
ConnectFuture future = connector.connect(new InetSocketAddress(
HOST, PORT));// 创建连接
future.awaitUninterruptibly();// 等待连接创建完成
session = future.getSession();// 获得session提一点，在多态的使用上，ConnectFuture完全可以换成IoFuture，这对后面的代码没有一点儿影响，getSession本身就是继承自IoFuture的。ConnectFuture接口由DefaultConnectFuture来具体实现，由于继承了DefaultIoFuture，所以这里面用到最多的就是DefaultIoFuture中的setValue和getValue方法，上面我们也特别强调了这两个方法的重要性，通过对result（setValue）的传递实现了对session、exception等状态的传递。
稍微总结一下future对异步的贡献，官方对future的描述就是处理了异步操作，从源码中我们很明显的可以看到future是通过await和notify来控制操作的连续性，通过死锁检测来做wait时的保障，上层（session）通过队列来缓冲各种任务，然后通过竞争，谁抢到了线程，谁就执行。Future不难，组织结构也很清楚，我觉得看这节的源代码最主要的还是要做好两点，第一是搞懂什么是异步，第二是要明白future为异步贡献了什么。
下一篇就要讲mina中最庞大的filter chain了，这是mina中代码最多，也是最具特色的一部分。
阅读笔记（七）—Mina的拦截器FilterChain
Filter我们很熟悉，在Mina中，filter chain的用法也类似于Servlet的filters，这种拦截器的设计思想能够狠轻松的帮助我们实现对资源的统一处理。我们先大致连接下mina中的filter能给我们带来什么。
- LoggingFilter logs all events and requests.
- ProtocolCodecFilter converts an incoming ByteBuffer into message POJO and vice versa.
- CompressionFilter compresses all data.
- SSLFilter adds SSL - TLS - StartTLS support.
- and many more!
当然这中间最实用，而且源码篇幅最多的就是对codec的拦截器，这部分的应用就是可以实现自定义的编码器和解码器，并附上自定义的协议来进行通信。这部分的应用可以看：《Mina实现自定义协议的通信》
在Mina源码中，对filter的描述主要分两部分，org.apache.mina.core.filterchain以及org.apache.mina.filter.*这两部分。在核心包里主要定义了规则，然后再filter包中进行具体的实现。在讲filter的时候我们需要分清楚filter和filter chain的区别，在chain中加载的是filter，所以filter的生命周期也容易理解，加载到chain中的为active，否则就在生命周期之外。
上图仅org.apache.mina.core.filterchain内的部分关联（ReferenceCountingFilter除外），不过弄明白图上画的这些类的作用和实现原理就基本能明白整个filter的内容，后面的各种filter都是通过IoFilterAdapter派生而来。这也是我们很熟悉的适配器模式的应用。
我们先来看IoFilter，IoFilter加载到chain需经历如下过程：
1、 调用init方法，被ReferenceCountingFilter初始化（此时未加载到chain）
2、 调用onPreAdd方法，告诉filter，将要被加载到chain中去
3、 当filter被加载到chain中，filter chain开始正式起作用
4、 调用onPostAdd方法，做一些加载完成的后处理。
这样的处理方式非常的常用，在Android中的asynctask好像经常会有类似的用法，执行前、执行中、执行后分别处理。其实也不用说那么远，mina中，在handler中处理session的连接也分session create，session open等等。在IoFilter中有一个内部接口NextFilter，用来引用在chain中的下一个filter（这个next filter会在当前的filter中被用到，所以要这么设计）。并且，在next接口中，除了上面提到的加载到chain的方法没有之外，其他的都与外部类一致。
我们再来看IoFilter的实现类DefaultIoFilter。我们要提一下这个类，之前我们介绍的无论是service还是session都没有让开发者去重写或者继承，而这里不一样，由于这个类是一个适配器的设计模式，因为必然需可以由有大量的实现方法去充实它。所以在mina官网的user guide上给出了如何去继承这个adapter，具体可见：http://mina.apache.org/mina-project/userguide/ch5-filters/ch5-filters.html
我们可以注意到，DefaultIoFilter的方法里几乎都为空，什么都没有做，最多来一句：
public void sessionCreated(NextFilter nextFilter, IoSession session) throws Exception {
nextFilter.sessionCreated(session);
}这样做就一个原因摆明了让其他类去做具体的实现，同时也可以让开发者自己去实现。注意，那些未实现的方法都是跟filter生命周期有关的，而receive和send这样的方法都会由next filter来实现。
接下来我们看IoFilterChain，这个容器装载了我们添加的各种拦截器，每个session都有一个拦截器，这是一对一的关系，当然不要忘记了，拦截器的最后一步是IoHandle，这个我们在之前就说过。我们还是稍稍关注一下这个接口中的内部类Entry，这是一个name-filter对，存放在chain中，好比一个map，给filter一个名字。这节最主要的就是DefaultIoFilterChain，他是整个chain能实现起来的中心，我们主要是看mina是如何给这个filter排序的。
我们截一段代码，Entry的构造方法：
private EntryImpl(EntryImpl prevEntry, EntryImpl nextEntry, String name, IoFilter filter) {
if (filter == null) {
throw new IllegalArgumentException("filter");
}
if (name == null) {
throw new IllegalArgumentException("name");
}
this.prevEntry = prevEntry;
this.nextEntry = nextEntry;
this.name = name;
this.filter = filter;
this.nextFilter = new NextFilter() {
public void sessionCreated(IoSession session) {
Entry nextEntry = EntryImpl.this.nextEntry;
callNextSessionCreated(nextEntry, session);
}主要看构造方法里的这几个参数，后面的两个我们在Entry接口的时候提过，name-filter对，所以这两个必须要，否则抛异常。那前面两个参数从命名中我们就能看见顺序了，继续往下看，我们现在来看DefaultIoFilterChain（Entry是DefaultIoFilterChain的一个内部类）的构造方法：
private final Map<String, Entry> name2entry = new ConcurrentHashMap<String, Entry>();
/** The chain head */
private final EntryImpl head;
/** The chain tail */
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());
tail = new EntryImpl(head, null, "tail", new TailFilter());
head.nextEntry = tail;
}很明显，这个chain在初始化的时候只有这两个entry，一头一尾，并且这个一头一尾属于唯一一个和DefaultIoFilterChain绑定的session。这个是不是很像链表。所以你就很容易理解AddFirst和AddLast方法是怎么实现的了：
public synchronized void addFirst(String name, IoFilter filter) {
checkAddable(name);
register(head, name, filter);
}
private void checkAddable(String name) {
if (name2entry.containsKey(name)) {
throw new IllegalArgumentException("Other filter is using the same name '" + name + "'");
}
}
private void register(EntryImpl prevEntry, String name, IoFilter filter) {
EntryImpl newEntry = new EntryImpl(prevEntry, prevEntry.nextEntry, name, filter);
try {
filter.onPreAdd(this, name, newEntry.getNextFilter());
} catch (Exception e) {
throw new IoFilterLifeCycleException("onPreAdd(): " + name + ':' + filter + " in " + getSession(), e);
}
prevEntry.nextEntry.prevEntry = newEntry;
prevEntry.nextEntry = newEntry;
name2entry.put(name, newEntry);
try {
filter.onPostAdd(this, name, newEntry.getNextFilter());
} catch (Exception e) {
deregister0(newEntry);
throw new IoFilterLifeCycleException("onPostAdd(): " + name + ':' + filter + " in " + getSession(), e);
}
}看了代码是不是就一目了然了，这不就是数据结构里我们常用的双向链表。所以数据结构不是白学的，软件设计上经常会用到。这个类里基本就是对链表的操作，只要对双向链表的指针比较清楚的，读懂也应该没什么问题。
提一下IoFilterChainBuilder和DefaultIoFilterChainBuilder。这又是一组继承关系，DefaultIoFilterChainBuilder的实现和DefaultIoFilterChain很像，几乎一样。这也是能够用来做拦截器链的，但是它和DefaultIoFilterChain还是有不同：
l DefaultIoFilterChainBuilder不管理IoFuture的生命周期
l DefaultIoFilterChainBuilder不会影响已经创建好的session
我们来看一下IoService中的getFilterChain：
/**
* A shortcut for <tt>( ( DefaultIoFilterChainBuilder ) </tt>{@link #getFilterChainBuilder()}<tt> )</tt>.
* Please note that the returned object is not a <b>real</b> {@link IoFilterChain}
* but a {@link DefaultIoFilterChainBuilder}. Modifying the returned builder
* won't affect the existing {@link IoSession}s at all, because
* {@link IoFilterChainBuilder}s affect only newly created {@link IoSession}s.
*
* @throws IllegalStateException if the current {@link IoFilterChainBuilder} is
* not a {@link DefaultIoFilterChainBuilder}
*/
DefaultIoFilterChainBuilder getFilterChain();得到的是DefaultIoFilterChainBuilder而不是IoFilterChain。那如果你要IoFilterChain就需要用下面的方法来built：
void setFilterChainBuilder(IoFilterChainBuilder builder);不过我看官网上也是用默认的DefaultIoFilterChainBuilder来生成chain。可能后面只要管理到handler就行了，而且一般的应用中，一条通道中也只用一个session。
Filter只讲到这里了，至于那十几个常用的我觉得没必要写了，他们总有那么一个类要去继承IoFilterAdapter，然后再不断的派生开来，所以如果你要用到哪个就再自己读就行了。我觉得这部分还是注重应用为主，至少我不关心这些实现，它和通信的关系就不是很大了。
这系列的东西快结束了，明天再写点儿杂七杂八零碎的东西，然后就结束mina这一系列的文章。感谢各位的支持了。在最后我会写一个索引页，把关于mina的文章都串起来方便大家阅读。
标签：session,Mina,private,final,解读,源码,mina,new,public 来源： https://www.cnblogs.com/hanease/p/16200230.html
本站声明： 1. iCode9 技术分享网（下文简称本站）提供的所有内容，仅供技术学习、探讨和分享； 2. 关于本站的所有留言、评论、转载及引用，纯属内容发起人的个人观点，与本站观点和立场无关； 3. 关于本站的所有言论和文字，纯属内容发起人的个人观点，与本站观点和立场无关； 4. 本站文章均是网友提供，不完全保证技术分享内容的完整性、准确性、时效性、风险性和版权归属；如您发现该文章侵犯了您的权益，可联系我们第一时间进行删除； 5. 本站为非盈利性的个人网站，所有内容不会用来进行牟利，也不会利用任何形式的广告来间接获益，纯粹是为了广大技术爱好者提供技术内容和技术思想的分享性交流网站。