一个IoSession的I/O事件是注册在一个Selector对象上，并且每个Processor线程只轮询一个Selector对象，即每一个链接只有一个线程处理I/O事件，这样能保证同一IoSession数据的有序性。
下面就从部分源码探究其中的原理，以NioAcceptor为例子：
public NioSocketAcceptor() {
super(new DefaultSocketSessionConfig(), NioProcessor.class);
((DefaultSocketSessionConfig) getSessionConfig()).init(this);
}
这里的NioProcessor.class就是Processor的具体类型。
protected AbstractPollingIoAcceptor(IoSessionConfig sessionConfig, Class<? extends IoProcessor<S>> processorClass) {
this(sessionConfig, null, new SimpleIoProcessorPool<S>(processorClass), true, null);
}
SimpleIoProcessorPool是Processor的线程池，使用NioProcessor创建具体的线程。
跳过Acceptor的初始化过程，当客户端请求建立链接，服务端Acceptor线程会执行以下代码:
private void processHandles(Iterator<H> handles) throws Exception {
while (handles.hasNext()) {
H handle = handles.next();
handles.remove();
// Associates a new created connection to a processor,
// and get back a session
S session = accept(processor, handle); //这里的processor是processor线程池
if (session == null) {
continue;
}
initSession(session, null, null);
// add the session to the SocketIoProcessor
session.getProcessor().add(session);
}
}
@Override
protected NioSession accept(IoProcessor<NioSession> processor, ServerSocketChannel handle) throws Exception {
SelectionKey key = null;
if (handle != null) {
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);
}
这里创建了NioSocketSession将Processor线程池与SocketChannel绑定在一起。然后通过 session.getProcessor().add(session)将会话注册到SimpleIoProcessorPool线程池中的一个Processor对象内部的Selector对象。
为什么这里的processor是线程池？还记得NioSocketAcceptor的构造函数中的SimpleIoProcessorPool，processor就是它的实例。
看以下NioSocketSession的getProcessor()方法:
public IoProcessor<NioSession> getProcessor() {
return processor;
}
返回的就是与它关联的SimpleIoProcessorPool线程池对象.再看SimpleIoProcessorPool的addI()方法：
public final void add(S session) {
getProcessor(session).add(session);
}
private IoProcessor<S> getProcessor(S session) {
IoProcessor<S> processor = (IoProcessor<S>) session.getAttribute(PROCESSOR);
if (processor == null) {
if (disposed || disposing) {
throw new IllegalStateException("A disposed processor cannot be accessed.");
}
processor = pool[Math.abs((int) session.getId()) % pool.length];
if (processor == null) {
throw new IllegalStateException("A disposed processor cannot be accessed.");
}
session.setAttributeIfAbsent(PROCESSOR, processor);
}
return processor;
}
getProcessor()这个方法是SimpleIoProcessorPool中的，负责根据Session返回一个与之关联的Processor线程，这里用了session id对线程池中的线程总数取模的算法。与Session关联的Processor被添加到Session的Attribute中以便下次直接取出。
到这一部还没有看到Session内部的SocketChannel的IO事件是怎么注册到Processor线程的Selector对象上的，继续分析Processor的add()方法:
@Override
public final void add(S session) {
if (disposed || disposing) {
throw new IllegalStateException("Already disposed.");
}
// Adds the session to the newSession queue and starts the worker
newSessions.add(session);
startupProcessor();
}
private void startupProcessor() {
Processor processor = processorRef.get();
if (processor == null) {
processor = new Processor();
if (processorRef.compareAndSet(null, processor)) {
executor.execute(new NamePreservingRunnable(processor, threadName));
}
}
// Just stop the select() and start it again, so that the processor
// can be activated immediately.
wakeup();
}
//NamePreservingRunnable的run方法，显示给线程命名，然后执行Processor的run方法。
public void run() {
Thread currentThread = Thread.currentThread();
String oldName = currentThread.getName();
if (newName != null) {
setName(currentThread, newName);
}
try {
runnable.run();
} finally {
setName(currentThread, oldName);
}
}
private class Processor implements Runnable {
public void run() {
assert (processorRef.get() == this);
int nSessions = 0;
lastIdleCheckTime = System.currentTimeMillis();
int nbTries = 10;
for (;;) {
try {
...
int selected = select(SELECT_TIMEOUT);
...
nSessions += handleNewSessions();
...
if (selected > 0) {
// LOG.debug("Processing ..."); // This log hurts one of
// the MDCFilter test...
process();
}
...
}
} catch (ClosedSelectorException cse) {
ExceptionMonitor.getInstance().exceptionCaught(cse);
break;
} catch (Exception e) {
ExceptionMonitor.getInstance().exceptionCaught(e);
try {
Thread.sleep(1000);
} catch (InterruptedException e1) {
ExceptionMonitor.getInstance().exceptionCaught(e1);
}
}
}
}
}
这里开始显露一些马脚了，先是把session添加到newSessions这个队列中。然后建立了Processor实例，这就是具体的Processor线程。通过executor的execute()方法先是执行了NamePreservingRunnable的run()方法，其内部执行了Processor的run()方法。
执行Processor的run()中的select()其实就是调用其内部Selector对象的select()方法，会导致Processor线程的阻塞:
protected int select(long timeout) throws Exception {
return selector.select(timeout);
}
然后调用了Processor内部的Selector对象的wakeup()方法,wakeup()这个方法是当Selector对象执行select()方法阻塞时，立即返回。
@Override
protected void wakeup() {
wakeupCalled.getAndSet(true);
selector.wakeup();
}
于是后续就执行了:
private int handleNewSessions() {
int addedSessions = 0;
for (S session = newSessions.poll(); session != null; session = newSessions.poll()) {
if (addNow(session)) {
// A new session has been created
addedSessions++;
}
}
return addedSessions;
}
private boolean addNow(S session) {
boolean registered = false;
try {
init(session);
registered = true;
// Build the filter chain of this session.
IoFilterChainBuilder chainBuilder = session.getService().getFilterChainBuilder();
chainBuilder.buildFilterChain(session.getFilterChain());
// DefaultIoFilterChain.CONNECT_FUTURE is cleared inside here
// in AbstractIoFilterChain.fireSessionOpened().
// Propagate the SESSION_CREATED event up to the chain
IoServiceListenerSupport listeners = ((AbstractIoService) session.getService()).getListeners();
listeners.fireSessionCreated(session);
} catch (Exception e) {
ExceptionMonitor.getInstance().exceptionCaught(e);
try {
destroy(session);
} catch (Exception e1) {
ExceptionMonitor.getInstance().exceptionCaught(e1);
} finally {
registered = false;
}
}
return registered;
}
@Override
protected void init(NioSession session) throws Exception {
SelectableChannel ch = (SelectableChannel) session.getChannel();
ch.configureBlocking(false);
session.setSelectionKey(ch.register(selector, SelectionKey.OP_READ, session));
}
到此终于理清了，Processor线程先是阻塞的，由Acceptor线程把session添加到newSessions队列，然后通过wakeup将Processor从Selector对象的select()方法返回执行到handleNewSessions()方法，此方法会取出newSessions队列中的session然后通过addNow()方法执行NioProcessor的init()方法,由init()方法将session中的Channel的OP_READ事件注册到Selector对象上。
所以一个IoSession对应的是一个Proceccor线程，也是一个Selector对象，每个IoSession的读取数据处理一定是同步的。
既然有读就一定有写，记得上述代码中有一段:
private void processHandles(Iterator<H> handles) throws Exception {
while (handles.hasNext()) {
H handle = handles.next();
handles.remove();
// Associates a new created connection to a processor,
// and get back a session
S session = accept(processor, handle); //这里的processor是processor线程池
if (session == null) {
continue;
}
initSession(session, null, null);
// add the session to the SocketIoProcessor
session.getProcessor().add(session);
}
}
重点是initSession()方法:
protected final void initSession(IoSession session, IoFuture future, IoSessionInitializer sessionInitializer) {
...
((AbstractIoSession) session).setWriteRequestQueue(session.getService().getSessionDataStructureFactory()
.getWriteRequestQueue(session));
...
}
这里为session添加了WriteRequestQueue其实就是session的消息写入队列，当session被暂停或者WriteRequestQueue队列非空写入的消息会添加到这个队列里:
if (!s.isWriteSuspended()) {
if (writeRequestQueue.isEmpty(session)) {
// We can write directly the message
s.getProcessor().write(s, writeRequest);
} else {
s.getWriteRequestQueue().offer(s, writeRequest);
s.getProcessor().flush(s);
}
} else {
s.getWriteRequestQueue().offer(s, writeRequest);
}
而如果队列是空的则会执行write()方法，其实也是将写入请求插入队列然后直接执行flush()方法。
@Override
public void write(S session, WriteRequest writeRequest) {
WriteRequestQueue writeRequestQueue = session.getWriteRequestQueue();
writeRequestQueue.offer(session, writeRequest);
if (!session.isWriteSuspended()) {
this.flush(session);
}
}
flush()方法会在flushingSessions队列添加session并通过wakeup()方法将Processor线程从阻塞中恢复:
@Override
public final void flush(S session) {
// add the session to the queue if it's not already
// in the queue, then wake up the select()
if (session.setScheduledForFlush(true)) {
flushingSessions.add(session);
wakeup();
}
}
在Processor线程中会执行flush(long currentTime)方法,依次取出队列的每个session，注意这里的队列是ConcurrentLinkedQueue，所以不管在任何线程调用IoSession的write()方法写入消息，最终都会同步的插入到这个队列。
通过flushNow(session, currentTime)方法先是取出session的WriteRequestQueue队列（每个session都有一个写入消息的同步队列），然后依次取出其中的写消息请求，然后调用writeBuffer(S session, WriteRequest req, boolean hasFragmentation, int maxLength, long currentTime)，最终调用write(NioSession session, IoBuffer buf, int length) 通过session关联的Channel的write()方法将字节流发送。由于代码过多只贴出最终部分:
@Override
protected int write(NioSession session, IoBuffer buf, int length) throws IOException {
if (buf.remaining() <= length) {
return session.getChannel().write(buf.buf());
}
int oldLimit = buf.limit();
buf.limit(buf.position() + length);
try {
return session.getChannel().write(buf.buf());
} finally {
buf.limit(oldLimit);
}
}
到此分析Processor线程读写终于结束了，可以得出结论，会话的读写都是在Processor线程池中的一个Processor线程执行的。其中读消息是按事件顺序依次完成的，写消息可以由多个线程同时写，但是写入的请求一定是同步地插入到Session地写消息队列中，然后由Processor线程按顺序依次完成发送。担心Mina框架读写的并发问题可以打住了。
网友评论