说明

本文源码基于Openfire4.0.2。

Openfire的启动

Openfire的启动过程非常的简单，通过一个入口初始化lib目录下的openfire.jar包，并启动一个XMPPServer实例。

下面就是ServerStarter.start方法的代码片断：

Class containerClass = loader.loadClass("org.jivesoftware.openfire.XMPPServer");

containerClass.newInstance();

这样一个openfire实例就已经启动了。

XMPPServer类

这个XmppServer类是单实例的对象，这样在服务器调用时可以获取一个实例。既然是个对象就会有构造的过程，XMPPServer在构造过程中会对服务进行初始化，这个过程包括：

初始化配置参数

检查是否需要安装

初始化Module

启动统计模块

启动plugin

基本就是这么简单，还是非常简洁明了。这里也可以大概知道在openfire里主要是module和plugin两类模块，一般情况下内部的模块都用module，对于一些功能的扩展或者第三方的开发扩展使用Plugin。官方其实也会自己写一个插件来扩展功能，说明插件还是比较灵活的。

提一提Module的加载过程

下面代码是module的加载过程

if (!setupMode) {

verifyDataSource();//First load all the modules so that modules may access other modules while//being initialized

loadModules();//Initize all the modules

initModules();//Start all the modules

startModules();

}

可以看到，分了三个步骤：

加载模块：是对模块类的实例化过程，就是创建对象

初始化模块：就是调用module.initialize(this);，其实就是调用模块的初始化方法

启动模块：module.start();，同理就是调用启动模块

这是因为openfire规范了module的接口抽象Module，所有的模块都要按照这个规范实现，看代码：

public interfaceModule {/*** Returns the name of the module for display in administration interfaces.

*

*@returnThe name of the module.*/String getName();/*** Initialize the module with the container.

* Modules may be initialized and never started, so modules

* should be prepared for a call to destroy() to follow initialize().

*

*@paramserver the server hosting this module.*/

voidinitialize(XMPPServer server);/*** Start the module (must return quickly). Any long running

* operations should spawn a thread and allow the method to return

* immediately.*/

voidstart();/*** Stop the module. The module should attempt to free up threads

* and prepare for either another call to initialize (reconfigure the module)

* or for destruction.*/

voidstop();/*** Module should free all resources and prepare for deallocation.*/

voiddestroy();

}

这也标示了Module的生命周期，Openfire会管理这些Module的生命周期，以此来保证各个模块的启动与释放。

Connection管理模块

整个启动过程有点奇怪，并没有看到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就是连接的管理模块，这里有个注释，就是在其他模块启动后之后再启动监听模块。

在ConnectionManagerImpl中管理了主要的连接，都是以ConnectionListener的来管理，这个类用于包装连接。我的理解就是一个连接抽象吧，这样对于代码来说写起来比较统一。看下面代码中Manager管理着哪些：

private finalConnectionListener clientListener;private finalConnectionListener clientSslListener;private finalConnectionListener boshListener;private finalConnectionListener boshSslListener;private finalConnectionListener serverListener;private finalConnectionListener componentListener;private finalConnectionListener componentSslListener;private final ConnectionListener connectionManagerListener; //Also known as 'multiplexer'

private final ConnectionListener connectionManagerSslListener; //Also known as 'multiplexer'

private finalConnectionListener webAdminListener;private final ConnectionListener webAdminSslListener;

这里面除了server只有一个外，其他的都是两个，其中一个是SSL的。它们主要是什么链接？

client：表示客户端连接

bosh：就是HTTP绑定的连接

server：服务器到服务器的socket连接

component：组件到服务器的连接

connectionManager：是指通过connectionManager连接器过来的连接

webAdmin：是指web控制台的连接

这里面bosh和webAdmin使用的是http协议，所以连接并不是长连接，其他的都是socket。

openfire里使用了Mina来实现socket网络处理。只不过看代码中对于S2S类型的连接使用的不是mina，如下代码：

if ( getType() ==ConnectionType.SOCKET_S2S )

{

connectionAcceptor= newLegacyConnectionAcceptor( generateConnectionConfiguration() );

}else{

connectionAcceptor= newMINAConnectionAcceptor( generateConnectionConfiguration() );

}

LegacyConnectionAcceptor是个废弃的类，但不知道为什么s2s还要用这个呢？看了看实现，LegacyConnectionAcceptor就是起了一个线程，在线程里建了一个ServerSocket。可能以后还是会迁移这部分代码吧。

