一、 ConnectionOpenThread 使用单例模式,他是负责建立服务端SOCKET和 接收连接客户端socket 线程。
在初始化ConnectionManager的时候ConnectionManager.connectThread 属性所引用的服务端SOCKET连接线程ConnectionOpenThread 就被初始化了
ConnectionManager:: private static ConnectionOpenThread connectThread = ConnectionOpenThread.getInstance();
ConnectionOpenThread .getInstance()的实现
private Selector selector= null; public static ConnectionOpenThread getInstance() { if (acceptThread == null) { acceptThread = new ConnectionOpenThread(); Thread thrd = new Thread(acceptThread); thrd.setName("ConnectionOpenThread"); thrd.start(); //启动ConnectionOpenThread线程,则this.run()方法将被被执行 if (log.isLoggable(Level.FINER)) { log.finer("ConnectionOpenThread started."); } } // end of if (acceptThread == null) return acceptThread; } private ConnectionOpenThread() { ....... try { selector = Selector.open();//得到一个选择器,可以去了解下nio api } catch (Exception e) { log.log(Level.SEVERE, "Server I/O error, can't continue my work.", e); stopping = true; } // end of try-catch }
ConnectionOpenThread .run()的实现
在该方法中,selector管理的都是服务端SOCKET
public void run() { while (!stopping) { try { selector.select(); //此方法为阻塞方法,当选择器管理channel(也就是向selector注册的channel) 中发生读、写或异常事件时,select()将会被触发会往下执行 // Set<SelectionKey> selected_keys = selector.selectedKeys(); // for (SelectionKey sk : selected_keys) { //返回已此通道已准备就绪的键集,已选择始终是键集的一个子集。 //begin iterator for (Iterator i = selector.selectedKeys().iterator(); i.hasNext(); ) { SelectionKey sk = (SelectionKey) i.next(); i.remove(); SocketChannel sc = null; boolean throttled = false; int port_no = 0; if ((sk.readyOps() & SelectionKey.OP_ACCEPT) != 0) { //在此是否为被动SOCKET也就是服务端SOCKET,是则接受客户端socket ServerSocketChannel nextReady = (ServerSocketChannel) sk.channel(); port_no = nextReady.socket().getLocalPort(); sc = nextReady.accept();//得到一个客户端SOCKET ... } // end of if (sk.readyOps() & SelectionKey.OP_ACCEPT) if ((sk.readyOps() & SelectionKey.OP_CONNECT) != 0) { sk.cancel(); // 从Selector中删除指定的SelectionKey //所以这个普通的conect socket只会处理一次侦听到的发生事件 sc = (SocketChannel) sk.channel();//得到connect SOCKET } // end of if (sk.readyOps() & SelectionKey.OP_ACCEPT) if (sc != null) { //设置接收到的SOCKET的一些信息 try { sc.configureBlocking(false);//将客户端通道设置为非阻塞 sc.socket().setSoLinger(false, 0); sc.socket().setReuseAddress(true); //每个ServerSocketChannel在创建注册到selector 时就被绑定了一个ConnectionOpenListener对象, 用这个对象来处理该接受到的socket,该注册过程 在addAllWaiting()中进行 ConnectionOpenListener al = (ConnectionOpenListener) sk.attachment(); sc.socket().setTrafficClass(al.getTrafficClass()); sc.socket().setReceiveBufferSize(al.getReceiveBufferSize()); al.accept(sc);//此方法 为建立连接socket的进行后续处理的设定 } catch (java.net.SocketException e) { ConnectionOpenListener al = (ConnectionOpenListener) sk.attachment(); al.accept(sc); } } else { log.log(Level.INFO, "Can not obtain socket channel from selection key, throttling activated = {0}, for port: {1}", new Object[] { throttled, port_no }); } // end of if (sc != null) else ++accept_counter; } //end of iterator addAllWaiting();//加载要注册到selector中的ServerSocketChannel或connect socket } catch (IOException e) { log.log(Level.SEVERE, "Server I/O error.", e); // stopping = true; } // end of catch catch (Exception e) { log.log(Level.SEVERE, "Other service exception.", e); // stopping = true; } // end of catch } }
在说解到addAllWaiting();加载要注册到selector中的ServerSocketChannel时,先看下源码:
在waiting队列中如果有等待处理的ConnectionOpenListener对象,则创建一个对应的ServerSocketChannel
private void addAllWaiting() throws IOException { ConnectionOpenListener al = null; while ((al = waiting.poll()) != null) { try { addPort(al);//绑定相关的端口进行监听 } catch (Exception e) { log.log(Level.WARNING, "Error: creating connection for: " + al, e); al.accept(null); } // end of try-catch } // end of for () } private void addPort(ConnectionOpenListener al) throws IOException { if ((al.getConnectionType() == ConnectionType.connect) && (al.getRemoteAddress() != null)) { addISA(al.getRemoteAddress(), al); } else if ((al.getIfcs() == null) || (al.getIfcs().length == 0) || al.getIfcs()[0] .equals("ifc") || al.getIfcs()[0].equals("*")) { addISA(new InetSocketAddress(al.getPort()), al);//绑定到InetSocketAddress进行监听服务 } else { for (String ifc : al.getIfcs()) { addISA(new InetSocketAddress(ifc, al.getPort()), al); } // end of for () } // end of if (ip == null || ip.equals("")) else } //addISA(..)这才是真正创建ServerSocketChannel方法,绑定到服务器某一个端口上进行监听服务, //开启了服务端socket private void addISA(InetSocketAddress isa, ConnectionOpenListener al)throws IOException { switch (al.getConnectionType()) { case accept : ... ServerSocketChannel ssc = ServerSocketChannel.open(); ssc.socket().setReceiveBufferSize(al.getReceiveBufferSize()); ssc.configureBlocking(false);//服务端socket也是非阻塞方法 ssc.socket().bind(isa, (int) (port_throttling)); //绑定到相关地址的某一个端口上 ssc.register(selector, SelectionKey.OP_ACCEPT, al);//注册服务端socket到selector中, 并且附带绑定一个ConnectionOpenListener对象,该对象为服务端socket接收到新来的socket 进行后续处理。所以selector能监听这些已注册socket的事件发生 break; case connect : ... //服务器socket之间要进行通讯,则先要连接 SocketChannel sc = SocketChannel.open(); sc.socket().setReceiveBufferSize(al.getReceiveBufferSize()); sc.socket().setTrafficClass(al.getTrafficClass()); sc.configureBlocking(false); sc.connect(isa); sc.register(selector, SelectionKey.OP_CONNECT, al); //在此也注册到ConnectionOpenThread.selector中 break; default : .. break; } }
二、从以上addAllWaiting();分析中看到处理的都是waiting队列里的ConnectionOpenListener对象,那这个ConnectionOpenListener对象是什么时候就会被放到waiting队列的呢,这得从ConnectionManager.initializationCompleted()中说起,在启动章节中分析到MessageRouter.setProperties(map)负责加载了其它的组件最后对每一个组件都执行了初始化完成动作。从而ConnectionManager.initializationCompleted()将会被执行
MessageRouter::
for (ServerComponent comp : components.values()) {
comp.initializationCompleted();
}
ConnectionManager.initializationCompleted()源码如下
public void initializationCompleted() { if (isInitializationComplete()) { // Do we really need to do this again? return; } super.initializationCompleted(); initializationCompleted = true; //加载组件中的服务配置 for (Map<String, Object> params : waitingTasks) { //启动一个定时任务,设置准备加入ConnectionOpenThread.waiting的ConnectionListener对象 reconnectService(params, connectionDelay); } waitingTasks.clear(); if ( null != watchdog ){ watchdog.start(); } }2.1 也许看到上面的waitingTask你在想他是什么样的配置信息呢,其实他就是启动服务器监听端口的配置,系统 默认的有如下几种,bosh,c2s,s2s,ws2s组件的服务器配置, waitingTask装的是每一个组件的端口配置信息
c2s/connections/5222/type[S]=accept c2s/connections/5222/socket[S]=plain c2s/connections/5222/ifc[s]=* c2s/connections/5222/remote-host[S]=localhost c2s/connections/5222/connections/tls/required[B]=false c2s/connections/5223/type[S]=accept c2s/connections/5223/socket[S]=ssl c2s/connections/5223/ifc[s]=* c2s/connections/5223/remote-host[S]=localhost c2s/connections/5223/connections/tls/required[B]=false c2s/connections/ports[i]=5222, 5223 bosh/connections/5280/type[S]=accept bosh/connections/5280/socket[S]=plain bosh/connections/5280/ifc[s]=* bosh/connections/5280/remote-host[S]=localhost bosh/connections/5280/connections/tls/required[B]=false bosh/connections/ports[i]=5280 s2s/connections/5269/type[S]=accept s2s/connections/5269/socket[S]=plain s2s/connections/5269/ifc[s]=* s2s/connections/5269/remote-host[S]=localhost s2s/connections/5269/connections/tls/required[B]=false s2s/connections/ports[i]=5269 ws2s/connections/5290/type[S]=accept ws2s/connections/5290/socket[S]=plain ws2s/connections/5290/ifc[s]=* ws2s/connections/5290/remote-host[S]=localhost ws2s/connections/5290/connections/tls/required[B]=false ws2s/connections/ports[i]=5290
ConnectionManager.setProperties(Map<String, Object> props){ ... for (int i = 0; i < ports.length; i++) { addWaitingTask(port_props); } } //conn信息加入waitingTasks队列 protected void addWaitingTask(Map<String, Object> conn) { if (initializationCompleted) { reconnectService(conn, connectionDelay); } else { waitingTasks.add(conn); } }
2.2 然后回来ConnectionManager对象继续分析ConnectionOpenThread.waiting队列是怎么样增加了数据的。
ConnectionManager:: //启动一个定时任务,设置准备加入ConnectionOpenThread.waiting的ConnectionListener对象 private void reconnectService(final Map<String, Object> port_props, long delay) { ... addTimerTask(new tigase.util.TimerTask() { @Override public void run() { startService(port_props); } }, delay); } //ConnectionOpenThread是单例模式,在此该对象只被初始化一次 private static ConnectionOpenThread connectThread = ConnectionOpenThread.getInstance(); private void startService(Map<String, Object> port_props) { if (port_props == null) { throw new NullPointerException("port_props cannot be null."); } //根据组件的配置信息生成一个相关的ConnectionListener对象 ConnectionListenerImpl cli = new ConnectionListenerImpl(port_props); if (cli.getConnectionType() == ConnectionType.accept) { pending_open.add(cli); } //将ConnectionListener对象加入ConnectionOpenThread.waiting队列中 connectThread.addConnectionOpenListener(cli); }
最后在ConnectionOpenThread中可以看到ConnectionListener是怎么样加入waiting队列的了
ConnectionOpenThread:: public void addConnectionOpenListener(ConnectionOpenListener al) { waiting.offer(al); selector.wakeup(); }