标签：etl filter 端口号 目录 信息 模式 简单 monit failed
XMPPServer.start()方法，完成Openfire的启动。但是，XMPPServer.start()方法中，并没有提及如何监听端口，那么Openfire是如何接收客户端的请求？
因为Openfire的核心功能，是通过Module来管理的，那么对应的连接管理应该就在Module中。
查看在XMPPServer.loadModules()方法中，有如下代码：
//Load this module always last since we don‘t want to start listening for clients // before the rest of the modules have been started loadModule(ConnectionManagerImpl.class.getName());
这个ConnectionManagerImpl类，就是连接的管理模块，而且注释中说到，它还是在其他模块启动后之后再启动。
那么下面，我们就来重点研究这个module，看看ConnectionManagerImpl如何实现连接监听，并处理消息响应的。
请求一般都与端口相对应，当客户端发出连接请求时，服务器要能够做出响应，首先需要对该请求的端口做监听。
ConnectionManagerImpl的继承关系中，它实现了ConnectionManager接口，在ConnectionManager中，除了定义端口的设定、监听开关等方法外，还定义一系列默认监听的端口号：
final int DEFAULT_PORT = 5222; final int DEFAULT_SSL_PORT = 5223; final int DEFAULT_COMPONENT_PORT = 5275; final int DEFAULT_COMPONENT_SSL_PORT = 5276; final int DEFAULT_SERVER_PORT = 5269; final int DEFAULT_MULTIPLEX_PORT = 5262; final int DEFAULT_MULTIPLEX_SSL_PORT = 5263;
这些端口号，在模块初始化的时候，被设定到对应的监听器对象中。
ConnectionManagerImpl的初始化，除了自身的构造方法外， 还有module中的initialize()方法（module的概况在第二章有提及）。
ConnectionManagerImpl的初始化，首先是构造了各类连接监听器，有如下几种：
private final ConnectionListener clientListener; private final ConnectionListener clientSslListener; private final ConnectionListener boshListener; private final ConnectionListener boshSslListener; private final ConnectionListener serverListener; private final ConnectionListener componentListener; private final ConnectionListener componentSslListener; private final ConnectionListener connectionManagerListener; // Also known as ‘multiplexer‘ private final ConnectionListener connectionManagerSslListener; // Also known as ‘multiplexer‘ private final ConnectionListener webAdminListener; private final ConnectionListener webAdminSslListener;
所有监听器都用ConnectionListener进行包装，以ConnectionType来做区分。这么处理可以制定一套方法来管理各类ConnectionListener，做到抽象统一。所有的ConnectionListener是模块启动时开启监听。
拿其中一种类型——SOCKET_C2S（即客户端-服务端），来观察一下它的构造方法，以下的分析也基于这一类型，这一类型是使用最多的。构造如下：
clientListener = new ConnectionListener( ConnectionType.SOCKET_C2S, ConnectionSettings.Client.PORT, DEFAULT_PORT, ConnectionSettings.Client.SOCKET_ACTIVE, ConnectionSettings.Client.MAX_THREADS, ConnectionSettings.Client.MAX_READ_BUFFER, ConnectionSettings.Client.TLS_POLICY, ConnectionSettings.Client.AUTH_PER_CLIENTCERT_POLICY, bindAddress, certificateStoreManager.getIdentityStoreConfiguration( ConnectionType.SOCKET_C2S ), certificateStoreManager.getTrustStoreConfiguration( ConnectionType.SOCKET_C2S ), ConnectionSettings.Client.COMPRESSION_SETTINGS );
这些参数的意义：
这些参数，设置了连接监听器的端口号、最大并发数等信息，最后封装在ConnectionConfiguration对象中，绑定到MINA的适配器NioSocketAcceptor。当MINA收到连接请求时，会根据端口的信息触发指定的监听器，进而执行相应的通信业务。
这部分比较简单，检查了是否需要配置MINA来使用直接缓冲区、或堆缓冲区，并调用IoBuffer做相应的配置。默认是只使用堆内存。
@Override public void initialize(XMPPServer server) { super.initialize(server); // Check if we need to configure MINA to use Direct or Heap Buffers // Note: It has been reported that heap buffers are 50% faster than direct buffers if (JiveGlobals.getBooleanProperty("xmpp.socket.heapBuffer", true)) { IoBuffer.setUseDirectBuffer(false); IoBuffer.setAllocator(new SimpleBufferAllocator()); } }
关于缓冲区的使用，稍微提一下：
模块启动的start()方法由module中定义，在相应的模块实现，在XMPPServer中被调用。start()方法的代码如下：
@Override public void start() { super.start(); startListeners(); SocketSendingTracker.getInstance().start(); CertificateManager.addListener(this); }
该方法执行了如下三步操作：
我们主要分析startListeners()方法，代码如下：
private synchronized void startListeners() { // Check if plugins have been loaded PluginManager pluginManager = XMPPServer.getInstance().getPluginManager(); if (!pluginManager.isExecuted()) { pluginManager.addPluginManagerListener(new PluginManagerListener() { public void pluginsMonitored() { // Stop listening for plugin events XMPPServer.getInstance().getPluginManager().removePluginManagerListener(this); // Start listeners startListeners(); } }); return; } for ( final ConnectionListener listener : getListeners() ) { try { listener.start(); } catch ( RuntimeException ex ) { Log.error( "An exception occurred while starting listener " + listener, ex ); } } // Start the HTTP client listener. try { HttpBindManager.getInstance().start(); } catch ( RuntimeException ex ) { Log.error( "An exception occurred while starting HTTP Bind listener ", ex ); } }
主要启动两个监听： （1）ConnectionListener （2）HttpBindManager
这两个我们分别来看一下。
为了分析方便，这里只保留关键代码：
public synchronized void start() { ...... Log.debug("Starting..."); if (getType() == ConnectionType.SOCKET_S2S) { connectionAcceptor = new LegacyConnectionAcceptor(generateConnectionConfiguration()); } else { connectionAcceptor = new MINAConnectionAcceptor(generateConnectionConfiguration()); } connectionAcceptor.start(); Log.info("Started."); }
该方法中，根据不同的ConnectionType初始化了连接的接受器ConnectionAcceptor并启动。
ConnectionAcceptor是个抽像类，被LegacyConnectionAcceptor、MINAConnectionAcceptor实现。
LegacyConnectionAcceptor仅能用于S2S的连接，且是之前所使用的方式。现在Oppenfire主要用的是MINA框架，这里我们只研究一下MINAConnectionAcceptor。
MINAConnectionAcceptor构造方法中，根据不同的连接类型，构造不同的ConnectionHandler。
完成MINAConnectionAcceptor构造之后，执行了MINAConnectionAcceptor.start()方法。
MINAConnectionAcceptor.start()方法代码如下：
public synchronized void start() { if ( socketAcceptor != null ) { Log.warn( "Unable to start acceptor (it is already started!)" ); return; } try { // Configure the thread pool that is to be used. final int initialSize = ( configuration.getMaxThreadPoolSize() / 4 ) + 1; final ExecutorFilter executorFilter = new ExecutorFilter( initialSize, configuration.getMaxThreadPoolSize(), 60, TimeUnit.SECONDS ); final ThreadPoolExecutor eventExecutor = (ThreadPoolExecutor) executorFilter.getExecutor(); final ThreadFactory threadFactory = new NamedThreadFactory( name + "-thread-", eventExecutor.getThreadFactory(), true, null ); eventExecutor.setThreadFactory( threadFactory ); // Construct a new socket acceptor, and configure it. socketAcceptor = buildSocketAcceptor(); if ( JMXManager.isEnabled() ) { configureJMX( socketAcceptor, name ); } final DefaultIoFilterChainBuilder filterChain = socketAcceptor.getFilterChain(); filterChain.addFirst( ConnectionManagerImpl.EXECUTOR_FILTER_NAME, executorFilter ); // Add the XMPP codec filter filterChain.addAfter( ConnectionManagerImpl.EXECUTOR_FILTER_NAME, ConnectionManagerImpl.XMPP_CODEC_FILTER_NAME, new ProtocolCodecFilter( new XMPPCodecFactory() ) ); // Kill sessions whose outgoing queues keep growing and fail to send traffic filterChain.addAfter( ConnectionManagerImpl.XMPP_CODEC_FILTER_NAME, ConnectionManagerImpl.CAPACITY_FILTER_NAME, new StalledSessionsFilter() ); // Ports can be configured to start connections in SSL (as opposed to upgrade a non-encrypted socket to an encrypted one, typically using StartTLS) if ( configuration.getTlsPolicy() == Connection.TLSPolicy.legacyMode ) { final SslFilter sslFilter = encryptionArtifactFactory.createServerModeSslFilter(); filterChain.addAfter( ConnectionManagerImpl.EXECUTOR_FILTER_NAME, ConnectionManagerImpl.TLS_FILTER_NAME, sslFilter ); } // Throttle sessions who send data too fast if ( configuration.getMaxBufferSize() > 0 ) { socketAcceptor.getSessionConfig().setMaxReadBufferSize( configuration.getMaxBufferSize() ); Log.debug( "Throttling read buffer for connections to max={} bytes", configuration.getMaxBufferSize() ); } // Start accepting connections socketAcceptor.setHandler( connectionHandler ); socketAcceptor.bind( new InetSocketAddress( configuration.getBindAddress(), configuration.getPort() ) ); } catch ( Exception e ) { System.err.println( "Error starting " + configuration.getPort() + ": " + e.getMessage() ); Log.error( "Error starting: " + configuration.getPort(), e ); // Reset for future use. if (socketAcceptor != null) { try { socketAcceptor.unbind(); } finally { socketAcceptor = null; } } } }
从上面的代码可以看到，MINAConnectionAcceptor.start()做了四件事：
（1）建立线程池
（2）构建了一个socketAcceptor
（3）添加xmpp解码器与编码器到socketAcceptor
（4）将connectionHandler注入socketAcceptor并绑定socketAcceptor.bind，
其中：
ConnectionHandler是连接处理器，MINA接收到的请求最后都转由ConnectionHandler处理，其内部的处理机制，将在下文另起一节来分析。
XMPPCodecFactory负责解码接收到的消息、编码要发送的消息。
即Openfire中的连接处理模型为：
request->XMPPCodecFactory.XMPPDecoder->ConnectionHandler->XMPPCodecFactory.XMPPEncoder->response
NioSocketAcceptor是MINA的适配器，MINA中有个过滤器和处理器的概念，过滤器用来过滤数据，处理器用来处理数据。
总的来说MINA的处理模型就是：
request->过滤器A->过滤器B->处理器->过滤器B->过滤器A->response
request和response类似serlvet的request和response。
刚刚分析startListeners（）方法时，其中除了启动ConnetctionListener外，还启动了另一种监听：HttpBindManager，没忘记吧？下来对它也做一下分析。
这部分主要用于启用7070、7443端口，作为HTTP、HTTPS绑定端口，服务框架用的是Jetty。一般是在Web IM端用到。
HttpBinManager中绑定监听了7070端口，并初始化HttpSessionManager。
HttpSessionManager管理所有通过httpbing连接到openfire的议定，它是一个同步http的双向流。
HttpBindManager构造方法:
private HttpBindManager() { ... this.httpSessionManager = new HttpSessionManager(); .... }
构造方法中虽然实例化了HttpSessionManager，然而，在HttpBindManager类中并没有对它做任何操作，只是提供了get方法。HttpSessionManager是在HttpBindServlet中使用的。
Why？其实好理解，HttpSessionManager顾名思义，Http会话管理，要能管理首先是需要有会话产生，那会话在哪里产生？
So，答案就出来了。
至于为什么在要HttpBindManager中实例化，因为HttpBindManager中使用了单例，这样整个会话管理变得统一有序。
OK，其他不多说，继续往下走：
public void start() { certificateListener = new CertificateListener(); CertificateManager.addListener(certificateListener); if (!isHttpBindServiceEnabled()) { return; } bindPort = getHttpBindUnsecurePort(); bindSecurePort = getHttpBindSecurePort(); configureHttpBindServer(bindPort, bindSecurePort); try { httpBindServer.start(); Log.info("HTTP bind service started"); } catch (Exception e) { Log.error("Error starting HTTP bind service", e); } }
configureHttpBindServer()：
private synchronized void configureHttpBindServer(int port, int securePort) { final QueuedThreadPool tp = new QueuedThreadPool(processingThreads); tp.setName("Jetty-QTP-BOSH"); httpBindServer = new Server(tp); .... createBoshHandler(contexts, "/http-bind"); createCrossDomainHandler(contexts, "/crossdomain.xml"); loadStaticDirectory(contexts); HandlerCollection collection = new HandlerCollection(); httpBindServer.setHandler(collection); collection.setHandlers(new Handler[] { contexts, new DefaultHandler() }); }
解释一下QueuedThreadPool类，该类是jetty的一个线程池，它实现了org.eclipse.jetty.util.thread.ThreadPool接口，并继承org.eclipse.jetty.util.component.AbstractLifeCycle。
createBoshHandler()：
private void createBoshHandler(ContextHandlerCollection contexts, String boshPath) { ServletContextHandler context = new ServletContextHandler(contexts, boshPath, ServletContextHandler.SESSIONS); ...... context.addServlet(new ServletHolder(new HttpBindServlet()),"/*"); ...... }
createCrossDomainHandler()：
private void createCrossDomainHandler(ContextHandlerCollection contexts, String crossPath) { ServletContextHandler context = new ServletContextHandler(contexts, crossPath, ServletContextHandler.SESSIONS); ...... context.addServlet(new ServletHolder(new FlashCrossDomainServlet()),""); }
loadStaticDirectory()：
private void loadStaticDirectory(ContextHandlerCollection contexts) { File spankDirectory = new File(JiveGlobals.getHomeDirectory() + File.separator + "resources" + File.separator + "spank"); ...... WebAppContext context = new WebAppContext(contexts, spankDirectory.getPath(), "/"); context.setWelcomeFiles(new String[]{"index.html"}); }
最后在ConnectionManagerImpl中调用HttpBindManager.start()就完成启动，Openfire与Jetty开始进行连接，关于Jetty的相关机制，这里就不做延伸了。
而HttpSessionManager在HttpBindServlet的初始化中开启，当然在HttpBindServlet被destroy()时，也自然会stop()掉。
HttpBindServlet.init()：
public void init(ServletConfig servletConfig) throws ServletException { super.init(servletConfig); boshManager = HttpBindManager.getInstance(); sessionManager = boshManager.getSessionManager(); sessionManager.start(); }
HttpSessionManager.start()：
public void start() { Log.info( "Starting instance" ); this.sessionManager = SessionManager.getInstance(); final int maxClientPoolSize = JiveGlobals.getIntProperty( "xmpp.client.processing.threads", 8 ); final int maxPoolSize = JiveGlobals.getIntProperty("xmpp.httpbind.worker.threads", maxClientPoolSize ); final int keepAlive = JiveGlobals.getIntProperty( "xmpp.httpbind.worker.timeout", 60 ); sendPacketPool = new ThreadPoolExecutor(getCorePoolSize(maxPoolSize), maxPoolSize, keepAlive, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>(), // unbounded task queue new NamedThreadFactory( "httpbind-worker-", true, null, Thread.currentThread().getThreadGroup(), null ) ); sendPacketPool.prestartCoreThread(); // Periodically check for Sessions that need a cleanup. inactivityTask = new HttpSessionReaper(); TaskEngine.getInstance().schedule( inactivityTask, 30 * JiveConstants.SECOND, 30 * JiveConstants.SECOND ); }
解释一下：
（1）keepAlive，多余空闲线程等待心任务的的最长时间60秒
（2）ThreadPoolExecutor配置了线程池，池中所保持的线程数和最大线程数均为8个
（3）newLinkedBlockingQueue<Runnable>()，执行前保持的队列，此队列仅保持由execute 方法提交的 Runnable 任务
（4）NamedThreadFactory，创建新线程的工厂
（5）sendPacketPool.prestartCoreThread()：该方法为启动核心线程，使其处于等待工作的空闲状态。仅当执行新任务时，此操作才重写默认的启动核心线程策略。
最后启动了一个线程来查看哪些会话需要被关闭：
inactivityTask = new HttpSessionReaper(); TaskEngine.getInstance().schedule( inactivityTask, 30 * JiveConstants.SECOND, 30 * JiveConstants.SECOND );
进入看看HttpSessionReaper.run()方法：
private class HttpSessionReaper extends TimerTask { @Override public void run() { long currentTime = System.currentTimeMillis(); for (HttpSession session : sessionMap.values()) { try { long lastActive = currentTime - session.getLastActivity(); if (Log.isDebugEnabled()) { Log.debug("Session was last active " + lastActive + " ms ago: " + session.getAddress()); } if (lastActive > session.getInactivityTimeout() * JiveConstants.SECOND) { Log.info("Closing idle session: " + session.getAddress()); session.close(); } } catch (Exception e) { Log.error("Failed to determine idle state for session: " + session, e); } } } }
这个线程的意义：定时将一些超时了的闲置状态的会话清理掉。
其中：
session.getLastActivity()：这个方法以毫秒为时间单位返回关闭http连接的时间
getInactivityTimeout()：这个方法以秒为单位返回不活跃或被终止会话时间
那么Openfire的整个网络监听，就分解完了。
注意一点的是，上面内容，是以客户端-服务端模式为例来讲解Openfire的连接监听，但Openfire的ConnetionType并不止这一种，还有server-server等其他类型，分析方法类似，内容也相似，这里就不再赘述，有兴趣的朋友可以自已做了解。
Openfire分析之三：ConnectionManager 连接管理（1）
标签：etl filter 端口号 目录 信息 模式 简单 monit failed
原文地址：http://www.cnblogs.com/Fordestiny/p/7467513.html