在connectionAcceptor中会根据类型创建一个ConnectionHandler用于实现具体的业务功能，而ConnectionHandler都是基于org.apache.mina.core.service.IoHandlerAdapter派生的类，而IoHandlerAdapter又是IoHandler的适配接口，所以实质上就是IoHandler。下面是类继承关系：

在这些Handler里完成的主要是每个连接打开、关闭和数据收发等操作的处理。而其中比较关键的一个步骤就是在sessionOpened中设置了StanzeHandler，而每种ConnectionHandler都有自己的StanzeHandler实现。以ClientConnectionHandler为例子，其中ClientConnectionHandler复写了父类的createStanzaHandler方法，这里面

@Override

StanzaHandler createStanzaHandler(NIOConnection connection) {return newClientStanzaHandler(XMPPServer.getInstance().getPacketRouter(), connection);

}

这里使用的是clientStanzaHandler，表示是客户端的数据节处理者。而最终的createStanzaHandler调用是在父类ConnectionHandler的sessionOpened完成的，

@Overridepublic void sessionOpened(IoSession session) throwsException {//Create a new XML parser for the new connection. The parser will be used by the XMPPDecoder filter.

final XMLLightweightParser parser = newXMLLightweightParser(StandardCharsets.UTF_8);

session.setAttribute(XML_PARSER, parser);//Create a new NIOConnection for the new session

final NIOConnection connection =createNIOConnection(session);

session.setAttribute(CONNECTION, connection);

session.setAttribute(HANDLER, createStanzaHandler(connection));//Set the max time a connection can be idle before closing it. This amount of seconds//is divided in two, as Openfire will ping idle clients first (at 50% of the max idle time)//before disconnecting them (at 100% of the max idle time). This prevents Openfire from//removing connections without warning.

final int idleTime = getMaxIdleTime() / 2;if (idleTime > 0) {

session.getConfig().setIdleTime(IdleStatus.READER_IDLE, idleTime);

}

}

这样每一个session在打开时都会设置handler，而具体的handler由各个派生类创建返回。这里的StanzHandler就是Openfire里的数据包处理单元。和connection类型一样，包处理也是对应的几个类：

注：

Session模块

对于Openfire来说一个比较重要的功能就是管理session，因为要与客户端实时的进行数据通讯，所以必须保持着连接。在Openfire中对于Session的管理都集中在SessionManager模块。但在前面说到连接管理时已经知道了IoSession的创建过程，但并没有看到openfire是如何管理它的。接着ConnectionHandler和StanzaHandler就能知道其中有奥秘。

前面知道了ConnectionHandler是连接的处理者，这里会有连接的创建、关闭、数据收发的处理，回到ConnectionHandler这个抽象类中。对于创建时(sessionOpend)主要是创建了StanzaHandler，这样就把数据包的处理委托给了StzanzHandler(派生类)。但是这个时候并没有将session放入到openfire的session管理模块中，而是在客户端发送数据过来后才开始的。

先看看ConnectionHandler的messageReceived方法：

@Overridepublic void messageReceived(IoSession session, Object message) throwsException {//Get the stanza handler for this session

StanzaHandler handler =(StanzaHandler) session.getAttribute(HANDLER);//Get the parser to use to process stanza. For optimization there is going//to be a parser for each running thread. Each Filter will be executed//by the Executor placed as the first Filter. So we can have a parser associated//to each Thread

final XMPPPacketReader parser =PARSER_CACHE.get();//Update counter of read btyes

updateReadBytesCounter(session);//System.out.println("RCVD: " + message);//Let the stanza handler process the received stanza

try{

handler.process((String) message, parser);

}catch(Exception e) {

Log.error("Closing connection due to error while processing message: " +message, e);final Connection connection =(Connection) session.getAttribute(CONNECTION);if ( connection != null) {

connection.close();

}

}

}

在接收到数据包后获取到StanzaHandler，然后调用了它的process方法，也就是让实际的包处理者去处理数据。这样就回到了StanzeHanler，以ClientStanzaHandler为例子。只不过这个派生类中没有重写process方法，也就是说要看父类的实现：

public void process(String stanza, XMPPPacketReader reader) throwsException {boolean initialStream = stanza.startsWith("

..........//Found an stream:stream tag...

if (!sessionCreated) {

sessionCreated= true;

MXParser parser=reader.getXPPParser();

parser.setInput(newStringReader(stanza));

createSession(parser);

}

..........return;

}

..........

}

由于代码较多，我省略了一些代码。看到这应该明白了吧，对于当前的连接没有创建Openfire的session对象时，会进行创建过程createSession，对于不同的StanzeHandler会有些不一样，这里ClientStanzaHandler的实现就是把创建好的session放到本地的LocalClientSession中：

