mina源码分析——bind
关于Mina
mina是开源的NIO框架，其project地址：
http://mina.apache.org/mina-project/features.html
想快速了解mina就看user guide：
http://mina.apache.org/mina-project/userguide/user-guide-toc.html
mina给我的感觉：干净、利落的抽象，非常容易上手，使用mina你只需要写不需要超过10行code就可以搭建一个TCP服务器，就像mina自身带的例子：
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.charset.Charset;
import org.apache.mina.core.service.IoAcceptor;
import org.apache.mina.core.session.IdleStatus;
import org.apache.mina.example.gettingstarted.timeserver.TimeServerHandler;
import org.apache.mina.filter.codec.ProtocolCodecFilter;
import org.apache.mina.filter.codec.textline.TextLineCodecFactory;
import org.apache.mina.filter.logging.LoggingFilter;
import org.apache.mina.transport.socket.nio.NioSocketAcceptor;
public class MinaTimeServer {
/**
* @param args
*/
private static final int PORT = 9123;
public static void main(String[] args) throws IOException {
IoAcceptor acceptor = new NioSocketAcceptor();
acceptor.getFilterChain().addLast( "logger", new LoggingFilter());
acceptor.getFilterChain().addLast( "codec", new ProtocolCodecFilter(new TextLineCodecFactory(Charset.forName( "UTF-8" ))));
acceptor.setHandler(new TimeServerHandler());
acceptor.getSessionConfig().setReadBufferSize(2048);
acceptor.getSessionConfig().setIdleTime( IdleStatus.BOTH_IDLE, 10);
acceptor.bind(new InetSocketAddress(PORT) );
}
}
Mina源码分析之bind
mina好用只是它把复杂很好的进行了分解和隐藏，要想真正掌握它，还是要深入了解源码，所以从bind方法开始,bind方法是interface IoAcceptor声明的方法，先看看它的javadoc，了解bind方法的职责：
/**
* Binds to the default local address(es) and start to accept incoming
* connections.
*
* @throws IOException if failed to bind
*/
void bind() throws IOException;
/**
* Binds to the specified local address and start to accept incoming
* connections.
*
* @param localAddress The SocketAddress to bind to
*
* @throws IOException if failed to bind
*/
void bind(SocketAddress localAddress) throws IOException;
/**
* Binds to the specified local addresses and start to accept incoming
* connections. If no address is given, bind on the default local address.
*
* @param firstLocalAddresses The first address to bind to
* @param addresses The SocketAddresses to bind to
*
* @throws IOException if failed to bind
*/
void bind(SocketAddress firstLocalAddress, SocketAddress... addresses) throws IOException;
/**
* Binds to the specified local addresses and start to accept incoming
* connections. If no address is given, bind on the default local address.
*
* @param addresses The SocketAddresses to bind to
*
* @throws IOException if failed to bind
*/
void bind(SocketAddress... addresses) throws IOException;
/**
* Binds to the specified local addresses and start to accept incoming
* connections.
*
* @throws IOException if failed to bind
*/
void bind(Iterable<? extends SocketAddress> localAddresses) throws IOException;bind方法职责
1.绑定到本地ip地址（如果没有指定就绑定到默认ip）
2.开始接受外面进来的请求。
通俗地说，一个Tcp的或则UDP的IO Server调用了bind方法后，就可以开始干活了，就可以接受请求了。顺便说一句，mina对IO的两端（Server，Client）都有很好的抽象和封装，IO Server由IoAcceptor代表，IO Client由IoConnector代表，而它们又都是IoService的直接子接口。关于mina的整体架构还是去看user guide，非常清晰。
明白了bind方法的职责，不妨先不要往下看，而是结合java NIO api 思考一下其大概的实现细节，然后再去翻code来验证你的推测。
对于一个基于NIO的server来说，要想能够开始接受外面的请求，有几件事必须做：
1.open一个SelectableChannel，SelectableChannel代表了可以支持非阻塞IO操作的channel
2.open一个Selector，Selector通过select方法来监控所有可以进行IO操作的SelectableChannel
3.完成SelectableChannel在Selector上的注册，这样Selector方可监控他。而类SelectionKey则代表一个成功的注册。
当然对于一个框架来说，肯定会做的更多，但以上3点是必须的，那接下来就去翻code吧
首先看bind方法所在的类层次：
从类图中可以看出，真正处理bind逻辑的是bindInternal方法，该方法在AbstractIoAcceptor抽象类中声明，在AbstractPollingIoAcceptor类中实现。正如javadoc的说明，AbstractPollingIoAcceptor是A base class for implementing transport using a polling strategy.
bindInternal方法
下面是bindInternal方法的code：
protected final Set<SocketAddress> bindInternal(List<? extends SocketAddress> localAddresses) throws Exception {
// Create a bind request as a Future operation. When the selector
// have handled the registration, it will signal this future.
AcceptorOperationFuture request = new AcceptorOperationFuture(localAddresses);
// adds the Registration request to the queue for the Workers
// to handle
registerQueue.add(request);
// creates the Acceptor instance and has the local
// executor kick it off.
startupAcceptor();
// As we just started the acceptor, we have to unblock the select()
// in order to process the bind request we just have added to the
// registerQueue.
try {
lock.acquire();
// Wait a bit to give a chance to the Acceptor thread to do the select()
Thread.sleep(10);
wakeup();
} finally {
lock.release();
}
// Now, we wait until this request is completed.
request.awaitUninterruptibly();
if (request.getException() != null) {
throw request.getException();
}
// Update the local addresses.
// setLocalAddresses() shouldn't be called from the worker thread
// because of deadlock.
Set<SocketAddress> newLocalAddresses = new HashSet<SocketAddress>();
for (H handle : boundHandles.values()) {
newLocalAddresses.add(localAddress(handle));
}
return newLocalAddresses;
}
1.创建一个bind请求：
AcceptorOperationFuture request = new AcceptorOperationFuture(localAddresses);
当前你可以把AcceptorOperationFuture简单理解为支持异步处理和事件通知机制的一个request，后续再详细分析它。
2.把请求加到队列中：
registerQueue.add(request);
registerQueue保存所有新注册的请求，这些请求在registerHandles方法中被消费掉。registerHandles方法又被Acceptor.run方法调用。新注册成功的连接都被保存在：
private final Map<SocketAddress, H> boundHandles = Collections.synchronizedMap(new HashMap<SocketAddress, H>());
3.创建Acceptor类，并且异步执行该类的run方法。
startupAcceptor();
Acceptor是一个内部类，实现了Runnable方法。它负责接收和处理从客户端进来的bind和unbind请求。下面会重点分析这个类。
4.等待10毫秒，然后调用wakeup方法来确保selector的select方法被唤醒。
try {
lock.acquire();
// Wait a bit to give a chance to the Acceptor thread to do the select()
Thread.sleep(10);
wakeup();
} finally {
lock.release();
}
lock是一个信号量：
private final Semaphore lock = new Semaphore(1);由于可用值是1，因此相当于排他锁。
这里要重点解释一下为什么要sleep这10毫秒：要搞清楚这点，首先要看Acceptor.run方法的实现（下面介绍），该方法调用了
int selected = select();
而select方法是abstract的：
/**
* Check for acceptable connections, interrupt when at least a server is ready for accepting.
* All the ready server socket descriptors need to be returned by {@link #selectedHandles()}
* @return The number of sockets having got incoming client
* @throws Exception any exception thrown by the underlying systems calls
*/
protected abstract int select() throws Exception;大家能够猜到其子类：NioSocketAcceptor的实现就是调用Selector.select方法：
@Override
protected int select() throws Exception {
return selector.select();
}而select方法在没有io事件也没有调用selector.wakeup方法时是阻塞的，反之就会被唤醒。所以才有了上面sleep10毫秒后调用wakeup方法（也是abstract方法，子类NioSocketAcceptor实现就是调用Selector.wakeUp),来唤醒select。但由于Accecptor是在独立线程运行的，因而sleep一会以保证那个线程已经启动，而不会错过唤醒select.
5.然后就等待请求被处理完毕，并判断是否有异常，没有异常就把绑定的本地地址返回。
Acceptor
Acceptor是一个内部类，实现了Runnable方法。它负责接收和处理从客户端进来的bind和unbind请求
接下来重点看Acceptor的run方法：
public void run() {
assert (acceptorRef.get() == this);
int nHandles = 0;
// Release the lock
lock.release();
while (selectable) {
try {
// Detect if we have some keys ready to be processed
// The select() will be woke up if some new connection
// have occurred, or if the selector has been explicitly
// woke up
int selected = select();
// this actually sets the selector to OP_ACCEPT,
// and binds to the port on which this class will
// listen on
nHandles += registerHandles();
// 这里推出while循环的逻辑 省略。。。
if (selected > 0) {
// We have some connection request, let's process
// them here.
processHandles(selectedHandles());
}
// check to see if any cancellation request has been made.
nHandles -= unregisterHandles();
} catch (ClosedSelectorException cse) {
// If the selector has been closed, we can exit the loop
break;
} catch (Throwable e) {
ExceptionMonitor.getInstance().exceptionCaught(e);
try {
Thread.sleep(1000);
} catch (InterruptedException e1) {
ExceptionMonitor.getInstance().exceptionCaught(e1);
}
}
}
// Cleanup all the processors, and shutdown the acceptor.
// .....
}
}run方法重点是while循环：
1.首先调用select方法来看是否有新io请求进来（由于Acceptor只被bind和unbind调用，因此此时的请求只可能是bind请求和unbind请求）
2.调用registerHandles处理bind请求，该方法返回新处理的bind请求数目，如果＝0，则推出循环，Acceptor的处理结束。处理bind请求的大概过程是：
1）从registerQueue队列中拿出bind请求：AcceptorOperationFuture（之前放进去的）
2）从AcceptorOperationFuture中拿出要bind的SocketAddress（可能不止一个）；
3）对于每一个SocketAddress，调用open方法来建立ServerSocketChannel通道。open方法是abstract，其子类NioSocketAcceptor给出了实现，没错正如我们之前设想的那样，NioSocketAcceptor.open主要就是完成ServerSocketChannel的注册，源码如下：
@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;
}4）把第3）建立好的ServerSocketChannel放到boundHandles中。boundHandles是一个线程安全的map：
private final Map<SocketAddress, H> boundHandles = Collections.synchronizedMap(new HashMap<SocketAddress, H>());
5）返回新注册的size。
3. 如果select方法返回值>0，就调用processHandles(selectedHandles());来处理。处理过程大致如下：
1）从selector中拿出有io发生的ServerSocketChannel集合；这个过程是在子类NioSocketAcceptor .selectedHandles方法中发生的。
2）对于每一个ServerSocketChannel，调用accept方法来建立会话：IoSession,IoSession是mina中非常重要的概念，以后再分析。accept方法也是abstract的，其子类NioSocketAccepor给出了实现：
@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);
}3）初始化session：initSession(session, null, null);
4）保存session到IoProcessor
4.处理unbind的逻辑：
nHandles -= unregisterHandles();
前面说过Acceptor的用户是bind和unbind，所以run方法里自然会有处理unbind的逻辑，不了解这一点看到这段code可能会有些困惑。
unbind的逻辑就很简单了，大致过程如下：
1）从cancelQueue中取出unbind的请求：AcceptorOperationFuture
2）从AcceptorOperationFuture中取出SocketAddress对象（可能不止一个）
3）对于每一个SocketAddress,把它从boundHandles中删除
4）对于每一个SocketAddress,调用close方法来关闭它，close方法也是abstract方法，子类NioSocketAcceptor给出了实现：
@Override
protected void close(ServerSocketChannel handle) throws Exception {
SelectionKey key = handle.keyFor(selector);
if (key != null) {
key.cancel();
}
handle.close();
}5）调用AcceptorOperationFuture.setDone来做事件通知
AbstractPollingIoAcceptor.bindInternal
->startupAcceptor//启动Acceptor线程
-->Acceptor.run
--->registerHandles
主要逻辑：完成ServerSocketChannel在selector中的注册，感兴趣的事件是： SelectionKey.OP_ACCEPT，注册后ServerSocketChannel就可以在指定的ip和port上监听新连接。
1）从registerQueue中取出头元素，registerQueue保存待注册的请求
AcceptorOperationFuture future = registerQueue.poll();
1）创建ServerSocketChannel：ServerSocketChannel.open();
2）把ServerSocket绑定到指定的SocketAddress，SocketAddress来自
List<SocketAddress> localAddresses = future.getLocalAddresses();
3）把ServerSocketChannel注册到selector：channel.register(selector, SelectionKey.OP_ACCEPT);
4）返回channel并保存在boundHandles中
--->processHandles
主要逻辑：处理ServerSocketChannel上的新连接请求，把新连接请求构建新的IoSession并初始化，然后把IoSession实例交给IoProcessor来管理和处理io操作。
1)调用NioSocketAcceptor.accept方法返回NioSocketSession实例。构建NioSocketSession需要SocketChannel、IoProcessor等参数。
2）调用initSession方法
3）把当前session加到IoProcessor中：void add(S session)
--->unregisterHandles
主要逻辑：响应doUnbind方法，关闭ServerSocketChannel