这篇来看看AbstractPollingIoConnector抽象类，它用于用于实现客户端连接的轮询策略。处理逻辑基本上和上一篇文章说的AbstractPollingIoAcceptor类似，它继承自AbstractIoConnector，两个泛型参数分别是所处理的会话和客户端socket连接。底层的sockets会被不断检测，并当有任何一个socket需要被处理时就会被唤醒去处理。这个类封装了客户端socket的bind,connect和dispose等动作，其成员变量Executor用于发起连接请求，另一个AbstractPollingIoProcessor用于处理已经连接客户端的I/O操作请求，如读写和关闭连接。
其最重要的几个成员变量是：
private final Queue<ConnectionRequest> connectQueue = new ConcurrentLinkedQueue<ConnectionRequest>();//连接队列
private final Queue<ConnectionRequest> cancelQueue = new ConcurrentLinkedQueue<ConnectionRequest>();// 取消连接队列
先来看看当服务端调用connect后的处理过程：
复制代码
protected final ConnectFuture connect0(
SocketAddress remoteAddress, SocketAddress localAddress,
IoSessionInitializer<? extends ConnectFuture> sessionInitializer) {
H handle = null;
boolean success = false;
try {
handle = newHandle(localAddress);
if (connect(handle, remoteAddress)) {//若已经连接服务器成功
ConnectFuture future = new DefaultConnectFuture();
T session = newSession(processor, handle);//创建新会话
finishSessionInitialization(session, future, sessionInitializer);//结束会话初始化
session.getProcessor().add(session);//将剩下的处理交给IoProcessor
success = true;
return future;
}
success = true;
} catch (Exception e) {
return DefaultConnectFuture.newFailedFuture(e);
} finally {
if (!success && handle != null) {
try {
close(handle);
} catch (Exception e) {
ExceptionMonitor.getInstance().exceptionCaught(e);
}
}
}
ConnectionRequest request = new ConnectionRequest(handle, sessionInitializer);
connectQueue.add(request);//连接请求加入连接队列中
startupWorker();//开启工作线程处理连接请求
wakeup();//中断select操作
return request;
}
复制代码
真正的负责处理客户端请求的工作都是Worker线程完成的,
复制代码
private class Worker implements Runnable {
public void run() {
int nHandles = 0;
while (selectable) {
try {
int timeout = (int)Math.min(getConnectTimeoutMillis(), 1000L);//等待超时时间
boolean selected = select(timeout);//在超时时限内查看是否有可以被处理的选择键（状态
nHandles += registerNew();//取出连接队列队头的连接请求，将其注册一个用于连接的新的客户端socket， 并把它加入连接轮询池中
if (selected) {
nHandles-= processSessions(selectedHandles());//处理连接请求
}
processTimedOutSessions(allHandles());//处理超时连接请求
nHandles -= cancelKeys();
if (nHandles == 0) {
synchronized (lock) {
if (connectQueue.isEmpty()) {
worker = null;
break;
}
}
}
} catch (Throwable e) {
ExceptionMonitor.getInstance().exceptionCaught(e);
try {
Thread.sleep(1000);
} catch (InterruptedException e1) {
ExceptionMonitor.getInstance().exceptionCaught(e1);
}
}
}
if (selectable && isDisposing()) {
selectable = false;
try {
if (createdProcessor) {
processor.dispose();
}
} finally {
try {
synchronized (disposalLock) {
if (isDisposing()) {
destroy();
}
}
} catch (Exception e) {
ExceptionMonitor.getInstance().exceptionCaught(e);
} finally {
disposalFuture.setDone();
}
}
}
}
}
private int registerNew() {
int nHandles = 0;
for (; ;) {
ConnectionRequest req = connectQueue.poll();//取连接队列队头请求
if (req == null) {
break;
}
H handle = req.handle;
try {
register(handle, req);//注册一个用于连接的新的客户端socket， 并把它加入连接轮询池中
nHandles ++;
} catch (Exception e) {
req.setException(e);
try {
close(handle);
} catch (Exception e2) {
ExceptionMonitor.getInstance().exceptionCaught(e2);
}
}
}
return nHandles;
}
private int processSessions(Iterator<H> handlers) {//处理连接请求
int nHandles = 0;
while (handlers.hasNext()) {
H handle = handlers.next();
handlers.remove();
ConnectionRequest entry = connectionRequest(handle);
boolean success = false;
try {
if (finishConnect(handle)) {//连接请求成功完成，创建一个新会话
T session = newSession(processor, handle);
finishSessionInitialization(session, entry, entry.getSessionInitializer());//结束会话初始化
session.getProcessor().add(session);//将剩下的工作交给IoProcessor去处理
nHandles ++;
}
success = true;
} catch (Throwable e) {
entry.setException(e);
} finally {
if (!success) {//若连接失败，则将此连接请求放到取消连接队列中
cancelQueue.offer(entry);
}
}
}
return nHandles;
}
private void processTimedOutSessions(Iterator<H> handles) {//处理超时的连接请求
long currentTime = System.currentTimeMillis();//当前时间
while (handles.hasNext()) {
H handle = handles.next();
ConnectionRequest entry = connectionRequest(handle);
if (currentTime >= entry.deadline) {//当前时间已经超出了连接请求的底限
entry.setException(
new ConnectException("Connection timed out."));
cancelQueue.offer(entry);//将此连接请求放入取消连接队列中
}
}
}
private int cancelKeys() {//把取消队列中的连接请求给cancel掉
int nHandles = 0;
for (; ;) {
ConnectionRequest req = cancelQueue.poll();
if (req == null) {
break;
}
H handle = req.handle;
try {
close(handle);//关闭对应的客户端socket
} catch (Exception e) {
ExceptionMonitor.getInstance().exceptionCaught(e);
} finally {
nHandles ++;
}
}
return nHandles;
}
复制代码
本文转自Phinecos(洞庭散人)博客园博客，原文链接：http://www.cnblogs.com/phinecos/archive/2008/12/08/1350391.html，如需转载请自行联系原作者
