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;    }

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);}

    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);    }

    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            }        }    }

    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;    }