在讲监听器时说过处理的新的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的处理就说这么多吧。
时间: 2024-10-07 12:15:41