即时通信系统Openfire分析之三:ConnectionManager 连接管理

时间:2023-01-05 12:38:46

  Openfire是怎么实现连接请求的?

  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的概况在第二章有提及)。

    1. 初始化之一:ConnectionManagerImpl的构造方法

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

  这些参数的意义:

  • ConnectionType.SOCKET_C2S:ConectionType是个枚举类型,定义了所有connection的类型
  • ConnectionSettings.Client:提供了各个参数在数据库ofProperty表中的键,ConnectionListener构造方法会根据传入的键,从中读取相应的配置值
  • DEFAULT_PORT:设置监听的端口,对于C2S连接,openfire默认为5222端口
  • bindAddress: 配置文件中的network.interface,转化一个InetAddress。InetAddress是Java对IP地址的封装
  • certificateStoreManager:配置证书信息

  这些参数,设置了连接监听器的端口号、最大并发数等信息,最后封装在ConnectionConfiguration对象中,绑定到MINA的适配器NioSocketAcceptor。当MINA收到连接请求时,会根据端口的信息触发指定的监听器,进而执行相应的通信业务。

    2. 初始化之二:Module中定义的初始化方法

  这部分比较简单,检查了是否需要配置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());
}
} 

  关于缓冲区的使用,稍微提一下:

  • directBuffer:直接缓冲区, 为本地内存,不在Java堆中,不会被JVM回收。申请内存的API:ByteBuffer.allcateDirect(size)
  • heepBuffer:堆缓冲区,在堆中分配,当不再被引用的时候,buffer对象会被回收。申请内存的API:ByteBuffer.allocate(size)
  • 一般情况下:堆缓冲区的性能已经相当高,若无必要,使用堆缓冲区就足够。

  启动监听

模块启动的start()方法由module中定义,在相应的模块实现,在XMPPServer中被调用。start()方法的代码如下:
@Override
public void start() {
super.start();
startListeners();
SocketSendingTracker.getInstance().start();
CertificateManager.addListener(this);
}

  该方法执行了如下三步操作:

  • 启动所有监听,包括各个plugins、ConnectionListener、HTTP client
  • 启动SocketSendingTracker线程,每隔10秒调用checkHealth检查连接的Socket的状态。SocketSendingTracker.start()中,执行checkHealth()做了一件事情:如果某个Socket发送数据的事件大于60秒,或者长时间处于idle状态(表示长时间没有接收到客户端发来的心跳数据包),就调用forceClose将其关闭。
  • CertificateManager用来管理证书、监听ssl的相关时间。

  我们主要分析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

  这两个我们分别来看一下。

    1. ConnectionListener.start()方法:

  为了分析方便,这里只保留关键代码:

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处理。ConnetcionHandler其内部的处理机制,将在下一篇文章做分析。

XMPPCodecFactory负责解码接收到的消息、编码要发送的消息。

  即Openfire中的连接处理模型为:

request->XMPPCodecFactory.XMPPDecoder->ConnectionHandler->XMPPCodecFactory.XMPPEncoder->response
    关于MINA的处理逻辑,这里简述一下:

  NioSocketAcceptor是MINA的适配器,MINA中有个过滤器和处理器的概念,过滤器用来过滤数据,处理器用来处理数据。

  总的来说MINA的处理模型就是:

request->过滤器A->过滤器B->处理器->过滤器B->过滤器A->response

  request和response类似serlvet的request和response。

    至此,系统就开始能响应客户端的连接请求了!!

 

  刚刚分析startListeners()方法时,其中除了启动ConnetctionListener外,还启动了另一种监听:HttpBindManager,没忘记吧?下来对它也做一下分析。

    2. HttpBindManager.start()方法:

  这部分主要用于启用7070、7443端口,作为HTTP、HTTPS绑定端口,服务框架用的是Jetty。一般是在Web IM端用到。

  HttpBinManager中绑定监听了7070端口,并初始化HttpSessionManager。

  HttpSessionManager管理所有通过httpbing连接到openfire的议定,它是一个同步http的双向流。

    下面简单跟一下代码,部分代码省略掉了

    (1)HttpBindManager

  HttpBindManager构造方法:

private HttpBindManager() {
...
this.httpSessionManager = new HttpSessionManager();
....
}

  构造方法中虽然实例化了HttpSessionManager,然而,在HttpBindManager类中并没有对它做任何操作,只是提供了get方法。HttpSessionManager是在HttpBindServlet中使用的。

  Why?其实好理解,HttpSessionManager顾名思义,Http会话管理,要能管理首先是需要有会话产生,那会话在哪里产生?

  So,答案就出来了。

  至于为什么在要HttpBindManager中实例化,因为HttpBindManager中使用了单例,这样整个会话管理变得统一有序。

  OK,其他不多说,继续往下走:

    Start()方法中:configureHttpBindServer()函数做了端口绑定、Servlet绑定、以及WEB目录绑定,然后服务启动。 
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();
}

    (2)HttpSessionManager中所做的工作,就在其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开始能响应Http形式的请求。

    那么Openfire的整个网络监听,就分解完了。

  需要注意一点的是,上面内容,以C2S模式为例来讲解Openfire如何实现连接监听,但Openfire的ConnetionType并不止这一种,可以看一下这个枚举类:

public enum ConnectionType {

    SOCKET_S2S( "xmpp.socket.ssl.", null ),

    SOCKET_C2S( "xmpp.socket.ssl.client.", null ),

    BOSH_C2S( "xmpp.bosh.ssl.client.", SOCKET_C2S ),

    WEBADMIN( "admin.web.ssl.", SOCKET_S2S ),

    COMPONENT( "xmpp.component.", SOCKET_S2S ),

    CONNECTION_MANAGER( "xmpp.multiplex.", SOCKET_S2S );
}

  其他的几种类型,有兴趣的读者可以阅读代码做了解,分析方法与上文类似,内容与逻辑也相似,这里就不再赘述。

    而Openfire在接收到请求之后,是如何进行响应,在下一章讲解。