Mina2.0框架源码剖析（八）

这篇来看看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://phinecos.cnblogs.com/

评论

1 楼 liangqingxiu 2012-05-24