《用Java写一个通用的服务器程序》03 处理新socket

在讲监听器时说过处理的新的socket要尽快返回，监听器调用的是ClientFactory的createPhysicalConnection方法，那么就来看这个方法：

    public boolean createPhysicalConnection(PushClientSocket socket,
            boolean isObserver, ListenerOptions listenerOptions) {
        PhysicalConnectionPool thePhysicalConnectionPool = 
            serverImpl.getPhysicalConnectionPool();
        IOQueue<PhysicalConnection> ioQueue = serverImpl.getIOQueue();

        // 内置了一个PhysicalConnection的对象池，这样可以避免每次都要
        // 创建PhysicalConnection对象，可以加快处理速度
        PhysicalConnection connection = 
            thePhysicalConnectionPool.borrowObject();
        // 把PhysicalConnection对象和socket对象关联起来
        connection.reset(socket, isObserver, listenerOptions);

        // 初始化协议，分配buffer，用来缓存解析请求时的数据
        if (!connection.setUpProtocolContexts()) {
            thePhysicalConnectionPool.returnObject(connection);
            return false;
        }

        Debug.debug("Physical Connection Created for client from: " + 
                socket.getIP());

        // 把连接注册到I/O队列中，这样就可以监听请求
        if (!ioQueue.addSocketContext(socket, connection)) {
            thePhysicalConnectionPool.returnObject(connection);
            //leave socket close to acceptor
            return false;
        }

        Debug.debug("Queue adds client from: " + socket.getIP());

        // 把创建的PhysicalConnection加入pending队列中，此时连接
        // 还不算是真正的已连接状态，要等到第一个请求到达并正确
        // 处理之后才会是已连接状态，并且会创建一个LogicalConnection
        // 和这个PhysicalConnection相关联
        addPhysicalConnection(connection);
        
        // 初始化PhysicalConnection
        serverImpl.getDispatcher().handleInitialize(connection);

        return true;
    }

ClientFactory是PhysicalConnection的管理程序，这个方法的的作用就是创建PhysicalConnection和新的socket相关联，并且把PhysicalConnection加入请求监听的I/O队列。因此来说说IOQueue。

IOQueue本身是一个接口：

public interface IOQueue<T> {

    public boolean create();

    public void free();

    // 从队列中获取事件，默认实现是带有阻塞超时的，即当没有事件
    // 时会阻塞一段时间，超时就会返回null
    public IOEvent<T> getQueuedEvent(boolean isInputEvents);

    // 注册连接，context是关联对象，类似于附件
    public boolean addSocketContext(PushClientSocket socket, T context);

    // 取消注册
    public void deleteSocketContext(PushClientSocket socket);

    // IOQueue的事件监听是一次性，这是为了防止事件在没有被处理之前，这个事件
    // 再次被捕捉到（Java的read/write事件都是这样），因此这个方法会在事件
    // 被处理后调用，再次注册。
    public boolean rearmSocketForWrite(PushClientSocket socket, T context);

    // Read事件代表的就是从客户端有数据过来
    public boolean rearmSocketForRead(PushClientSocket socket, T context);
}

IOQueueImpl是IOQueue的Java NIO版本的实现。IOQueueImpl会内置一个独立线程以及一个Selector，这里关于注册有一点需要说明：

PushClientSocketImpl的registerSelector方法用于注册socket，这里需要调用wakeup方法。因为如果独立线程会调用Selector的select方法等待新的数据，这个时候直接

调用register方法会被阻塞，因此需要先调用wakeup唤醒selector。

    public SelectionKey registerSelector(Selector selector, int ops, 
            Object attachment) throws IOException {
        selector.wakeup(); // To prevent block when calling register method
        return channel.register(selector, ops, attachment);
    }

接着说说独立线程监听事件，因为OP_WRITE的特殊性，这里只监听OP_READ事件。

    private void pollEvents(boolean isOutPoll) {
        Selector selector;
        BlockingQueue<SelectionKey> queue;
        if (isOutPoll) {
            return;
        } else {
            selector = this.inPollSelector;
            queue = this.inPollQueue;
        }
        
        List<SelectionKey> cache = new LinkedList<SelectionKey>();

        while (isPolling) {
            try {
                selector.select();
            } catch (IOException e) {
                continue;
            }
            
            // 这里调用yield释放控制权是为了刚刚提到的register方法能被顺利执行
            Thread.yield();
            
            // Add into cache (Add into the blocking queue directly
            // may block so that the selector cannot release the selection 
            // key in time)
            if (selector.isOpen()) {
                for (SelectionKey key : selector.selectedKeys()) {
                    // 前面提到监听事件是一次性的，因此这里取消监听
                    // 后面再调用rearm方法重新注册
                    key.cancel(); 
                    cache.add(key);
                }
                
                // Clear the keys
                selector.selectedKeys().clear();
                
                // 因为使用了限定长度的BlockingQueue，可能因为队列已满导致阻塞
                // 因此先把事件转移到缓存中，释放Selector
                queue.addAll(cache);
                cache.clear();
            } else {
                break; // selector closed
            }
        }
    }

顺便说说Demultiplexor获取事件调用的getQueuedEvent方法，这里使用BlockingQueue来实现阻塞等待：

    public IOEvent<PhysicalConnection> getQueuedEvent(boolean isInputEvents) {
        final IOEventType type;
        final BlockingQueue<SelectionKey> pollQueue;
        
        if (isInputEvents) {
            type = IOEventType.read;
            pollQueue = inPollQueue;
        } else {
            type = null;
            pollQueue = null;
        }
        
        if (pollQueue == null) {
            return null;
        }

        try {
            // 设置1秒的超时，这样后面关闭时清空I/O队列的时候不会导致
            // Demultiplexor一直被阻塞
            SelectionKey key = pollQueue.poll(1000L, TimeUnit.MILLISECONDS);
            if (key != null) {
                if (key.attachment() instanceof PhysicalConnection) {
                    return new IOEvent<PhysicalConnection>(type, 
                            (PhysicalConnection)(key.attachment()));
                }
            }
        } catch (InterruptedException e) {
            // Ignore
        }
        
        return null;
    }

关于新socket的处理就说这么多吧。