@OverridebooleancreateSession(String namespace, String serverName, XmlPullParser xpp, Connection connection)throwsXmlPullParserException {if ("jabber:client".equals(namespace)) {//The connected client is a regular client so create a ClientSession

session =LocalClientSession.createSession(serverName, xpp, connection);return true;

}return false;

}

到这一个session算是建立完成了。

集群下的session

之前一篇关于《Openfire集群源码分析》提到了session的一些内容。其中也提到了session是不会向每一台服务器进行同步复制的，这就有一个问题，如果A用户先是连接了服务器1，但是接下来的操作又到服务器2，这不就会造成session无法找到吗？同样的问题，如果想要获取到当前所有的client session怎么办？

1、如何在集群中发消息

对于消息最终还是通过session来发送的，前后代码太多，就直接看一下sessionManager中的getSession方法吧。

publicClientSession getSession(JID from) {//Return null if the JID is null or belongs to a foreign server. If the server is//shutting down then serverName will be null so answer null too in this case.

if (from == null || serverName == null || !serverName.equals(from.getDomain())) {return null;

}//Initially Check preAuthenticated Sessions

if (from.getResource() != null) {

ClientSession session=localSessionManager.getPreAuthenticatedSessions().get(from.getResource());if (session != null) {returnsession;

}

}if (from.getResource() == null || from.getNode() == null) {return null;

}returnroutingTable.getClientRoute(from);

}

先是获取本地的session，如果能找到直接返回，找不到则跳到routingTable里获取客户端的路由信息。

@OverridepublicClientSession getClientRoute(JID jid) {//Check if this session is hosted by this cluster node

ClientSession session =(ClientSession) localRoutingTable.getRoute(jid.toString());if (session == null) {//The session is not in this JVM so assume remote

RemoteSessionLocator locator =server.getRemoteSessionLocator();if (locator != null) {//Check if the session is hosted by other cluster node

ClientRoute route =usersCache.get(jid.toString());if (route == null) {

route=anonymouSUSErsCache.get(jid.toString());

}if (route != null) {

session=locator.getClientSession(route.getNodeID().toByteArray(), jid);

}

}

}returnsession;

}

这里更直接的可以看到，查找本地路由不null则会通过RemoteSessionLocator来完成。当然这里最大的奥秘其实是usersCache和anonymousUsersCache这两个cache。之前写的集群源码分析中提过，最终openfire集群后会对缓存进行同步，这样每台服务器上都会有缓存的副本。所以usersCache是拥有所有��户信息的，有了user的信息就有了jid的信息，这样不管是哪台服务器都可以对数据包处理并发送给客户端。

这里的RemoteSessionLocator是由于适配不同的集群组件所抽象的接口，使得加入不同集群组件提供了透明处理。

2、如何获利所有的在线用户

对于获取所有在线用户这个功能思路也挺简单，一样是找本地所有的缓存。看getSessions的代码：

public CollectiongetSessions() {return routingTable.getClientsRoutes(false);

}

其实就是访问路由表，因为路由表里有所有的cache，和获取单个的session不一样，需要对所有的路由都遍历返回。

@Overridepublic Collection getClientsRoutes(booleanonlyLocal) {//Add sessions hosted by this cluster node

Collection sessions = new ArrayList(localRoutingTable.getClientRoutes());if (!onlyLocal) {//Add sessions not hosted by this JVM

RemoteSessionLocator locator =server.getRemoteSessionLocator();if (locator != null) {//Add sessions of non-anonymous users hosted by other cluster nodes

for (Map.Entryentry : usersCache.entrySet()) {

ClientRoute route=entry.getValue();if (!server.getNodeID().equals(route.getNodeID())) {

sessions.add(locator.getClientSession(route.getNodeID().toByteArray(),newJID(entry.getKey())));

}

}//Add sessions of anonymous users hosted by other cluster nodes

for (Map.Entryentry : anonymousUsersCache.entrySet()) {

ClientRoute route=entry.getValue();if (!server.getNodeID().equals(route.getNodeID())) {

sessions.add(locator.getClientSession(route.getNodeID().toByteArray(),newJID(entry.getKey())));

}

}

}

}returnsessions;

}

总结

对于查看Openfire的源代码学习了一些东西，特别是对于服务化系统的开发思路。而且在集群化上也有了一些认识，知道了多机部署后系统应该要解决哪些问题。

继续学习吧。

Openfire 的详细介绍：请点这里

Openfire 的下载地址：请点这里